test/integration/jw-pkg/secrets/Makefile

@@ -0,0 +1,19 @@
+TOPDIR                     = ../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+# -- Persistent fixture base directory. Empty by default: the test
+#    runs in a temporary directory. If set, the test builds its
+#    fixture there and leaves it for inspection until `make clean`
+SECRETS_TEST_BASE          ?=
+
+EXE_ARGS                   = $(SECRETS_TEST_BASE)
+
+all:
+test: run
+
+clean: clean.secrets-test
+.PHONY: clean.secrets-test
+clean.secrets-test:
+	@if [ -n "$(SECRETS_TEST_BASE)" ]; then rm -rf "$(SECRETS_TEST_BASE)"; fi

test/integration/jw-pkg/secrets/runme.py

@@ -0,0 +1,367 @@
+#!/usr/bin/env python3
+"""
+Integration test for the jw-pkg secrets command.
+
+Drives the real CLI (App.run) in-process, so the argument parsing, the
+command tree, DistroContext, the secret source, and the template
+compilation all run as production code against a scratch directory
+through the Local exec context. The only stubbed part is the distro's
+package file list, which would otherwise be read from the system rpm
+database and would require a fixture package to be installed on the
+test host.
+
+The secret source is a GPG-encrypted tarball in the canonical layout:
+reverse-domain host prefixes plus a default tree, host wins. Trees
+whose FileContext supports the targpg:// scheme address it through
+that scheme, with the tarball location as deployment information in
+the environment; earlier trees take the tarball URI as the install
+src argument. The targpg:// sections are skipped on the earlier
+trees.
+
+The fake target is a Local exec context whose uri carries a hostname,
+so the host-specific secret store prefix is exercised just like on a
+real ssh target.
+"""
+
+import contextlib
+import grp
+import io
+import os
+import pwd
+import shutil
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+
+from typing import Collection, override
+
+from jw.pkg.App import App
+from jw.pkg.lib.distros.suse.Distro import Distro as SuseDistro
+from jw.pkg.lib.ec.Local import Local
+from jw.pkg.lib.FileContext import FileContext
+from jw.pkg.lib.Package import Package
+
+HOST = 'jannet.example.com'
+HOST_PREFIX = '/'.join(reversed(HOST.split('.')))
+PKG = 'fixture-pkg'
+USER = pwd.getpwuid(os.getuid()).pw_name
+GROUP = grp.getgrgid(os.getgid()).gr_name
+
+def have_targpg() -> bool:
+    """Whether this tree's FileContext supports the targpg:// scheme;
+    trees without it take the tarball URI as the install src argument"""
+    try:
+        FileContext.create('targpg://')
+    except Exception:
+        return False
+    return True
+
+HAVE_TARGPG = have_targpg()
+
+class FakeDistro(SuseDistro):
+    """Distro whose package file list comes from a dict, not rpm"""
+
+    def __init__(self, ec: Local, files: dict[str, list[str]]):
+        super().__init__(ec, id = 'suse')
+        self.__files = files
+
+    @override
+    async def _select_by_name(self, names: Collection[str]) -> list[Package]:
+        return [Package(name) for name in names if name in self.__files]
+
+    @override
+    async def _pkg_files(self, name: str) -> list[str]:
+        return list(self.__files.get(name, []))
+
+def run_cli(distro: FakeDistro, argv: list[str]) -> tuple[list[str], list[str]]:
+    """Run the real jw-pkg CLI in-process, return stdout and stderr
+    lines; a failing command exits, which is folded into a RuntimeError
+    carrying the exit code and the log"""
+    app = App(distro = distro)
+    out_buf = io.StringIO()
+    err_buf = io.StringIO()
+    with contextlib.redirect_stdout(out_buf), contextlib.redirect_stderr(err_buf):
+        try:
+            app.run(argv = argv)
+        except SystemExit as e:
+            err = [line for line in err_buf.getvalue().splitlines() if line]
+            raise RuntimeError(f'command exited {e.code}: {err}') from e
+    out = [line for line in out_buf.getvalue().splitlines() if line]
+    err = [line for line in err_buf.getvalue().splitlines() if line]
+    return out, err
+
+def gen_gpg_key(gpghome: str) -> None:
+    """Generate a throw-away key in a private GPG home"""
+    p = subprocess.run(
+        [
+            'gpg',
+            '--batch',
+            '--pinentry-mode',
+            'loopback',
+            '--passphrase',
+            '',
+            '--quick-gen-key',
+            'test@example.com',
+        ],
+        capture_output = True,
+        env = dict(os.environ, GNUPGHOME = gpghome),
+    )
+    if p.returncode != 0:
+        raise Exception(
+            'gpg key generation failed: ' + p.stderr.decode(errors = 'replace')
+        )
+
+def make_gpg_tar(path: str, entries: dict[str, str], gpghome: str) -> None:
+    """Build a GPG-encrypted tar store in the canonical layout"""
+    buf = io.BytesIO()
+    with tarfile.open(fileobj = buf, mode = 'w') as tf:
+        for name, text in entries.items():
+            info = tarfile.TarInfo(name)
+            data = text.encode()
+            info.size = len(data)
+            info.mode = 0o400
+            tf.addfile(info, io.BytesIO(data))
+    tar_path = path + '.plain'
+    with open(tar_path, 'wb') as f:
+        f.write(buf.getvalue())
+    subprocess.run(
+        [
+            'gpg',
+            '--batch',
+            '--yes',
+            '--quiet',
+            '--no-tty',
+            '--trust-model',
+            'always',
+            '--recipient',
+            'test@example.com',
+            '--encrypt',
+            '--output',
+            path,
+            tar_path,
+        ],
+        check = True,
+        capture_output = True,
+        env = dict(os.environ, GNUPGHOME = gpghome),
+    )
+    os.unlink(tar_path)
+
+def main() -> None:
+    if not shutil.which('gpg'):
+        print('SKIP: gpg not available')
+        return
+
+    # -- Base directory: given on the command line (via the Makefile's
+    #    SECRETS_TEST_BASE), it persists for inspection until `make
+    #    clean`; otherwise the test runs in a temporary directory
+    persistent = len(sys.argv) > 1
+    if persistent:
+        base = os.path.abspath(sys.argv[1])
+        # -- Wipe stale state: installed secrets are mode 0400 and
+        #    would refuse the overwrite of a rerun
+        shutil.rmtree(base, ignore_errors = True)
+    else:
+        base = tempfile.mkdtemp(prefix = 'jw-pkg-secrets-test.')
+    # -- The GPG home lives outside the fixture base: the agent's
+    #    socket path is built from it and would exceed the 108
+    #    character unix limit for a long, persistent base
+    gpghome = tempfile.mkdtemp(prefix = 'jw-pkg-secrets-gpg.')
+    try:
+        # -- The fake target filesystem: a package's template files
+
+        target = os.path.join(base, 'target')
+        os.makedirs(target + '/etc')
+        with open(target + '/etc/foo.conf.jw-tmpl', 'w') as f:
+            f.write(
+                f'# conf: owner={USER}, group={GROUP}, mode=0640\n'
+                'host=HOSTNAME\n'
+                'token=TOKEN\n'
+            )
+        with open(target + '/etc/bar.conf.jw-tmpl', 'w') as f:
+            f.write('plain=PLAIN\n')
+        # -- baz has a template but no store entry: install must not
+        #    extract it, and must not leave an empty file behind
+        with open(target + '/etc/baz.conf.jw-tmpl', 'w') as f:
+            f.write('x=X\n')
+        with open(target + '/etc/readme.txt', 'w') as f:
+            f.write('plain package file\n')
+
+        foo_tmpl = target + '/etc/foo.conf.jw-tmpl'
+        bar_tmpl = target + '/etc/bar.conf.jw-tmpl'
+        baz_tmpl = target + '/etc/baz.conf.jw-tmpl'
+        foo_secret = target + '/etc/foo.conf.jw-secret'
+        bar_secret = target + '/etc/bar.conf.jw-secret'
+        baz_secret = target + '/etc/baz.conf.jw-secret'
+        foo_out = target + '/etc/foo.conf'
+        bar_out = target + '/etc/bar.conf'
+        baz_out = target + '/etc/baz.conf'
+
+        # -- The GPG-encrypted tar store in canonical layout. Member
+        #    names are <prefix>/<target path without leading slash>
+
+        trel = os.path.relpath(target, '/')
+        gen_gpg_key(gpghome)
+        # -- The store decrypts in a local exec context, which
+        #    inherits the throw-away GPG home
+        os.environ['GNUPGHOME'] = gpghome
+        enc = os.path.join(base, 'secrets.tar.gpg')
+        make_gpg_tar(
+            enc,
+            {
+                f'default/{trel}/etc/foo.conf.jw-secret': \
+                    'HOSTNAME=default-host\nTOKEN=default-token\n',
+                f'{HOST_PREFIX}/{trel}/etc/foo.conf.jw-secret': \
+                    'HOSTNAME=jannet\nTOKEN=host-token\n',
+                f'default/{trel}/etc/bar.conf.jw-secret': \
+                    'PLAIN=plain-value\n',
+                # A member that is not a secret of the fixture
+                # package: install must not extract it
+                f'default/{trel}/etc/cert.pem': 'decoy certificate\n',
+            },
+            gpghome,
+        )
+
+        assert not os.path.exists(target + '/etc/cert.pem')
+
+        distro = FakeDistro(
+            Local('local://' + HOST),
+            {PKG: [foo_tmpl, bar_tmpl, baz_tmpl, target + '/etc/readme.txt']},
+        )
+
+        # -- list-templates: only the .jw-tmpl files of the package
+
+        out, _ = run_cli(distro, ['secrets', 'list-templates', PKG])
+        assert out == [foo_tmpl, bar_tmpl, baz_tmpl], out
+
+        # -- list-secrets: nothing installed yet
+
+        out, _ = run_cli(distro, ['secrets', 'list-secrets', PKG])
+        assert out == [], out
+
+        out, _ = run_cli(distro, ['secrets', 'list-secrets', '--all', PKG])
+        assert out == [foo_secret, bar_secret, baz_secret], out
+
+        # -- list-compilation-output: nothing compiled yet
+
+        out, _ = run_cli(distro, ['secrets', 'list-compilation-output', PKG])
+        assert out == [], out
+
+        out, _ = run_cli(distro, ['secrets', 'list-compilation-output', '--all', PKG])
+        assert out == [foo_out, bar_out, baz_out], out
+
+        # -- install: decrypt the tarball, keep the host-specific entry
+        #    over the default one, extract only the needed secrets
+        if HAVE_TARGPG:
+            os.environ['JW_SECRETS_TARGPG_TAR'] = 'local://' + enc
+            out, _ = run_cli(distro, ['secrets', 'install', 'targpg://', PKG])
+        else:
+            out, _ = run_cli(distro, ['secrets', 'install', 'local://' + enc, PKG])
+        assert out == [], out
+        with open(foo_secret) as f:
+            s = f.read()
+            assert s == 'HOSTNAME=jannet\nTOKEN=host-token\n', s
+        with open(bar_secret) as f:
+            assert f.read() == 'PLAIN=plain-value\n'
+        assert not os.path.exists(baz_secret)
+        assert not os.path.exists(target + '/etc/cert.pem')
+        assert stat.S_IMODE(os.stat(foo_secret).st_mode) == 0o400
+
+        out, _ = run_cli(distro, ['secrets', 'list-secrets', PKG])
+        assert out == [foo_secret, bar_secret], out
+
+        if HAVE_TARGPG:
+            # -- a rooted targpg:// URI is rejected: install addresses
+            #    the whole namespace, so the src URI must be the
+            #    scheme root, not a host
+            try:
+                run_cli(distro, ['secrets', 'install', f'targpg://{HOST}', PKG])
+            except RuntimeError as e:
+                assert 'whole store' in str(e), e
+            else:
+                raise AssertionError('rooted targpg:// URI must be rejected')
+
+            # -- install again from the on-host file tree itself: the
+            #    src URI carries a hostname, so install reads the
+            #    entry files by config path from that host's file
+            #    tree. The entry files written by the first install
+            #    are mode 0400; install drops and rewrites them
+            out, _ = run_cli(distro, ['secrets', 'install', f'local://{HOST}', PKG])
+            assert out == [], out
+            with open(foo_secret) as f:
+                s = f.read()
+                assert s == 'HOSTNAME=jannet\nTOKEN=host-token\n', s
+            with open(bar_secret) as f:
+                assert f.read() == 'PLAIN=plain-value\n'
+            assert not os.path.exists(baz_secret)
+            assert stat.S_IMODE(os.stat(foo_secret).st_mode) == 0o400
+
+        # -- compile-templates: substitute the keys, apply the metadata
+
+        out, _ = run_cli(
+            distro, [
+                'secrets',
+                'compile-templates',
+                '--owner',
+                USER,
+                '--group',
+                GROUP,
+                PKG,
+            ]
+        )
+        assert out == [], out
+        st = os.stat(foo_out)
+        assert (st.st_uid, stat.S_IMODE(st.st_mode)) == (os.getuid(), 0o640), st
+        with open(foo_out) as f:
+            assert f.read() == (
+                f'# conf: owner={USER}, group={GROUP}, mode=0640\n'
+                'host=jannet\n'
+                'token=host-token\n'
+            )
+        st = os.stat(bar_out)
+        assert (st.st_uid, stat.S_IMODE(st.st_mode)) == (os.getuid(), 0o400), st
+        with open(bar_out) as f:
+            assert f.read() == 'plain=plain-value\n'
+
+        # -- compile-templates is idempotent
+
+        out, _ = run_cli(
+            distro, [
+                'secrets',
+                'compile-templates',
+                '--owner',
+                USER,
+                '--group',
+                GROUP,
+                PKG,
+            ]
+        )
+        assert out == [], out
+        with open(foo_out) as f:
+            assert f.read() == (
+                f'# conf: owner={USER}, group={GROUP}, mode=0640\n'
+                'host=jannet\n'
+                'token=host-token\n'
+            )
+        with open(bar_out) as f:
+            assert f.read() == 'plain=plain-value\n'
+
+        out, _ = run_cli(distro, ['secrets', 'list-compilation-output', PKG])
+        assert out == [foo_out, bar_out, baz_out], out
+
+        # -- rm-compilation-output: drop the compiled files
+
+        out, _ = run_cli(distro, ['secrets', 'rm-compilation-output', PKG])
+        assert out == [], out
+        assert not os.path.exists(foo_out)
+        assert not os.path.exists(bar_out)
+        assert not os.path.exists(baz_out)
+
+        print('All secrets command tests passed')
+    finally:
+        shutil.rmtree(gpghome, ignore_errors = True)
+        if not persistent:
+            shutil.rmtree(base)
+
+main()