nixos/tests/systemd-confinement/default.nix

   1 import ../make-test-python.nix {
   2   name = "systemd-confinement";
   3
   4   nodes.machine = { pkgs, lib, ... }: let
   5     testLib = pkgs.python3Packages.buildPythonPackage {
   6       name = "confinement-testlib";
   7       unpackPhase = ''
   8         cat > setup.py <<EOF
   9         from setuptools import setup
  10         setup(name='confinement-testlib', py_modules=["checkperms"])
  11         EOF
  12         cp ${./checkperms.py} checkperms.py
  13       '';
  14     };
  15
  16     mkTest = name: testScript: pkgs.writers.writePython3 "${name}.py" {
  17       libraries = [ pkgs.python3Packages.pytest testLib ];
  18     } ''
  19       # This runs our test script by using pytest's assertion rewriting, so
  20       # that whenever we use "assert <something>", the actual values are
  21       # printed rather than getting a generic AssertionError or the need to
  22       # pass an explicit assertion error message.
  23       import ast
  24       from pathlib import Path
  25       from _pytest.assertion.rewrite import rewrite_asserts
  26
  27       script = Path('${pkgs.writeText "${name}-main.py" ''
  28         import errno, os, pytest, signal
  29         from subprocess import run
  30         from checkperms import Accessibility, assert_permissions
  31
  32         ${testScript}
  33       ''}') # noqa
  34       filename = str(script)
  35       source = script.read_bytes()
  36
  37       tree = ast.parse(source, filename=filename)
  38       rewrite_asserts(tree, source, filename)
  39       exec(compile(tree, filename, 'exec', dont_inherit=True))
  40     '';
  41
  42     mkTestStep = num: {
  43       description,
  44       testScript,
  45       config ? {},
  46       serviceName ? "test${toString num}",
  47       rawUnit ? null,
  48     }: {
  49       systemd.packages = lib.optional (rawUnit != null) (pkgs.writeTextFile {
  50         name = serviceName;
  51         destination = "/etc/systemd/system/${serviceName}.service";
  52         text = rawUnit;
  53       });
  54
  55       systemd.services.${serviceName} = {
  56         inherit description;
  57         requiredBy = [ "multi-user.target" ];
  58         confinement = (config.confinement or {}) // { enable = true; };
  59         serviceConfig = (config.serviceConfig or {}) // {
  60           ExecStart = mkTest serviceName testScript;
  61           Type = "oneshot";
  62         };
  63       } // removeAttrs config [ "confinement" "serviceConfig" ];
  64     };
  65
  66     parametrisedTests = lib.concatMap ({ user, privateTmp }: let
  67       withTmp = if privateTmp then "with PrivateTmp" else "without PrivateTmp";
  68
  69       serviceConfig = if user == "static-user" then {
  70         User = "chroot-testuser";
  71         Group = "chroot-testgroup";
  72       } else if user == "dynamic-user" then {
  73         DynamicUser = true;
  74       } else {};
  75
  76     in [
  77       { description = "${user}, chroot-only confinement ${withTmp}";
  78         config = {
  79           confinement.mode = "chroot-only";
  80           # Only set if privateTmp is true to ensure that the default is false.
  81           serviceConfig = serviceConfig // lib.optionalAttrs privateTmp {
  82             PrivateTmp = true;
  83           };
  84         };
  85         testScript = if user == "root" then ''
  86           assert os.getuid() == 0
  87           assert os.getgid() == 0
  88
  89           assert_permissions({
  90             'bin': Accessibility.READABLE,
  91             'nix': Accessibility.READABLE,
  92             'run': Accessibility.READABLE,
  93             ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
  94             ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
  95             ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
  96           })
  97         '' else ''
  98           assert os.getuid() != 0
  99           assert os.getgid() != 0
 100
 101           assert_permissions({
 102             'bin': Accessibility.READABLE,
 103             'nix': Accessibility.READABLE,
 104             'run': Accessibility.READABLE,
 105             ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
 106             ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
 107             ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
 108           })
 109         '';
 110       }
 111       { description = "${user}, full APIVFS confinement ${withTmp}";
 112         config = {
 113           # Only set if privateTmp is false to ensure that the default is true.
 114           serviceConfig = serviceConfig // lib.optionalAttrs (!privateTmp) {
 115             PrivateTmp = false;
 116           };
 117         };
 118         testScript = if user == "root" then ''
 119           assert os.getuid() == 0
 120           assert os.getgid() == 0
 121
 122           assert_permissions({
 123             'bin': Accessibility.READABLE,
 124             'nix': Accessibility.READABLE,
 125             ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
 126             'run': Accessibility.WRITABLE,
 127
 128             'proc': Accessibility.SPECIAL,
 129             'sys': Accessibility.SPECIAL,
 130             'dev': Accessibility.WRITABLE,
 131
 132             ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
 133             ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
 134           })
 135         '' else ''
 136           assert os.getuid() != 0
 137           assert os.getgid() != 0
 138
 139           assert_permissions({
 140             'bin': Accessibility.READABLE,
 141             'nix': Accessibility.READABLE,
 142             ${lib.optionalString privateTmp "'tmp': Accessibility.STICKY,"}
 143             'run': Accessibility.STICKY,
 144
 145             'proc': Accessibility.SPECIAL,
 146             'sys': Accessibility.SPECIAL,
 147             'dev': Accessibility.SPECIAL,
 148             'dev/shm': Accessibility.STICKY,
 149             'dev/mqueue': Accessibility.STICKY,
 150
 151             ${lib.optionalString privateTmp "'var': Accessibility.READABLE,"}
 152             ${lib.optionalString privateTmp "'var/tmp': Accessibility.STICKY,"}
 153           })
 154         '';
 155       }
 156     ]) (lib.cartesianProduct {
 157       user = [ "root" "dynamic-user" "static-user" ];
 158       privateTmp = [ true false ];
 159     });
 160
 161   in {
 162     imports = lib.imap1 mkTestStep (parametrisedTests ++ [
 163       { description = "existence of bind-mounted /etc";
 164         config.serviceConfig.BindReadOnlyPaths = [ "/etc" ];
 165         testScript = ''
 166           assert Path('/etc/passwd').read_text()
 167         '';
 168       }
 169       (let
 170         symlink = pkgs.runCommand "symlink" {
 171           target = pkgs.writeText "symlink-target" "got me";
 172         } "ln -s \"$target\" \"$out\"";
 173       in {
 174         description = "check if symlinks are properly bind-mounted";
 175         config.confinement.packages = lib.singleton symlink;
 176         testScript = ''
 177           assert Path('${symlink}').read_text() == 'got me'
 178         '';
 179       })
 180       { description = "check if StateDirectory works";
 181         config.serviceConfig.User = "chroot-testuser";
 182         config.serviceConfig.Group = "chroot-testgroup";
 183         config.serviceConfig.StateDirectory = "testme";
 184
 185         # We restart on purpose here since we want to check whether the state
 186         # directory actually persists.
 187         config.serviceConfig.Restart = "on-failure";
 188         config.serviceConfig.RestartMode = "direct";
 189
 190         testScript = ''
 191           assert not Path('/tmp/canary').exists()
 192           Path('/tmp/canary').touch()
 193
 194           if (foo := Path('/var/lib/testme/foo')).exists():
 195             assert Path('/var/lib/testme/foo').read_text() == 'works'
 196           else:
 197             Path('/var/lib/testme/foo').write_text('works')
 198             print('<4>Exiting with failure to check persistence on restart.')
 199             raise SystemExit(1)
 200         '';
 201       }
 202       { description = "check if /bin/sh works";
 203         testScript = ''
 204           assert Path('/bin/sh').exists()
 205
 206           result = run(
 207             ['/bin/sh', '-c', 'echo -n bar'],
 208             capture_output=True,
 209             check=True,
 210           )
 211           assert result.stdout == b'bar'
 212         '';
 213       }
 214       { description = "check if suppressing /bin/sh works";
 215         config.confinement.binSh = null;
 216         testScript = ''
 217           assert not Path('/bin/sh').exists()
 218           with pytest.raises(FileNotFoundError):
 219             run(['/bin/sh', '-c', 'echo foo'])
 220         '';
 221       }
 222       { description = "check if we can set /bin/sh to something different";
 223         config.confinement.binSh = "${pkgs.hello}/bin/hello";
 224         testScript = ''
 225           assert Path('/bin/sh').exists()
 226           result = run(
 227             ['/bin/sh', '-g', 'foo'],
 228             capture_output=True,
 229             check=True,
 230           )
 231           assert result.stdout == b'foo\n'
 232         '';
 233       }
 234       { description = "check if only Exec* dependencies are included";
 235         config.environment.FOOBAR = pkgs.writeText "foobar" "eek";
 236         testScript = ''
 237           with pytest.raises(FileNotFoundError):
 238             Path(os.environ['FOOBAR']).read_text()
 239         '';
 240       }
 241       { description = "check if fullUnit includes all dependencies";
 242         config.environment.FOOBAR = pkgs.writeText "foobar" "eek";
 243         config.confinement.fullUnit = true;
 244         testScript = ''
 245           assert Path(os.environ['FOOBAR']).read_text() == 'eek'
 246         '';
 247       }
 248       { description = "check if shipped unit file still works";
 249         config.confinement.mode = "chroot-only";
 250         rawUnit = ''
 251           [Service]
 252           SystemCallFilter=~kill
 253           SystemCallErrorNumber=ELOOP
 254         '';
 255         testScript = ''
 256           with pytest.raises(OSError) as excinfo:
 257             os.kill(os.getpid(), signal.SIGKILL)
 258           assert excinfo.value.errno == errno.ELOOP
 259         '';
 260       }
 261     ]);
 262
 263     config.users.groups.chroot-testgroup = {};
 264     config.users.users.chroot-testuser = {
 265       isSystemUser = true;
 266       description = "Chroot Test User";
 267       group = "chroot-testgroup";
 268     };
 269   };
 270
 271   testScript = ''
 272     machine.wait_for_unit("multi-user.target")
 273   '';
 274 }