diff --git a/test/integration/jw-pkg/secrets/Makefile b/test/integration/jw-pkg/secrets/Makefile new file mode 100644 index 00000000..75c24903 --- /dev/null +++ b/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 diff --git a/test/integration/jw-pkg/secrets/runme.py b/test/integration/jw-pkg/secrets/runme.py new file mode 100644 index 00000000..777498d1 --- /dev/null +++ b/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 / + + 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()