tegwick
8f808577d6
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
188 lines
9.4 KiB
Python
188 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Private SOPS writes for the independently registered single-age profile.
|
|
|
|
Edit consumes an explicitly prepared private plaintext file. No editor or shell
|
|
is launched, and subprocess plaintext output is discarded, never forwarded.
|
|
"""
|
|
import contextlib
|
|
import fcntl
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
|
|
spec = importlib.util.spec_from_file_location('sops_private', Path(__file__).with_name('sops-private.py'))
|
|
base = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(base)
|
|
Refused = base.Refused
|
|
LIMIT = 8 * 1024 * 1024
|
|
POLICY = ('unencrypted_suffix', 'encrypted_suffix', 'unencrypted_regex', 'encrypted_regex',
|
|
'unencrypted_comment_regex', 'encrypted_comment_regex')
|
|
|
|
|
|
def in_git(parent):
|
|
return any((p / '.git').is_file() or (p / '.git/HEAD').is_file()
|
|
or ((p / 'HEAD').is_file() and (p / 'objects').is_dir() and (p / 'refs').is_dir())
|
|
for p in (parent, *parent.parents))
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def snapshot(name, *, plaintext=False, replacement=False, lock=False):
|
|
if not name: raise Refused('explicit FILE or INPUT is required')
|
|
path = Path(os.path.abspath(name))
|
|
if any(p.is_symlink() for p in (path, *path.parents)):
|
|
raise Refused('file paths must not contain symlinks')
|
|
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
fd = None
|
|
try:
|
|
parent = os.fstat(directory)
|
|
if parent.st_uid != os.getuid() or parent.st_mode & (0o077 if replacement else 0o022):
|
|
raise Refused('file directory ownership/mode refused')
|
|
if replacement and in_git(path.parent):
|
|
raise Refused('private replacement input must be outside Git')
|
|
fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
|
|
info = os.fstat(fd)
|
|
if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1
|
|
or info.st_mode & (0o077 if plaintext else 0o022)):
|
|
raise Refused('file ownership/type/mode refused')
|
|
if lock: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
with os.fdopen(os.dup(fd), 'rb') as stream:
|
|
raw = stream.read(LIMIT + 1)
|
|
if len(raw) > LIMIT: raise Refused('input exceeds the supported size')
|
|
yield path, directory, info, raw
|
|
finally:
|
|
if fd is not None: os.close(fd)
|
|
os.close(directory)
|
|
|
|
|
|
def plain_document(raw, suffix):
|
|
data, kind = base.parse_document(raw, suffix)
|
|
if not isinstance(data, dict) or not data or 'sops' in data:
|
|
raise Refused('plaintext must be a nonempty JSON/YAML mapping without SOPS metadata')
|
|
return kind
|
|
|
|
|
|
def encrypted_document(raw, path, recipient):
|
|
_, kind = base.age_document(path, recipient, raw=raw)
|
|
data, _ = base.parse_document(raw, path.suffix)
|
|
metadata = data['sops']
|
|
# Document recipient metadata is not an independent admission of another
|
|
# recipient. Never rewrap a data key to an injected or unreviewed identity.
|
|
if len(metadata['age']) != 1 or metadata['age'][0]['recipient'] != recipient:
|
|
raise Refused('write requires exactly the independently registered recipient')
|
|
return kind, metadata
|
|
|
|
|
|
def sops(binary, arguments, raw, env, isolated, keyfd=None, *, discard=False):
|
|
result = subprocess.run([binary, *arguments, '/dev/stdin'], input=raw,
|
|
stdout=subprocess.DEVNULL if discard else subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, cwd=isolated, env=env,
|
|
pass_fds=() if keyfd is None else (keyfd,), timeout=30)
|
|
if result.returncode: raise Refused('SOPS operation failed; no replacement published')
|
|
if not discard and (not result.stdout or len(result.stdout) > LIMIT):
|
|
raise Refused('SOPS output was empty or oversized')
|
|
return None if discard else result.stdout
|
|
|
|
|
|
def publish(path, directory, original, encrypted):
|
|
temporary = '.netkingdom-sops-write-' + uuid.uuid4().hex
|
|
try:
|
|
fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
|
|
with os.fdopen(fd, 'wb') as file:
|
|
file.write(encrypted); file.flush(); os.fsync(file.fileno())
|
|
current = os.stat(path.name, dir_fd=directory, follow_symlinks=False)
|
|
fields = ('st_dev', 'st_ino', 'st_size', 'st_mtime_ns', 'st_ctime_ns')
|
|
if any(getattr(current, key) != getattr(original, key) for key in fields):
|
|
raise Refused('source changed during operation; replacement refused')
|
|
os.replace(temporary, path.name, src_dir_fd=directory, dst_dir_fd=directory)
|
|
os.fsync(directory)
|
|
finally:
|
|
try: os.unlink(temporary, dir_fd=directory)
|
|
except FileNotFoundError: pass
|
|
|
|
|
|
def write(action, recipient):
|
|
binary = shutil.which('sops')
|
|
if binary is None: raise Refused('sops executable is unavailable')
|
|
with snapshot(os.environ.get('NK_SOPS_FILE', ''), plaintext=action == 'encrypt', lock=True) as target:
|
|
path, directory, original, raw = target
|
|
if action == 'encrypt':
|
|
kind = plain_document(raw, path.suffix)
|
|
if in_git(path.parent):
|
|
tracked = subprocess.run(['git', '-C', str(path.parent), 'ls-files', '--error-unmatch', '--', path.name],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
|
|
if tracked.returncode != 1:
|
|
raise Refused('tracked plaintext or unverifiable Git index requires incident review')
|
|
metadata = {}
|
|
else:
|
|
kind, metadata = encrypted_document(raw, path, recipient)
|
|
keyfd = None
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix='netkingdom-sops-write-home-') as isolated:
|
|
env = {'PATH': os.environ.get('PATH', os.defpath), 'HOME': isolated, 'XDG_CONFIG_HOME': isolated,
|
|
'GNUPGHOME': isolated, 'AWS_EC2_METADATA_DISABLED': 'true'}
|
|
if action != 'encrypt':
|
|
keyfd = base.private_key_fd()
|
|
derived = subprocess.run(['age-keygen', '-y', f'/proc/self/fd/{keyfd}'],
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
|
pass_fds=(keyfd,), timeout=10)
|
|
if derived.returncode or derived.stdout.strip() != recipient.encode():
|
|
raise Refused('custody key does not match registered recipient')
|
|
env['SOPS_AGE_KEY_FILE'] = f'/proc/self/fd/{keyfd}'
|
|
formats = ['--input-type', kind, '--output-type', kind]
|
|
if action == 'rotate':
|
|
encrypted = sops(binary, ['--rotate', *formats], raw, env, isolated, keyfd)
|
|
else:
|
|
options = []
|
|
if action == 'edit':
|
|
# MAC verification must precede replacement of even a
|
|
# syntactically valid encrypted document.
|
|
sops(binary, ['--decrypt', *formats], raw, env, isolated, keyfd, discard=True)
|
|
with snapshot(os.environ.get('NK_SOPS_INPUT', ''), plaintext=True, replacement=True) as replacement:
|
|
plain = replacement[3]
|
|
if plain_document(plain, replacement[0].suffix) != kind:
|
|
raise Refused('replacement format must match the encrypted document')
|
|
for name in POLICY:
|
|
value = metadata.get(name)
|
|
if value:
|
|
if not isinstance(value, str): raise Refused('invalid encryption policy')
|
|
options.extend(['--' + name.replace('_', '-'), value])
|
|
if metadata.get('mac_only_encrypted'):
|
|
options.append('--mac-only-encrypted')
|
|
else:
|
|
plain = raw
|
|
encrypted = sops(binary, ['--encrypt', '--age', recipient, *formats, *options],
|
|
plain, env, isolated, keyfd)
|
|
encrypted_document(encrypted, path, recipient)
|
|
if keyfd is not None:
|
|
sops(binary, ['--decrypt', *formats], encrypted, env, isolated, keyfd, discard=True)
|
|
if base.public_recipient() != recipient:
|
|
raise Refused('registered recipient changed during operation')
|
|
publish(path, directory, original, encrypted)
|
|
finally:
|
|
if keyfd is not None: os.close(keyfd)
|
|
print('Encrypted file updated; no plaintext was printed. Retain approved independent recovery custody.')
|
|
|
|
|
|
def main():
|
|
for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT): signal.signal(sig, base.interrupted)
|
|
try:
|
|
if len(sys.argv) != 2 or sys.argv[1] not in {'encrypt', 'rotate', 'edit'}:
|
|
raise Refused('unsupported SOPS write operation')
|
|
write(sys.argv[1], base.public_recipient())
|
|
return 0
|
|
except Refused as exc:
|
|
print('ERROR: ' + str(exc), file=sys.stderr)
|
|
except (OSError, ValueError, TypeError, KeyError, ImportError, subprocess.SubprocessError):
|
|
print('ERROR: SOPS write refused or failed; input and tool diagnostics suppressed.', file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|