depot/nixos/tests/systemd-confinement/default.nix

import ../make-test-python.nix {
  name = "systemd-confinement";

  nodes.machine =
    { pkgs, lib, ... }:
    let
      testLib = pkgs.python3Packages.buildPythonPackage {
        name = "confinement-testlib";
        unpackPhase = ''
          cat > setup.py <<EOF
          from setuptools import setup
          setup(name='confinement-testlib', py_modules=["checkperms"])
          EOF
          cp ${./checkperms.py} checkperms.py
        '';
      };

      mkTest =
        name: testScript:
        pkgs.writers.writePython3 "${name}.py"
          {
            libraries = [
              pkgs.python3Packages.pytest
              testLib
            ];
          }
          ''
            # This runs our test script by using pytest's assertion rewriting, so
            # that whenever we use "assert <something>", the actual values are
            # printed rather than getting a generic AssertionError or the need to
            # pass an explicit assertion error message.
            import ast
            from pathlib import Path
            from _pytest.assertion.rewrite import rewrite_asserts

            script = Path('${pkgs.writeText "${name}-main.py" ''
              import errno, os, pytest, signal
              from subprocess import run
              from checkperms import Accessibility, assert_permissions

              ${testScript}
            ''}') # noqa
            filename = str(script)
            source = script.read_bytes()

            tree = ast.parse(source, filename=filename)
            rewrite_asserts(tree, source, filename)
            exec(compile(tree, filename, 'exec', dont_inherit=True))
          '';

      mkTestStep =
        num:
        {
          description,
          testScript,
          config ? { },
          serviceName ? "test${toString num}",
          rawUnit ? null,
        }:
        {
          systemd.packages = lib.optional (rawUnit != null) (
            pkgs.writeTextFile {
              name = serviceName;
              destination = "/etc/systemd/system/${serviceName}.service";
              text = rawUnit;
            }
          );

          systemd.services.${serviceName} =
            {
              inherit description;
              requiredBy = [ "multi-user.target" ];
              confinement = (config.confinement or { }) // {
                enable = true;
              };
              serviceConfig = (config.serviceConfig or { }) // {
                ExecStart = mkTest serviceName testScript;
                Type = "oneshot";
              };
            }
            // removeAttrs config [
              "confinement"
              "serviceConfig"
            ];
        };

      parametrisedTests =
        lib.concatMap
          (
            { user, privateTmp }:
            let
              withTmp = if privateTmp then "with PrivateTmp" else "without PrivateTmp";

              serviceConfig =
                if user == "static-user" then
                  {
                    User = "chroot-testuser";
                    Group = "chroot-testgroup";
                  }
                else if user == "dynamic-user" then
                  {
                    DynamicUser = true;
                  }
                else
                  { };

            in
            [
              {
                description = "${user}, chroot-only confinement ${withTmp}";
                config = {
                  confinement.mode = "chroot-only";
                  # Only set if privateTmp is true to ensure that the default is false.
                  serviceConfig =
                    serviceConfig
                    // lib.optionalAttrs privateTmp {
                      PrivateTmp = true;
                    };
                };
                testScript =
                  if user == "root" then
                    ''
                      assert os.getuid() == 0
                      assert os.getgid() == 0

                      assert_permissions({
                        'bin': Accessibility.READABLE,
                        'nix': Accessibility.READABLE,
                        'run': Accessibility.READABLE,
                        ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
                        ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
                        ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
                      })
                    ''
                  else
                    ''
                      assert os.getuid() != 0
                      assert os.getgid() != 0

                      assert_permissions({
                        'bin': Accessibility.READABLE,
                        'nix': Accessibility.READABLE,
                        'run': Accessibility.READABLE,
                        ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
                        ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
                        ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
                      })
                    '';
              }
              {
                description = "${user}, full APIVFS confinement ${withTmp}";
                config = {
                  # Only set if privateTmp is false to ensure that the default is true.
                  serviceConfig =
                    serviceConfig
                    // lib.optionalAttrs (!privateTmp) {
                      PrivateTmp = false;
                    };
                };
                testScript =
                  if user == "root" then
                    ''
                      assert os.getuid() == 0
                      assert os.getgid() == 0

                      assert_permissions({
                        'bin': Accessibility.READABLE,
                        'nix': Accessibility.READABLE,
                        ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
                        'run': Accessibility.WRITABLE,

                        'proc': Accessibility.SPECIAL,
                        'sys': Accessibility.SPECIAL,
                        'dev': Accessibility.WRITABLE,

                        ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
                        ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
                      })
                    ''
                  else
                    ''
                      assert os.getuid() != 0
                      assert os.getgid() != 0

                      assert_permissions({
                        'bin': Accessibility.READABLE,
                        'nix': Accessibility.READABLE,
                        ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
                        'run': Accessibility.STICKY,

                        'proc': Accessibility.SPECIAL,
                        'sys': Accessibility.SPECIAL,
                        'dev': Accessibility.SPECIAL,
                        'dev/shm': Accessibility.STICKY,
                        'dev/mqueue': Accessibility.STICKY,

                        ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
                        ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
                      })
                    '';
              }
            ]
          )
          (
            lib.cartesianProduct {
              user = [
                "root"
                "dynamic-user"
                "static-user"
              ];
              privateTmp = [
                true
                false
              ];
            }
          );

    in
    {
      imports = lib.imap1 mkTestStep (
        parametrisedTests
        ++ [
          {
            description = "existence of bind-mounted /etc";
            config.serviceConfig.BindReadOnlyPaths = [ "/etc" ];
            testScript = ''
              assert Path('/etc/passwd').read_text()
            '';
          }
          (
            let
              symlink = pkgs.runCommand "symlink" {
                target = pkgs.writeText "symlink-target" "got me";
              } "ln -s \"$target\" \"$out\"";
            in
            {
              description = "check if symlinks are properly bind-mounted";
              config.confinement.packages = lib.singleton symlink;
              testScript = ''
                assert Path('${symlink}').read_text() == 'got me'
              '';
            }
          )
          {
            description = "check if StateDirectory works";
            config.serviceConfig.User = "chroot-testuser";
            config.serviceConfig.Group = "chroot-testgroup";
            config.serviceConfig.StateDirectory = "testme";

            # We restart on purpose here since we want to check whether the state
            # directory actually persists.
            config.serviceConfig.Restart = "on-failure";
            config.serviceConfig.RestartMode = "direct";

            testScript = ''
              assert not Path('/tmp/canary').exists()
              Path('/tmp/canary').touch()

              if (foo := Path('/var/lib/testme/foo')).exists():
                assert Path('/var/lib/testme/foo').read_text() == 'works'
              else:
                Path('/var/lib/testme/foo').write_text('works')
                print('<4>Exiting with failure to check persistence on restart.')
                raise SystemExit(1)
            '';
          }
          {
            description = "check if /bin/sh works";
            testScript = ''
              assert Path('/bin/sh').exists()

              result = run(
                ['/bin/sh', '-c', 'echo -n bar'],
                capture_output=True,
                check=True,
              )
              assert result.stdout == b'bar'
            '';
          }
          {
            description = "check if suppressing /bin/sh works";
            config.confinement.binSh = null;
            testScript = ''
              assert not Path('/bin/sh').exists()
              with pytest.raises(FileNotFoundError):
                run(['/bin/sh', '-c', 'echo foo'])
            '';
          }
          {
            description = "check if we can set /bin/sh to something different";
            config.confinement.binSh = "${pkgs.hello}/bin/hello";
            testScript = ''
              assert Path('/bin/sh').exists()
              result = run(
                ['/bin/sh', '-g', 'foo'],
                capture_output=True,
                check=True,
              )
              assert result.stdout == b'foo\n'
            '';
          }
          {
            description = "check if only Exec* dependencies are included";
            config.environment.FOOBAR = pkgs.writeText "foobar" "eek";
            testScript = ''
              with pytest.raises(FileNotFoundError):
                Path(os.environ['FOOBAR']).read_text()
            '';
          }
          {
            description = "check if fullUnit includes all dependencies";
            config.environment.FOOBAR = pkgs.writeText "foobar" "eek";
            config.confinement.fullUnit = true;
            testScript = ''
              assert Path(os.environ['FOOBAR']).read_text() == 'eek'
            '';
          }
          {
            description = "check if shipped unit file still works";
            config.confinement.mode = "chroot-only";
            rawUnit = ''
              [Service]
              SystemCallFilter=~kill
              SystemCallErrorNumber=ELOOP
            '';
            testScript = ''
              with pytest.raises(OSError) as excinfo:
                os.kill(os.getpid(), signal.SIGKILL)
              assert excinfo.value.errno == errno.ELOOP
            '';
          }
        ]
      );

      config.users.groups.chroot-testgroup = { };
      config.users.users.chroot-testuser = {
        isSystemUser = true;
        description = "Chroot Test User";
        group = "chroot-testgroup";
      };
    };

  testScript = ''
    machine.wait_for_unit("multi-user.target")
  '';
}