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

#!/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=640\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

        # -- The file attributes can also be specified atop the
        #    .jw-secret; bar takes its mode from there
        os.unlink(bar_secret)
        with open(bar_secret, 'w') as f:
            f.write(f'# conf: owner={USER}, group={GROUP}, mode=640\n')
            f.write('PLAIN=plain-value\n')

        # -- compile-templates: substitute the keys, apply the
        #    metadata; the conf lines win over the --mode default
        out, _ = run_cli(
            distro, [
                'secrets',
                'compile-templates',
                '--owner',
                USER,
                '--group',
                GROUP,
                '--mode',
                '400',
                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() == ('host=jannet\n'
                                'token=host-token\n')
        st = os.stat(bar_out)
        assert (st.st_uid, stat.S_IMODE(st.st_mode)) == (os.getuid(), 0o640), 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,
                '--mode',
                '400',
                PKG,
            ]
        )
        assert out == [], out
        with open(foo_out) as f:
            assert f.read() == ('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()