Jan Lindemann
6398406a5f
All checks were successful
CI / Packaging - Kali Linux (pull_request) Successful in 4m37s
CI / Packaging - OpenSUSE Tumbleweed (pull_request) Successful in 5m6s
CI / Packaging test (pull_request) Successful in 0s
CI / Packaging - Kali Linux (push) Successful in 4m39s
CI / Packaging - OpenSUSE Tumbleweed (push) Successful in 4m54s
CI / Packaging test (push) Successful in 0s
The unresolved-macro check scans the substituted content for __MACRO__ names, and a commented-out reference fails the compile like a live one. A disabled config section keeps its macros in the comments as documentation, and the matching store keys are retired along with it, so compiling the template fails on lines that no longer take effect. Skip comment lines when scanning for unresolved macros. The integration test gains a target whose template carries a commented-out macro without a store entry. Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.4.0 Signed-off-by: Jan Lindemann <jan@janware.com>
469 lines
18 KiB
Python
469 lines
18 KiB
Python
#!/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'
|
|
PKG2 = 'fixture-pkg2'
|
|
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, names: list[str] | None) -> list[str]:
|
|
if names is None:
|
|
names = list(self.__files)
|
|
ret: list[str] = []
|
|
for n in names:
|
|
ret.extend(self.__files.get(n, []))
|
|
return ret
|
|
|
|
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')
|
|
# -- corge has a conf line in the template and one in the
|
|
# secret: the secret's line takes precedence
|
|
with open(target + '/etc/corge.conf.jw-tmpl', 'w') as f:
|
|
f.write(f'# conf: owner={USER}, group={GROUP}, mode=640\n'
|
|
'key=KEY\n')
|
|
# -- quux carries a commented-out macro without a store entry:
|
|
# the compile must succeed, the comment stays verbatim
|
|
with open(target + '/etc/quux.conf.jw-tmpl', 'w') as f:
|
|
f.write(
|
|
f'# conf: owner={USER}, group={GROUP}, mode=640\n'
|
|
'# retired=__RETIRED__\n'
|
|
'live=LIVE\n'
|
|
)
|
|
# -- qux references a macro that has no store entry: the
|
|
# compile must fail rather than emit a literal macro
|
|
with open(target + '/etc/qux.conf.jw-tmpl', 'w') as f:
|
|
f.write(
|
|
f'# conf: owner={USER}, group={GROUP}, mode=640\n'
|
|
'present=PRESENT\n'
|
|
'missing=__UNRESOLVED__\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'
|
|
qux_tmpl = target + '/etc/qux.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'
|
|
corge_secret = target + '/etc/corge.conf.jw-secret'
|
|
quux_secret = target + '/etc/quux.conf.jw-secret'
|
|
foo_out = target + '/etc/foo.conf'
|
|
bar_out = target + '/etc/bar.conf'
|
|
baz_out = target + '/etc/baz.conf'
|
|
corge_out = target + '/etc/corge.conf'
|
|
quux_out = target + '/etc/quux.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',
|
|
f'default/{trel}/etc/corge.conf.jw-secret': \
|
|
'KEY=corge-value\n',
|
|
f'default/{trel}/etc/quux.conf.jw-secret': \
|
|
'LIVE=live-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/corge.conf.jw-tmpl',
|
|
target + '/etc/quux.conf.jw-tmpl',
|
|
target + '/etc/readme.txt',
|
|
],
|
|
PKG2: [qux_tmpl],
|
|
},
|
|
)
|
|
|
|
# -- 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,
|
|
target + '/etc/corge.conf.jw-tmpl',
|
|
target + '/etc/quux.conf.jw-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,
|
|
corge_secret, quux_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, corge_out, quux_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, corge_secret, quux_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. corge gets a
|
|
# line as well: it takes precedence over the template's
|
|
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')
|
|
os.unlink(corge_secret)
|
|
with open(corge_secret, 'w') as f:
|
|
f.write('# conf: mode=400\n')
|
|
f.write('KEY=corge-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'
|
|
# -- the secret's conf line takes precedence over the
|
|
# template's: corge comes out 0400, not the template's 640
|
|
st = os.stat(corge_out)
|
|
assert (st.st_uid, stat.S_IMODE(st.st_mode)) == (os.getuid(), 0o400), st
|
|
with open(corge_out) as f:
|
|
assert f.read() == 'key=corge-value\n'
|
|
# -- the commented-out macro does not fail the compile and
|
|
# stays in the output verbatim
|
|
with open(quux_out) as f:
|
|
assert f.read() == ('# retired=__RETIRED__\n'
|
|
'live=live-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'
|
|
|
|
# -- compile-templates fails on an unresolved macro: qux
|
|
# resolves PRESENT from its secret file, but the template
|
|
# also references __UNRESOLVED__, which is not in the
|
|
# store. The failure must name the macro, and no compiled
|
|
# file may be written
|
|
qux_secret = target + '/etc/qux.conf.jw-secret'
|
|
qux_out = target + '/etc/qux.conf'
|
|
with open(qux_secret, 'w') as f:
|
|
f.write('PRESENT=there\n')
|
|
try:
|
|
run_cli(
|
|
distro,
|
|
[
|
|
'secrets',
|
|
'compile-templates',
|
|
'--owner',
|
|
USER,
|
|
'--group',
|
|
GROUP,
|
|
'--mode',
|
|
'400',
|
|
PKG2,
|
|
],
|
|
)
|
|
except RuntimeError as e:
|
|
assert 'Unresolved macros' in str(e), e
|
|
assert '__UNRESOLVED__' in str(e), e
|
|
else:
|
|
raise AssertionError('unresolved macro must fail the compile')
|
|
assert not os.path.exists(qux_out)
|
|
|
|
out, _ = run_cli(distro, ['secrets', 'list-compilation-output', PKG])
|
|
assert out == [foo_out, bar_out, baz_out, corge_out, quux_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)
|
|
assert not os.path.exists(corge_out)
|
|
assert not os.path.exists(quux_out)
|
|
|
|
print('All secrets command tests passed')
|
|
finally:
|
|
shutil.rmtree(gpghome, ignore_errors = True)
|
|
if not persistent:
|
|
shutil.rmtree(base)
|
|
|
|
main()
|