test/integration/jw-pkg/secrets: Add test

Drive the real "jw-pkg secrets" CLI in-process against a GPG-encrypted
tarball in the canonical layout: reverse-domain host prefixes plus a
default tree, host wins. The distro is stubbed only in its package file
list; argument parsing, DistroContext, install, template compilation and
cleanup run as production code.

The fixture package carries three templates: one secret present in both
host and default trees (the host entry must win), one present only in the
default tree, and one missing from the tarball entirely (install must not
extract it). A tar member that is not a secret of the package must not be
extracted either.

Trees whose FileContext supports the targpg:// scheme address the tar
source through that scheme, with the tarball location in the environment,
and additionally exercise the rooted-URI rejection and the install from
the on-host file tree; earlier trees install from the tarball URI
directly.

The Makefile's SECRETS_TEST_BASE variable pins the fixture to a persistent
directory that survives the test run for inspection.

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.0.0

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()