net-kingdom/tests/test_isolated_keycape_recovery.py

import base64
import hashlib
import importlib.util
import io
import json
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

SOURCE=Path(__file__).parents[1]/'tools/bootstrap-recovery/isolated-keycape.py'
spec=importlib.util.spec_from_file_location('adapter',SOURCE)
a=importlib.util.module_from_spec(spec);spec.loader.exec_module(a)

class ArchiveTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name)
        config={'data':{k:base64.b64encode(v).decode() for k,v in {'config.yaml':b'issuer: https://issuer.example','key.pem':b'fixture-key'}.items()}}
        self.values={'resources/sso/secret/keycape-config.json':json.dumps(config).encode(),
           'resources/sso/deployment/keycape.json':json.dumps({'spec':{'template':{'spec':{'containers':[{'name':'keycape','image':a.IMAGE,'env':[
               {'name':'KEYCAPE_CONFIG','value':'/etc/keycape/config.yaml'},
               {'name':'KEYCAPE_POLICY_PATH','value':'/var/lib/keycape-policy/policy.json'},
               {'name':'KEYCAPE_ORIGINAL_SECRET','valueFrom':{'secretKeyRef':{'name':'client','key':'secret'}}}]}]}}}}).encode(),
           'resources/sso/secret/client.json':json.dumps({'data':{'secret':base64.b64encode(b'fixture-private').decode()}}).encode(),
           'data/keycape/policy.json':b'{}','data/keycape/receipts/one.json':b'{"original":true}'}
        for name,value in self.values.items():
            p=self.root/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(value);p.chmod(0o600)
        self.manifest={'profile':{'components':[{'namespace':'sso','deployment':'keycape','secret_refs':['keycape-config','client'],'configmap_refs':[]}]},
          'files':[{'path':k,'bytes':len(v),'sha256':hashlib.sha256(v).hexdigest()} for k,v in self.values.items()]}
    def payload(self):
        digest=hashlib.sha256(json.dumps(self.manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest()
        return a.private_payload(self.root,self.manifest,digest)
    def test_original_state_and_secret_delivery_retained(self):
        p=self.payload();self.assertEqual(p['env']['KEYCAPE_ORIGINAL_SECRET'],'fixture-private')
        self.assertEqual(base64.b64decode(p['files']['/var/lib/keycape-policy/receipts/one.json']),b'{"original":true}')
    def test_archive_digest_refused(self):
        with self.assertRaises(a.Refused):a.private_payload(self.root,self.manifest,'0'*64)
    def test_changed_file_refused(self):
        (self.root/'data/keycape/policy.json').write_text('{"changed":true}')
        with self.assertRaises(a.Refused):self.payload()
    def test_symlink_refused(self):
        p=self.root/'data/keycape/policy.json';p.unlink();p.symlink_to('/etc/hostname')
        with self.assertRaises(Exception):self.payload()
    def test_duplicate_manifest_refused(self):
        self.manifest['files'].append(self.manifest['files'][0])
        with self.assertRaises(a.Refused):self.payload()
    def test_image_mismatch_refused(self):
        d=json.loads(self.values['resources/sso/deployment/keycape.json']);d['spec']['template']['spec']['containers'][0]['image']='keycape:latest'
        raw=json.dumps(d).encode();p=self.root/'resources/sso/deployment/keycape.json';p.write_bytes(raw)
        for row in self.manifest['files']:
            if row['path'].endswith('/deployment/keycape.json'):row.update(bytes=len(raw),sha256=hashlib.sha256(raw).hexdigest())
        with self.assertRaises(a.Refused):self.payload()


class OriginalSelectorTests(unittest.TestCase):
    """Already-restored public fixture config; no archive/key/native action."""
    def assert_refused_before_further_inputs(self,clients):
        import yaml
        cfg={'issuer':'https://issuer.example','clients':clients}
        payload={'files':{'/etc/keycape/config.yaml':base64.b64encode(yaml.safe_dump(cfg).encode()).decode(),
                          '/etc/keycape/key.pem':'deliberately-not-base64'}}
        admission={'issuer':'https://issuer.example','consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET'}}
        # Empty session/no env and an undecodable key also expose early private access.
        from contextlib import ExitStack
        with ExitStack() as stack:
            blocked=[stack.enter_context(patch.object(a,name,side_effect=AssertionError('later input or native call')))
                     for name in ('container_file','runtime_file','runtime_files','command','check_jwks','process')]
            with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
            for call in blocked:call.assert_not_called()
    def client(self):
        return {'clientId':'existing','clientType':'confidential','secretRef':'env:KEYCAPE_ORIGINAL_SECRET',
                'grantTypes':['client_credentials']}
    def test_absent_original_selector_refused_before_further_inputs(self):
        self.assert_refused_before_further_inputs([dict(self.client(),clientId='unrelated')])
    def test_identical_duplicate_selector_refused_before_further_inputs(self):
        self.assert_refused_before_further_inputs([self.client(),self.client()])
    def test_conflicting_duplicate_selector_refused_before_further_inputs(self):
        self.assert_refused_before_further_inputs([self.client(),dict(self.client(),secretRef='env:OTHER')])
    def test_empty_registry_refused_before_further_inputs(self):
        self.assert_refused_before_further_inputs([])

class NativeTests(unittest.TestCase):
    def setUp(self):
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric import rsa
        self.key=rsa.generate_private_key(public_exponent=65537,key_size=2048).private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption())
        self.cfg=b'issuer: https://issuer.example\nclients:\n- clientId: existing\n  clientType: confidential\n  secretRef: env:KEYCAPE_ORIGINAL_SECRET\n  grantTypes: [client_credentials]\n'
        self.files={'/etc/keycape/config.yaml':self.cfg,'/etc/keycape/key.pem':self.key,'/var/lib/keycape-policy/policy.json':b'{}'}
        self.payload={'env':{'KEYCAPE_ORIGINAL_SECRET':'fixture-private'},'files':{k:base64.b64encode(v).decode() for k,v in self.files.items()}}
        self.admission={'issuer':'https://issuer.example','verifier_sha256':hashlib.sha256(b'binary').hexdigest(),
          'trust_path':'/trust.pem','trust_sha256':hashlib.sha256(b'original-trust').hexdigest(),
          'consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET','scope':'read','audience':'consumer',
                      'subject':'service:existing','tenant':'tenant:platform','roles':'reader','deny_scope':'write'}}
        self.session={'issuer':'isolated','volumes':{x:{'target':x} for x in ('/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls')}}
        self.output=''.join('PASS  '+x+'\n' for x in ['discovery and endpoint origin','exchange and JWKS signature for granted scopes','access-token claim bindings','principal_type is service','sub matches the registration','tenant matches the registration','roles present','granted scope carries no excess grant','predecessor secret refused','excess scope refused: write'])
    def file(self,container,path):
        return {**self.files,'/etc/hosts':b'127.0.0.1 issuer.example\n','/keycape':b'binary','/trust.pem':b'original-trust'}[path]
    def command(self,argv,**kwargs):
        if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem']).encode()
        output=io.BytesIO()
        with tarfile.open(fileobj=output,mode='w') as t:
            m=tarfile.TarInfo('policy.json');m.size=2;t.addfile(m,io.BytesIO(b'{}'))
        return output.getvalue()
    def verify(self,code=0,output=None):
        with patch.object(a,'check_jwks'),patch.object(a,'runtime_file',return_value=b'original-trust'),patch.object(a,'runtime_files',side_effect=lambda session,paths=(),tree=False:({k:v for k,v in self.files.items() if k.startswith('/var/lib/keycape-policy/')} if tree else dict(self.files))),patch.object(a,'container_file',side_effect=self.file),patch.object(a,'command',side_effect=self.command),patch.object(a.subprocess,'run',return_value=subprocess.CompletedProcess([],code,b'',(self.output if output is None else output).encode())) as run:
            result=a.run_verifier(self.session,self.admission,self.payload)
            argv=run.call_args.args[0]
            self.assertNotIn('fixture-private',' '.join(argv))
            self.assertIn('-previous-secret-env',argv)
            return result
    def test_original_registration_and_unrelated_secret(self):
        r=self.verify();self.assertTrue(r['wrong_unrelated_secret_explicitly_refused']);self.assertNotIn('fixture-private',json.dumps(r))
    def test_transport_failure_never_counts_as_wrong_secret(self):
        with self.assertRaises(a.Refused):self.verify(1,'FAIL  discovery and endpoint origin: TLS failure\n')
    def test_missing_explicit_rejection_never_counts(self):
        with self.assertRaises(a.Refused):self.verify(0,self.output.replace('PASS  predecessor secret refused\n',''))
    def test_fail_line_refused_even_zero_exit(self):
        with self.assertRaises(a.Refused):self.verify(0,self.output+'FAIL  transport\n')
    def test_security_environment_override_refused(self):
        original=self.command
        def altered(argv,**kwargs):
            if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem','SSL_CERT_DIR=/unreviewed']).encode()
            return original(argv,**kwargs)
        with patch.object(self,'command',side_effect=altered):
            with self.assertRaises(a.Refused):self.verify()
    def test_http_issuer_refused(self):
        self.admission['issuer']='http://issuer.example'
        with self.assertRaises(a.Refused):self.verify()
    def test_changed_trust_refused(self):
        self.admission['trust_sha256']='0'*64
        with self.assertRaises(a.Refused):self.verify()
    def test_config_drift_refused(self):
        self.payload['files']['/etc/keycape/config.yaml']=base64.b64encode(b'changed').decode()
        with self.assertRaises(a.Refused):self.verify()

class PublicReadinessTransportTests(unittest.TestCase):
    def test_provider_probe_failure_is_closed_false_data_not_failed_docker_exec(self):
        output=io.StringIO()
        with patch('sys.stdin',io.StringIO('{}')),patch('sys.stdout',output):
            exec(compile(a.JWKS_TRANSPORT,'<public-readiness>','exec',optimize=2),{})
        self.assertEqual(json.loads(output.getvalue()),{'original_public_signing_key_in_live_jwks':False})

class BoundaryTests(unittest.TestCase):
    def test_remote_docker_refused_before_info(self):
        session={'run_id':'a'*32,'containers':['nk49-kc-'+('a'*32)+'-worker'],'worker':'nk49-kc-'+('a'*32)+'-worker','issuer':'nk49-kc-'+('a'*32)+'-worker'}
        with patch.object(a,'command',return_value=b'[{"Endpoints":{"docker":{"Host":"tcp://remote:2375"}}}]') as command:
            with self.assertRaises(a.Refused):a.inspect_session(session)
            self.assertEqual(command.call_count,1)
    def test_failed_verifier_cleanup_is_attempted(self):
        receipt={};admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
        with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',side_effect=a.Refused),patch.object(a,'cleanup') as cleanup:
            with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
            cleanup.assert_called_once();self.assertTrue(receipt['container_cleanup_verified']);self.assertEqual(receipt['status'],'failed')

    def test_cleanup_failure_overrides_success(self):
        admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
        result={x:True for x in ('native_service_registration_verified','wrong_unrelated_secret_explicitly_refused','denied_scopes_refused','original_config_registry_policy_receipts_equal','original_https_issuer_and_trust_preserved','original_public_signing_key_in_live_jwks')}
        result['public_signing_spki_sha256']='e'*64;receipt={}
        with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',return_value=result),patch.object(a,'cleanup',side_effect=a.Refused):
            with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
        self.assertEqual(receipt['status'],'failed');self.assertFalse(receipt['container_cleanup_verified'])

class ContainmentTests(unittest.TestCase):
    def inspect(self,overrides=None,launch=None,mounts=None):
        name='nk49-kc-'+('a'*32)+'-issuer'
        session={'run_id':'a'*32,'containers':[name],'issuer':name,'worker':name,'images':{name:a.IMAGE}}
        host={'NetworkMode':'none','ReadonlyRootfs':True,'Privileged':False,'CapDrop':['ALL'],'SecurityOpt':['no-new-privileges']}
        host.update(overrides or {})
        def command(argv,**kwargs):
            if argv[1:3]==['context','inspect']:return b'[{"Endpoints":{"docker":{"Host":"unix:///var/run/docker.sock"}}}]'
            if argv[1]=='info':return b'{"ID":"74f93783-6cef-43a8-a3c4-f62586cd3bc1","Name":"bnt-lap001","OSType":"linux"}'
            if argv[-1]=='{{.Id}}':return b'anchor-id'
            if '"path"' in argv[-1]:return json.dumps(launch or {'entrypoint':['/keycape'],'cmd':None,'path':'/keycape','args':[]}).encode()
            return json.dumps({'host':host,'mounts':mounts or [],'labels':{a.LABEL:'a'*32},'image':a.IMAGE,'user':'1000:1000'}).encode()
        with patch.object(a,'command',side_effect=command),patch.dict('os.environ',{'DOCKER_HOST':'unix:///var/run/docker.sock'}):a.inspect_session(session)
    def test_minimal_network_none_admitted(self):self.inspect()
    def test_image_declared_anonymous_disk_volume_remains_refused(self):
        with self.assertRaises(a.Refused):
            self.inspect(mounts=[{'Type':'volume','Name':'anonymous-fixture','Destination':'/privacyidea/etc/persistent','RW':True}])

    def test_alternate_same_image_command_refused(self):
        with self.assertRaises(a.Refused):self.inspect(launch={'entrypoint':['/keycape'],'cmd':['check-config'],'path':'/keycape','args':['check-config']})
    def test_cap_add_refused(self):
        with self.assertRaises(a.Refused):self.inspect({'CapAdd':['NET_ADMIN']})
    def test_devices_refused(self):
        with self.assertRaises(a.Refused):self.inspect({'Devices':[{'PathOnHost':'/dev/mem'}]})
    def test_device_requests_refused(self):
        with self.assertRaises(a.Refused):self.inspect({'DeviceRequests':[{'Driver':'nvidia'}]})
    def test_host_namespace_refused(self):
        for field in ('PidMode','IpcMode','UTSMode','UsernsMode','CgroupnsMode'):
            with self.subTest(field=field),self.assertRaises(a.Refused):self.inspect({field:'host'})
    def test_security_override_refused(self):
        with self.assertRaises(a.Refused):self.inspect({'SecurityOpt':['no-new-privileges','seccomp=unconfined']})

class CleanupTests(unittest.TestCase):
    def test_partial_cleanup_attempts_every_owned_container(self):
        session={'run_id':'a'*32,'containers':['first','second','third']};receipt={}
        removed=[]
        def command(argv,**kwargs):
            if argv[1:3]==['container','inspect']:return ('a'*32).encode()
            if argv[1]=='rm':
                removed.append(argv[-1])
                if argv[-1]=='third':raise a.Refused()
                return b''
            if argv[1:3]==['container','ls']:
                name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
                return (name+'\n').encode() if name=='third' or name not in removed else b''
            return b''
        with patch.object(a,'command',side_effect=command):
            with self.assertRaises(a.Refused):a.cleanup(session,receipt)
        self.assertEqual(removed,['third','second','first'])
        self.assertEqual([r['status'] for r in receipt['container_cleanup_obligations']],['still_present','absence_verified','absence_verified'])
    def test_ownership_mismatch_does_not_remove_and_continues(self):
        session={'run_id':'a'*32,'containers':['first','second']};receipt={};removed=[]
        def command(argv,**kwargs):
            if argv[1:3]==['container','inspect']:return (('b' if argv[3]=='second' else 'a')*32).encode()
            if argv[1]=='rm':removed.append(argv[-1])
            if argv[1:3]==['container','ls']:
                name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
                return (name+'\n').encode() if name not in removed else b''
            return b''
        with patch.object(a,'command',side_effect=command):
            with self.assertRaises(a.Refused):a.cleanup(session,receipt)
        self.assertEqual(removed,['first'])
        self.assertEqual(receipt['container_cleanup_obligations'][0]['status'],'ownership_mismatch_not_removed')
    def test_inspect_failure_retains_uncertain_obligation(self):
        receipt={}
        with patch.object(a,'command',side_effect=a.Refused):
            with self.assertRaises(a.Refused):a.cleanup({'run_id':'a'*32,'containers':['exact-owned-name']},receipt)
        self.assertEqual(receipt['container_cleanup_obligations'],[{'container':'exact-owned-name','status':'unverified'}])

class JWKSTests(unittest.TestCase):
    def execute(self,n='123',endpoint='https://issuer.example/jwks',resolved='127.0.0.1',extra=False,kid='key-1'):
        request={'issuer':'https://issuer.example','trust':'public-certificate','n':'123','e':'65537'}
        def enc(i):return base64.urlsafe_b64encode(int(i).to_bytes((int(i).bit_length()+7)//8,'big')).decode().rstrip('=')
        responses=[{'issuer':'https://issuer.example','jwks_uri':endpoint},{'keys':[{'kty':'RSA','kid':kid,'use':'sig','alg':'RS256','n':enc(n),'e':enc(65537)}]}]
        if extra:responses[1]['keys'].append(dict(responses[1]['keys'][0],kid='key-2'))
        class Response:
            def __init__(self,value):self.value=value
            def __enter__(self):return self
            def __exit__(self,*args):pass
            def read(self,*args):return json.dumps(self.value).encode()
        class Opener:
            def open(self,*args,**kwargs):return Response(responses.pop(0))
        with patch('sys.stdin',io.StringIO(json.dumps(request))),patch('sys.stdout',io.StringIO()),patch('socket.getaddrinfo',return_value=[(2,1,6,'',(resolved,443))]),patch('ssl.create_default_context'),patch('urllib.request.build_opener',return_value=Opener()):
            exec(compile(a.JWKS_PROBE,'<jwks-probe>','exec',optimize=2),{})
    def test_exact_original_public_key(self):self.execute()
    def test_extra_signing_key_refused_under_optimization(self):
        with self.assertRaises(ValueError):self.execute(extra=True)
    def test_wrong_kid_refused_under_optimization(self):
        with self.assertRaises(ValueError):self.execute(kid='replacement')
    def test_wrong_live_signing_key_refused(self):
        with self.assertRaises(ValueError):self.execute(n='124')
    def test_cross_origin_jwks_refused(self):
        with self.assertRaises(ValueError):self.execute(endpoint='https://production.example/jwks')
    def test_production_dns_refused_before_fetch(self):
        with self.assertRaises(ValueError):self.execute(resolved='192.0.2.1')

if __name__=='__main__':unittest.main()