82 lines
4.2 KiB
Python
82 lines
4.2 KiB
Python
"""Conditional admission refusals; no service mutations or host test commands."""
|
|
import copy
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import sys
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
import c01_admission as admission
|
|
|
|
|
|
class AdmissionTests(unittest.TestCase):
|
|
def policy(self):
|
|
return {'version':1,'platform_sha256':'a'*64,'roles':sorted(admission.ROLES),'required_routes':['management'],
|
|
'assets':{k:'b'*64 for k in admission.ASSETS},
|
|
'services':{k:'uninstalled' for k in admission.UNITS},
|
|
'optional_unavailable':sorted(admission.UNITS)}
|
|
|
|
def test_required_contract_cannot_be_replaced_by_ready_flags(self):
|
|
value=self.policy();launch={'platform_profile_sha256':'a'*64}
|
|
admission.validate(value,launch)
|
|
for change in ({'ready':True},{'roles':['application']},{'assets':{}},
|
|
{'platform_sha256':'c'*64},{'optional_unavailable':[]}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):
|
|
admission.validate({**copy.deepcopy(value),**change},launch)
|
|
|
|
def test_running_service_needs_its_executable_and_unit_pins(self):
|
|
value=self.policy();value['services']['fips']='running'
|
|
value['optional_unavailable'].remove('fips')
|
|
with self.assertRaisesRegex(ValueError,'asset set'):
|
|
admission.validate(value,{'platform_profile_sha256':'a'*64})
|
|
value['assets'].update({name:'d'*64 for name in admission.SERVICE_ASSETS['fips']})
|
|
with self.assertRaisesRegex(ValueError,'service route'):
|
|
admission.validate(value,{'platform_profile_sha256':'a'*64})
|
|
value['required_routes'].append('fips-dns')
|
|
admission.validate(value,{'platform_profile_sha256':'a'*64})
|
|
|
|
def test_asset_digest_failure_does_not_leak_descriptor(self):
|
|
path=Path('/usr/bin/true')
|
|
before=len(os.listdir('/proc/self/fd'))
|
|
fd,metadata=admission.open_asset(path,hashlib.sha256(path.read_bytes()).hexdigest())
|
|
self.assertEqual(metadata[:2],(path.stat().st_dev,path.stat().st_ino));os.close(fd)
|
|
with self.assertRaisesRegex(ValueError,'digest'):
|
|
admission.open_asset(path,'0'*64)
|
|
self.assertEqual(len(os.listdir('/proc/self/fd')),before)
|
|
|
|
def test_installed_fips_identity_can_differ_from_nostr_without_rotation(self):
|
|
# Public-only installed Framework vector: the FIPS seed-derived key is
|
|
# independent of its Nostr identity. This npub is the retained file.
|
|
public=bytes.fromhex('d157c1dd326da3ed03882afa9718937879fe59ccd8e48cc7059b413c071a5cf3')
|
|
npub=b'npub169turhfjdk376qug9tafwxyn0pulukwvmrjge3c9ndqncpc6tnesrn4wa6\n'
|
|
for encoded in (public,public.hex().encode(),npub,npub.upper()):
|
|
admission.require_public_identity(encoded,public)
|
|
foreign=bytes.fromhex('07cb23083180dc86c542b31f95a61670d1d7cb965d53168c5356967515533405')
|
|
with self.assertRaisesRegex(ValueError,'differs'):
|
|
admission.require_public_identity(npub,foreign)
|
|
with self.assertRaises(ValueError):
|
|
admission.require_public_identity(npub[:-3]+b'aa\n',public)
|
|
policy=dict(self.policy(),version=2,fips_public_key=public.hex(),service_argv_sha256={name:None for name in admission.UNITS})
|
|
admission.validate(policy,{'platform_profile_sha256':'a'*64})
|
|
for value in ('0'*64,public.hex().upper(),'bad',None):
|
|
with self.assertRaises(ValueError):
|
|
admission.validate(dict(policy,fips_public_key=value),{'platform_profile_sha256':'a'*64})
|
|
|
|
def test_checker_pid_credentials_and_deadline_are_required(self):
|
|
a,b=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
a.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
|
|
hold=admission.Hold.__new__(admission.Hold)
|
|
hold.record=SimpleNamespace(channel=a,pid=os.getpid()+1);hold.deadline=time.monotonic()+3
|
|
try:
|
|
b.send(b'ADMISSION_HELD')
|
|
with self.assertRaisesRegex(ValueError,'credentials'):hold.check()
|
|
hold.deadline=0
|
|
with self.assertRaisesRegex(ValueError,'lifetime'):hold.check()
|
|
finally:a.close();b.close()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|