#!/usr/bin/env python3 """External deployment-only finite FIPS/WG owner. Not an application command broker. Inputs are inherited root/bundle directory FDs, never application paths. Exact source bytes and executable digests are appointed by the deployment principal. Existing assets must match exactly: migration/rotation is deliberately separate. """ import base64 import hashlib import ipaddress import json import os import stat import subprocess import selectors import time import yaml class Refused(Exception): pass def require(ok, reason): if not ok: raise Refused(reason) # Finite destination set. No unit/script/path field in runtime requests. ASSETS = { 'fips.yaml': ('etc/fips/fips.yaml', 0o644), 'fips.key': ('etc/fips/fips.key', 0o600), 'fips.pub': ('etc/fips/fips.pub', 0o644), 'fips.nft': ('etc/fips/fips.nft', 0o644), '80-web-ui.nft': ('etc/fips/fips.d/80-web-ui.nft', 0o644), '85-app-ports.nft': ('etc/fips/fips.d/85-app-ports.nft', 0o644), '90-ssh.nft': ('etc/fips/fips.d/90-ssh.nft', 0o644), 'wg.public': ('etc/archipelago-fips-wg/wg.public', 0o644), } PROGRAMS = {'wg': 'usr/bin/wg', 'systemctl': 'usr/bin/systemctl'} def protected(fd, directory=False, secret=False): s = os.fstat(fd) require(s.st_uid == os.geteuid() and s.st_mode & 0o022 == 0, 'unprotected owner object') require(stat.S_ISDIR(s.st_mode) if directory else stat.S_ISREG(s.st_mode) and s.st_nlink == 1, 'aliased owner object') if secret: require(s.st_mode & 0o077 == 0, 'exposed private key') def open_beneath(root, path, directory=False): require(all(p not in ('', '.', '..') for p in path.split('/')), 'invalid internal path') fd = os.dup(root) try: protected(fd, True) parts = path.split('/') for i, part in enumerate(parts): is_dir = i < len(parts) - 1 or directory new = os.open(part, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | (os.O_DIRECTORY if is_dir else 0), dir_fd=fd) os.close(fd) fd = new protected(fd, is_dir) return fd except BaseException: os.close(fd) raise def read(root, path, secret=False, limit=65536): fd = open_beneath(root, path) try: protected(fd, secret=secret) data = os.read(fd, limit + 1) require(len(data) <= limit, 'owner input exceeds bound') return data finally: os.close(fd) def digest(data): return hashlib.sha256(data).hexdigest() def decode_bech32(text, hrp): require(text.startswith(hrp + '1'), 'invalid FIPS key prefix') alphabet = 'qpzry9x8gf2tvdw0s3jn54khce6mua7l' values = [alphabet.index(c) for c in text[len(hrp) + 1:]] chk = 1 for v in [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] + values: top = chk >> 25 chk = ((chk & 0x1ffffff) << 5) ^ v for i, g in enumerate([0x3b6a57b2, 0x26508e6d, 0x1ea119fa, 0x3d4233dd, 0x2a1462b3]): if (top >> i) & 1: chk ^= g require(chk == 1 and len(values) == 58, 'invalid FIPS key encoding') bits = ''.join(f'{v:05b}' for v in values[:-6]) require(bits[256:] == '0000', 'invalid FIPS key padding') return int(bits[:256], 2).to_bytes(32, 'big') def public_bytes(data): return data if len(data) == 32 else decode_bech32(data.decode('ascii').strip(), 'npub') def public_from_secret(data): # FIPS uses secp256k1 x-only public keys. Decode nsec without a subprocess, # argv, temp file or secret-bearing diagnostics. No key generation occurs. text = data.decode('ascii').strip() if text.startswith('nsec1'): scalar = int.from_bytes(decode_bech32(text, 'nsec'), 'big') else: require(len(text) == 64, 'invalid FIPS key encoding') scalar = int(text, 16) p = 2**256 - 2**32 - 977 n = 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141 require(0 < scalar < n, 'invalid FIPS scalar') g = (0x79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798, 0x483ada7726a3c4655da4fbfc0e1108a8fd17b448a68554199c47d08ffb10d4b8) def add(a, b): if a is None: return b if b is None: return a x, y = a u, v = b if x == u and (y + v) % p == 0: return None slope = ((3*x*x) * pow(2*y, -1, p) if a == b else (v-y) * pow(u-x, -1, p)) % p z = (slope*slope-x-u) % p return z, (slope*(x-z)-y) % p point = None for bit in bin(scalar)[2:]: point = add(point, point) if bit == '1': point = add(point, g) return point[0].to_bytes(32, 'big') def peer(value): require(set(value) == {'public_key', 'ip'}, 'unknown peer fields') key = base64.b64decode(value['public_key'], validate=True) require(len(key) == 32 and key != bytes(32) and base64.b64encode(key).decode() == value['public_key'], 'invalid WG key') ip = ipaddress.IPv4Interface(value['ip']) require(str(ip) == value['ip'] and ip.network.prefixlen == 32 and ip.ip in ipaddress.IPv4Network('10.44.0.0/24') and 2 <= int(ip.ip) % 256 <= 254, 'invalid managed peer address') return value['public_key'], value['ip'] class Owner: def __init__(self, root, bundle, network=None): self.root, self.bundle = root, bundle self.network = network self.plan = json.loads(read(bundle, 'plan.json')) require(set(self.plan) == {'version', 'assets', 'programs', 'fips', 'wg', 'peers'}, 'unknown plan fields') require(self.plan['version'] == 1, 'unknown owner version') require(set(self.plan['assets']) == set(ASSETS), 'exact asset inventory required') require(set(self.plan['programs']) == set(PROGRAMS), 'exact executable pins required') for name in ('fips', 'wg'): require(self.plan[name] in ('running', 'stopped', 'uninstalled'), 'missing operator disposition') require(len(self.plan['peers']) <= 253, 'peer bound exceeded') self.peers = [peer(v) for v in self.plan['peers']] require(len({k for k, _ in self.peers}) == len(self.peers) and len({a for _, a in self.peers}) == len(self.peers), 'peer collision') self.sources = {} for name in ASSETS: data = read(bundle, name, name == 'fips.key') require(digest(data) == self.plan['assets'][name], 'source digest mismatch') self.sources[name] = data require(public_from_secret(self.sources['fips.key']) == public_bytes(self.sources['fips.pub']), 'FIPS identity mismatch') config = yaml.safe_load(self.sources['fips.yaml']) require(isinstance(config, dict) and config.get('node', {}).get('identity', {}).get('persistent') is True, 'FIPS must retain the installed persistent identity') key = base64.b64decode(self.sources['wg.public'].strip(), validate=True) require(len(key) == 32 and key != bytes(32), 'invalid WG server identity') def check_assets(self, allow_absent=False): missing = [] for name, (path, mode) in ASSETS.items(): try: data = read(self.root, path, name == 'fips.key') except FileNotFoundError: require(allow_absent, 'protected prerequisite absent') # Parent provisioning is external; no recursive mkdir broker. parent = open_beneath(self.root, path.rsplit('/', 1)[0], True) os.close(parent) missing.append(name) continue require(data == self.sources[name], 'existing asset differs; preserve and migrate explicitly') return missing def run(self, program, args): require(self.network is not None, 'deployment network capability required') current = os.stat('/proc/self/ns/net') appointed = os.fstat(self.network) require((current.st_dev, current.st_ino) == (appointed.st_dev, appointed.st_ino), 'wrong owner network namespace') fd = open_beneath(self.root, PROGRAMS[program]) try: # Hash the retained executable, then execute that same descriptor. h = hashlib.sha256() while chunk := os.read(fd, 65536): h.update(chunk) require(h.hexdigest() == self.plan['programs'][program], 'executable digest mismatch') os.lseek(fd, 0, 0) require(os.read(fd, 4) == b'\x7fELF', 'owner requires ELF tool') with subprocess.Popen([f'/proc/self/fd/{fd}', *args], pass_fds=(fd,), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, cwd='/', env={'PATH': '/usr/bin:/usr/sbin', 'LC_ALL': 'C'}) as child: output = bytearray() os.set_blocking(child.stdout.fileno(), False) try: with selectors.DefaultSelector() as selector: selector.register(child.stdout, selectors.EVENT_READ) end = time.monotonic() + 1 while True: require(time.monotonic() < end, 'owner deadline exceeded') if selector.select(min(0.02, end - time.monotonic())): chunk = os.read(child.stdout.fileno(), 4096) if not chunk: break output.extend(chunk) require(len(output) <= 65536, 'owner output exceeds bound') child.wait(timeout=max(0.001, end - time.monotonic())) except BaseException: child.kill(); child.wait() raise Refused('owner effect uncertain; retained intent blocks replay') from None require(child.returncode == 0 and len(output) <= 65536, 'owner operation failed or uncertain') return output finally: os.close(fd) def kernel_peers(self): require(self.plan['wg'] == 'running', 'WG stopped/uninstalled decision') require(self.run('wg', ['show', 'wg0', 'public-key']).strip() == self.sources['wg.public'].strip(), 'live WG identity mismatch') entries = {} for line in self.run('wg', ['show', 'wg0', 'allowed-ips']).decode('ascii').splitlines(): key, addresses = line.split('\t', 1) require(key not in entries, 'duplicate live peer') entries[key] = addresses.split() return entries def verify(self, request): self.check_assets() if request == 'FipsActive': request = {'FipsActive': None} require(len(request) == 1, 'invalid finite request') if 'FipsIdentity' in request: value = request['FipsIdentity'] require(set(value) == {'public'} and bytes(value['public']) == public_bytes(self.sources['fips.pub']), 'requested identity mismatch') elif 'FipsActive' in request: require(request['FipsActive'] is None and self.plan['fips'] == 'running', 'FIPS stopped/uninstalled decision') require(self.run('systemctl', ['is-active', 'archipelago-fips.service']).strip() == b'active', 'FIPS not active') elif 'WgPeer' in request: value = request['WgPeer'] require(set(value) == {'public', 'address'}, 'unknown peer request fields') key, address = peer({'public_key': base64.b64encode(bytes(value['public'])).decode(), 'ip': value['address'] + '/32'}) require((key, address) in self.peers, 'peer lacks external admission') require(self.kernel_peers().get(key) == [address], 'peer not restored exactly') else: raise Refused('unknown finite operation') return b'ok' def provision(self, activate=False): require(not activate or self.plan['fips'] == 'running', 'FIPS stopped/uninstalled decision') if activate: require(self.sources['fips.key'].startswith(b'nsec1') and self.sources['fips.key'].endswith(b'\n') and len(self.sources['fips.key']) == 64 and self.sources['fips.pub'].startswith(b'npub1') and self.sources['fips.pub'].endswith(b'\n') and len(self.sources['fips.pub']) == 64, 'activation requires daemon-canonical identity files; explicit representation migration required') missing = self.check_assets(True) plan_bytes = read(self.bundle, 'plan.json') try: require(read(self.root, 'etc/archipelago-fips-wg/plan.json') == plan_bytes, 'existing deployment plan differs; explicit transition required') except FileNotFoundError: pass # All preimages validated before any writes. Uninstalled decisions forbid # even installing missing configuration. No existing inode is replaced. require(not missing or (self.plan['fips'] != 'uninstalled' and self.plan['wg'] != 'uninstalled'), 'uninstall decision forbids provisioning') journal = open_beneath(self.root, 'etc/archipelago-fips-wg', True) try: intent = os.open('intent', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=journal) with os.fdopen(intent, 'wb') as out: out.write(b'uncertain\n'); out.flush(); os.fsync(out.fileno()) os.fsync(journal) for name in missing: path, mode = ASSETS[name] parent, leaf = path.rsplit('/', 1) directory = open_beneath(self.root, parent, True) try: fd = os.open(leaf, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, mode, dir_fd=directory) with os.fdopen(fd, 'wb') as out: out.write(self.sources[name]); out.flush(); os.fsync(out.fileno()) os.fsync(directory) finally: os.close(directory) self.check_assets() if activate: require(self.plan['fips'] == 'running', 'FIPS stopped/uninstalled decision') # No enable/unmask/reset-failed/kill. Only explicitly authorized # deployment activation; never called by normal app startup. self.run('systemctl', ['start', 'archipelago-fips.service']) self.verify({'FipsActive': None}) if self.plan['wg'] == 'running': existing = self.kernel_peers() for key, address in self.peers: require(key not in existing or existing[key] == [address], 'existing WG peer differs') wanted = ipaddress.IPv4Address(address[:-3]) for other, networks in existing.items(): if other != key: require(all(wanted not in ipaddress.ip_network(net) for net in networks), 'WG allowed-IP conflict') for key, address in self.peers: if key not in existing: self.run('wg', ['set', 'wg0', 'peer', key, 'allowed-ips', address]) final = self.kernel_peers() require(all(final.get(k) == [a] for k, a in self.peers) and all(final.get(k) == v for k, v in existing.items()), 'WG restoration verification failed') receipt = os.open('complete', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=journal) with os.fdopen(receipt, 'wb') as out: out.write(b'verified\n'); out.flush(); os.fsync(out.fileno()) os.fsync(journal) # Publish admission inputs only after effects and preservation checks. # Never replace a previous external plan implicitly. try: planfd = os.open('plan.json', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=journal) except FileExistsError: require(read(self.root, 'etc/archipelago-fips-wg/plan.json') == plan_bytes, 'deployment plan changed') else: with os.fdopen(planfd, 'wb') as out: out.write(plan_bytes); out.flush(); os.fsync(out.fileno()) os.fsync(journal) finally: os.close(journal) if __name__ == '__main__': import argparse parser = argparse.ArgumentParser() parser.add_argument('--root-fd', type=int, required=True) parser.add_argument('--bundle-fd', type=int, required=True) parser.add_argument('--network-fd', type=int, required=True) parser.add_argument('--activate', action='store_true') args = parser.parse_args() try: require(os.geteuid() == 0, 'external deployment principal must be root') Owner(args.root_fd, args.bundle_fd, args.network_fd).provision(args.activate) except Exception: # Deliberately never stringify secret-bearing parse/subprocess exceptions. raise SystemExit('FIPS/WG owner refused; preserve intent and reconcile externally') from None