import base64 import importlib.util import json import os from pathlib import Path import shutil import subprocess import tempfile import time import unittest SPEC = importlib.util.spec_from_file_location('owner', Path(__file__).parents[1] / 'fips_wg_owner.py') owner = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(owner) def encode_key(hrp, key): alphabet = 'qpzry9x8gf2tvdw0s3jn54khce6mua7l' bits = ''.join(f'{b:08b}' for b in key) + '0000' values = [int(bits[i:i+5], 2) for i in range(0, len(bits), 5)] chk = 1 for v in [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] + values + [0]*6: top = chk >> 25; chk = ((chk & 0x1ffffff) << 5) ^ v for i, g in enumerate([0x3b6a57b2, 0x26508e6d, 0x1ea119fa, 0x3d4233dd, 0x2a1462b3]): if (top >> i) & 1: chk ^= g chk ^= 1 return (hrp + '1' + ''.join(alphabet[v] for v in values) + ''.join(alphabet[(chk >> (5*(5-i))) & 31] for i in range(6))).encode() class Executor(unittest.TestCase): @classmethod def setUpClass(cls): if os.environ.get('FIPS_WG_NAMESPACE_TEST') == '1': assert os.stat('/proc/self/ns/mnt').st_ino != os.stat('/proc/1/ns/mnt').st_ino # FIPS's control socket must never see the real host /run. subprocess.run(['mount', '--make-rprivate', '/'], check=True) for path in ('/run', '/etc/fips', '/var/lib', '/tmp'): subprocess.run(['mount', '-t', 'tmpfs', '-o', 'mode=0755,size=128m', 'tmpfs', path], check=True) def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) / 'root' self.bundle = Path(self.temp.name) / 'bundle' self.root.mkdir(mode=0o700); self.bundle.mkdir(mode=0o700) for path, _ in owner.ASSETS.values(): (self.root / path).parent.mkdir(parents=True, exist_ok=True) self.assets = {name: b'# exact externally reviewed fixture\n' for name in owner.ASSETS} self.assets['fips.yaml'] = b'node:\n identity:\n persistent: true\n' self.assets['fips.key'] = b'0' * 63 + b'1' self.assets['fips.pub'] = owner.public_from_secret(self.assets['fips.key']) self.assets['wg.public'] = base64.b64encode(bytes([7]) * 32) self.plan = dict(version=1, assets={}, programs={'wg': '0'*64, 'systemctl': '0'*64}, fips='stopped', wg='stopped', peers=[]) for name, data in self.assets.items(): (self.bundle / name).write_bytes(data) (self.bundle / name).chmod(0o600) self.plan['assets'][name] = owner.digest(data) self.rootfd = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY) self.bundlefd = os.open(self.bundle, os.O_RDONLY | os.O_DIRECTORY) self.netfd = os.open('/proc/self/ns/net', os.O_RDONLY) for fd in (self.rootfd, self.bundlefd, self.netfd): self.addCleanup(os.close, fd) def make(self): (self.bundle / 'plan.json').write_text(json.dumps(self.plan)) return owner.Owner(self.rootfd, self.bundlefd, self.netfd) def test_actual_install_verify_preserve_and_replay_refusal(self): instance = self.make() instance.provision() before = {name: (self.root / path).stat().st_ino for name, (path, _) in owner.ASSETS.items()} self.assertEqual(instance.verify({'FipsIdentity': {'public': list(self.assets['fips.pub'])}}), b'ok') with self.assertRaises(FileExistsError): instance.provision() for name, (path, _) in owner.ASSETS.items(): self.assertTrue((self.root / path).read_bytes() == self.assets[name], 'protected asset bytes changed') self.assertEqual((self.root / path).stat().st_ino, before[name]) self.assertEqual((self.root / 'etc/fips/fips.key').stat().st_mode & 0o777, 0o600) def test_differing_existing_config_refuses_before_any_write(self): path = self.root / 'etc/fips/fips.yaml' path.write_bytes(b'operator config and static peers') inode = path.stat().st_ino with self.assertRaises(owner.Refused): self.make().provision() self.assertEqual(path.stat().st_ino, inode) self.assertFalse((self.root / 'etc/fips/fips.key').exists()) self.assertFalse((self.root / 'etc/archipelago-fips-wg/intent').exists()) def test_identity_and_source_tampering_refuse(self): (self.bundle / 'fips.key').write_bytes(b'0'*63+b'2') with self.assertRaises(owner.Refused): self.make() self.plan['assets']['fips.key'] = owner.digest(b'0'*63+b'2') with self.assertRaises(owner.Refused): self.make() (self.bundle / 'fips.key').write_bytes(self.assets['fips.key']) self.plan['assets']['fips.key'] = owner.digest(self.assets['fips.key']) changed = b'node:\n identity:\n persistent: false\n' (self.bundle / 'fips.yaml').write_bytes(changed) self.plan['assets']['fips.yaml'] = owner.digest(changed) with self.assertRaises(owner.Refused): self.make() def test_symlink_hardlink_and_key_mode_refuse(self): key = self.bundle / 'fips.key' key.chmod(0o644) with self.assertRaises(owner.Refused): self.make() key.chmod(0o600) os.link(key, self.bundle / 'alias') with self.assertRaises(owner.Refused): self.make() (self.bundle / 'alias').unlink() path = self.root / 'etc/fips/fips.yaml' path.symlink_to(self.bundle / 'fips.yaml') with self.assertRaises(OSError): self.make().provision() def test_stopped_and_uninstalled_decisions(self): instance = self.make(); instance.provision() with self.assertRaises(owner.Refused): instance.verify('FipsActive') self.plan['fips'] = 'uninstalled' (self.root / 'etc/fips/fips.yaml').unlink() with self.assertRaises(owner.Refused): self.make().provision() def test_peer_validation_and_collisions(self): key = base64.b64encode(bytes([4])*32).decode() for ip in ('0.0.0.0/0', '10.44.0.1/32', '10.44.0.255/32', '10.44.1.2/32'): with self.assertRaises(owner.Refused): owner.peer(dict(public_key=key, ip=ip)) self.plan['peers'] = [dict(public_key=key, ip='10.44.0.2/32')] * 2 with self.assertRaises(owner.Refused): self.make() @unittest.skipUnless(os.environ.get('FIPS_WG_NAMESPACE_TEST') == '1', 'requires disposable mount/network namespaces') def test_actual_fips_config_and_identity_start(self): self.assertNotEqual(os.stat('/proc/self/ns/net').st_ino, int(os.environ['FIPS_WG_HOST_NET_INODE'])) # Exact source snapshot is asserted against typed rendering by Rust tests. source = (Path(__file__).parents[2] / 'core/archipelago/src/fips/config.rs').read_text() yaml = source.split('let expected = "\\\n', 1)[1].split('\n";', 1)[0] + '\n' directory = self.root / 'etc/fips' encoded = encode_key('nsec', (1).to_bytes(32, 'big')) self.assertEqual(owner.public_from_secret(encoded), self.assets['fips.pub']) # Match identity::write_fips_key_from_seed's canonical nsec line. public_line = encode_key('npub', self.assets['fips.pub']) + b'\n' self.assertTrue(owner.public_bytes(public_line) == self.assets['fips.pub']) public = self.assets['fips.pub'] for name, data in [('fips.yaml', yaml.encode()), ('fips.key', encoded + b'\n'), ('fips.pub', public_line)]: self.assets[name] = data (self.bundle / name).write_bytes(data) self.plan['assets'][name] = owner.digest(data) self.plan['fips'] = 'running' instance = self.make() instance.provision() self.assertEqual(instance.verify({'FipsIdentity': {'public': list(public)}}), b'ok') before = {name: ((directory / name).stat().st_ino, (directory / name).read_bytes()) for name in ('fips.key', 'fips.pub', 'fips.yaml')} with subprocess.Popen(['/usr/bin/fips', '--config', str(directory / 'fips.yaml')], cwd=directory, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) as child: try: deadline = time.monotonic() + 5 ready = False while time.monotonic() < deadline and child.poll() is None: result = subprocess.run(['/usr/sbin/ip', 'link', 'show', 'fips0'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if result.returncode == 0: ready = True; break time.sleep(0.05) self.assertTrue(ready, 'actual FIPS failed to establish disposable fips0') self.assertIsNone(child.poll()) for name, (inode, data) in before.items(): self.assertTrue((directory / name).read_bytes() == data, f'daemon changed protected input {name}; length={(directory / name).stat().st_size}') self.assertEqual((directory / name).stat().st_ino, inode) self.assertEqual(instance.verify({'FipsIdentity': {'public': list(public)}}), b'ok') finally: if child.poll() is None: child.terminate() try: child.wait(timeout=2) except subprocess.TimeoutExpired: child.kill(); child.wait() @unittest.skipUnless(os.environ.get('FIPS_WG_NAMESPACE_TEST') == '1', 'requires dedicated disposable network namespace') def test_real_wireguard_restoration_and_conflict_preservation(self): # Runner creates a fresh network namespace. Never execute on the host. require_private = os.stat('/proc/self/ns/net').st_ino != int(os.environ['FIPS_WG_HOST_NET_INODE']) self.assertTrue(require_private) executable = Path('/usr/bin/wg') dest = self.root / 'usr/bin/wg' dest.parent.mkdir(parents=True) shutil.copyfile(executable, dest); dest.chmod(0o755) self.plan['programs']['wg'] = owner.digest(dest.read_bytes()) private = subprocess.check_output(['/usr/bin/wg', 'genkey']) public = subprocess.check_output(['/usr/bin/wg', 'pubkey'], input=private).strip() self.assets['wg.public'] = public (self.bundle / 'wg.public').write_bytes(public) self.plan['assets']['wg.public'] = owner.digest(public) subprocess.run(['/usr/sbin/ip', 'link', 'add', 'wg0', 'type', 'wireguard'], check=True) self.addCleanup(subprocess.run, ['/usr/sbin/ip', 'link', 'del', 'wg0'], check=True) keyfile = self.root / 'test-key' fd = os.open(keyfile, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) with os.fdopen(fd, 'wb') as output: output.write(private) subprocess.run(['/usr/bin/wg', 'set', 'wg0', 'private-key', str(keyfile)], check=True) keyfile.unlink() key = base64.b64encode(bytes([4])*32).decode() unrelated = base64.b64encode(bytes([5])*32).decode() subprocess.run(['/usr/bin/wg', 'set', 'wg0', 'peer', unrelated, 'allowed-ips', '10.44.0.3/32'], check=True) self.plan['wg'] = 'running' self.plan['peers'] = [dict(public_key=key, ip='10.44.0.2/32')] instance = self.make(); instance.provision() self.assertEqual(instance.kernel_peers(), {key: ['10.44.0.2/32'], unrelated: ['10.44.0.3/32']}) self.assertEqual(instance.verify({'WgPeer': {'public': [4]*32, 'address': '10.44.0.2'}}), b'ok') # A new external plan with a colliding IP refuses and retains both peers. self.plan['peers'] = [dict(public_key=base64.b64encode(bytes([6])*32).decode(), ip='10.44.0.3/32')] # Move the completed transaction aside, never clear an uncertain intent. journal = self.root / 'etc/archipelago-fips-wg' (journal / 'intent').rename(journal / 'previous-intent') (journal / 'complete').rename(journal / 'previous-complete') (journal / 'plan.json').rename(journal / 'previous-plan.json') next_owner = self.make() with self.assertRaises(owner.Refused): next_owner.provision() self.assertFalse((journal / 'complete').exists()) self.assertEqual(next_owner.kernel_peers(), instance.kernel_peers()) with self.assertRaises(FileExistsError): next_owner.provision() if __name__ == '__main__': unittest.main(verbosity=2)