Jan Lindemann
f08f658d33
All checks were successful
CI / Packaging - Kali Linux (pull_request) Successful in 4m48s
CI / Packaging - OpenSUSE Tumbleweed (pull_request) Successful in 5m12s
CI / Packaging test (pull_request) Successful in 0s
CI / Packaging - Kali Linux (push) Successful in 4m35s
CI / Packaging - OpenSUSE Tumbleweed (push) Successful in 4m49s
CI / Packaging test (push) Successful in 0s
"jw-pkg secrets install" copies secret material from a GPG-encrypted tar
archive to dotenv .jw-secret files on a target host. This commit refactors
the underlying code and adds hooks for the inverse copy direction, and for
more secret sources and targets.
The peers between which secrets are copied are represented by the new
abstract secret_store.Base class, and the existing tar blob and host
storages, inheriting from that class, are TarGpg and KvFile. Moreover,
secret_store.Base is inheriting from FileContext, so all of jw-pkg's
file-like operations are available as hooks.
Secrets inside instances of two file contexts are addressable as
(targpg|kvfile)://target.host.com/path/to/file.conf.jw-secret. The on-host
KvFile will look either locally or via SSH for that file, and TarGpg will
look inside a tar archive specified in the JW_SECRETS_TARGPG_TAR
environment variable.
Uris with these schemas can be passed to FileContext.create(), which will
return a chrooted file context of the correct type.
KvFile.stat("/path/to/file.conf.jw-secret").ftype is FT_SECRET which
inherits from FT_COMPOUND - which means that readdir() works on such
animals. Not entirely sure if that's a good idea, but it feels pretty
generic. Let's see where that takes us.
DistroContext.install() after this refactor reads its source through the
store interface and writes the landing through the KvFile store.
Both secret stores are exercised through the existing unit test of jw-pkg
secrets install, and unit tests for both stores are also added by this
commit.
Signed-off-by: Jan Lindemann <jan@janware.com>
420 lines
15 KiB
Python
420 lines
15 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')
|
|
# -- 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'
|
|
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'],
|
|
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], 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'
|
|
|
|
# -- 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], 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()
|