215 lines
12 KiB
Python
215 lines
12 KiB
Python
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)
|