122 lines
9.9 KiB
Python
122 lines
9.9 KiB
Python
"""Installer library tests use an in-memory BFE model, never Windows mutation."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
sys.path.insert(0,str(Path(__file__).resolve().parent.parent/'apps/network-broker'))
|
|
from windows_wfp_install import BoundaryInstaller,InstallError,PROVIDER,SUBLAYER,MARKER,SERVICE,_key
|
|
from windows_wfp_probe import verify_observed
|
|
IDENTITY='\\device\\harddiskvolume3\\program files\\safetynet\\safetynet.exe\0'.encode('utf-16le').hex()
|
|
class Backend:
|
|
def __init__(self):self.objects={};self.calls=[];self.failure=None;self.post_failure=False;self.snapshot=None;self.tamper=False
|
|
def event(self,event):
|
|
self.calls.append(event)
|
|
if self.failure==event:raise InstallError('injected-native-failure',5)
|
|
def app_id(self,_path):self.event('app-id');return IDENTITY
|
|
def open(self):self.event('open')
|
|
def close(self):self.event('close')
|
|
def prepare_topology(self):return True
|
|
def begin(self):self.event('begin');self.snapshot=copy.deepcopy(self.objects)
|
|
def commit(self):self.event('commit');self.snapshot=None
|
|
def abort(self):self.event('abort');self.objects=self.snapshot;self.snapshot=None
|
|
def lookup(self,kind,key):self.event('lookup-'+kind);return copy.deepcopy(self.objects.get((kind,key)))
|
|
def add_provider(self):self.event('add-provider');self.objects[('provider',PROVIDER)]={'flags':1,'data':MARKER.hex(),'service':SERVICE}
|
|
def add_sublayer(self):self.event('add-sublayer');self.objects[('sublayer',SUBLAYER)]={'flags':1,'data':MARKER.hex(),'provider':PROVIDER,'weight':65535}
|
|
def add_filter(self,rule):
|
|
self.event('add-'+rule['id']);value={'provider':PROVIDER,'sublayer':SUBLAYER,'layer':rule['layer'],'flags':1,'action':0x1002 if rule['action']=='FWP_ACTION_PERMIT' else 0x1001,'conditions':copy.deepcopy(rule['conditions']),'weight':rule.get('weight')}
|
|
for condition in value['conditions']:
|
|
if condition['type']=='FWP_SECURITY_DESCRIPTOR_TYPE':
|
|
from windows_wfp_receiver import service_descriptor
|
|
import ctypes as C
|
|
with service_descriptor(condition['value'][len('D:(A;;CC;;;'):-1]) as blob:condition['value']=C.string_at(blob.data,blob.size).hex()
|
|
if self.tamper:value['action']=0x1002
|
|
self.objects[('filter',_key(rule['id']))]=value
|
|
def verify(self,plan):
|
|
self.event('verify')
|
|
if self.post_failure:return {'verified':False}
|
|
return verify_observed(plan,[v for (kind,_key_),v in self.objects.items() if kind=='filter'],PROVIDER,SUBLAYER)
|
|
def delete(self,kind,key):self.event('delete-'+kind);del self.objects[(kind,key)]
|
|
class InstallTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
root=Path(__file__).resolve().parent.parent
|
|
script="const {planBrowserNetworkBoundary}=require('./apps/network-broker/windows-wfp-plan.cjs');console.log(JSON.stringify(planBrowserNetworkBoundary({appId:Buffer.from(process.argv[1],'hex'),brokerPort:43871})));"
|
|
result=subprocess.run(['node','-e',script,IDENTITY],cwd=root,capture_output=True,text=True,timeout=10,check=True);cls.plan=json.loads(result.stdout)
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name);self.exe=self.root/'safetynet.exe';self.exe.write_bytes(b'SYNTHETIC EXE FIXTURE NEVER EXECUTED');self.sha=hashlib.sha256(self.exe.read_bytes()).hexdigest();self.backend=Backend();self.installer=BoundaryInstaller(self.backend,is_elevated=lambda:True)
|
|
def tearDown(self):self.temp.cleanup()
|
|
def apply(self,plan=None):return self.installer.apply(self.root,self.exe,self.sha,43871,plan or self.plan)
|
|
def rollback(self):return self.installer.rollback(self.root,self.exe,self.sha,43871,self.plan)
|
|
def test_atomic_order_and_exact_six_deny_filters(self):
|
|
result=self.apply();self.assertTrue(result['nativeVerified']);self.assertFalse(result['receiverTestsVerified']);self.assertEqual(len(self.backend.objects),8)
|
|
calls=self.backend.calls;self.assertLess(calls.index('begin'),calls.index('add-provider'));self.assertLess(calls.index('add-sublayer'),calls.index('add-v4-non-tcp'));self.assertLess(calls.index('add-v6-all'),calls.index('commit'));self.assertEqual(calls[-2:],['verify','close'])
|
|
def test_incomplete_topology_prevents_transaction_or_mutation(self):
|
|
self.backend.prepare_topology=lambda:False
|
|
with self.assertRaisesRegex(InstallError,'topology-admission-unavailable'):self.apply()
|
|
self.assertNotIn('begin',self.backend.calls)
|
|
self.assertFalse(any(call.startswith('add-') for call in self.backend.calls))
|
|
self.assertEqual(self.backend.calls[-1],'close')
|
|
def test_exact_eight_filter_atomic_receiver_install_and_rollback(self):
|
|
script="const {planServiceReceiver}=require('./apps/network-broker/wfp-topology.cjs');console.log(JSON.stringify(planServiceReceiver({appId:Buffer.from(process.argv[1],'hex'),brokerPort:43871,serviceSid:'S-1-5-80-1-2-3-4-5'})));"
|
|
result=subprocess.run(['node','-e',script,IDENTITY],cwd=Path(__file__).resolve().parents[1],capture_output=True,text=True,check=True,timeout=10)
|
|
receiver=json.loads(result.stdout)
|
|
plan=copy.deepcopy(self.plan);plan.update(version=3,brokerApplicationId=IDENTITY,serviceSid=receiver['serviceSid']);plan['filters']+=receiver['filters']
|
|
with self.assertRaisesRegex(InstallError,'receiver-install-requires-verified-broker'):self.apply(plan)
|
|
result=self.installer.apply_managed(self.root,self.exe,self.sha,self.exe,self.sha,43871,plan)
|
|
self.assertTrue(result['nativeVerified']);self.assertEqual(len(self.backend.objects),10)
|
|
permit=self.backend.objects[('filter',_key('broker-service-receiver'))]
|
|
self.assertEqual(permit['action'],0x1002);self.assertEqual(permit['weight'],2)
|
|
self.installer.rollback(self.root,self.exe,self.sha,43871,plan)
|
|
self.assertEqual(self.backend.objects,{})
|
|
def test_repeat_install_does_not_recreate_or_delete_existing_filters(self):
|
|
self.apply();self.backend.calls=[];self.assertTrue(self.apply()['existing']);self.assertFalse(any(c.startswith(('add-','delete-')) for c in self.backend.calls))
|
|
def test_administrator_required_before_any_backend_operation(self):
|
|
self.installer.is_elevated=lambda:False
|
|
with self.assertRaisesRegex(InstallError,'administrator-required'):self.apply()
|
|
self.assertEqual(self.backend.calls,[])
|
|
def test_manifest_hash_and_containment_enforced(self):
|
|
with self.assertRaisesRegex(InstallError,'installation-hash-mismatch'):self.installer.apply(self.root,self.exe,'0'*64,43871,self.plan)
|
|
with self.assertRaisesRegex(InstallError,'installation-path-invalid'):self.installer.apply(self.root/'missing',self.exe,self.sha,43871,self.plan)
|
|
def test_changed_app_id_or_permit_filter_refused_before_open(self):
|
|
for mutate in ['identity','permit','port','extra']:
|
|
plan=copy.deepcopy(self.plan)
|
|
if mutate=='identity':plan['applicationId']='00'
|
|
if mutate=='permit':plan['filters'][0]['action']='FWP_ACTION_PERMIT'
|
|
if mutate=='port':plan['filters'][2]['conditions'][-1]['value']=443
|
|
if mutate=='extra':plan['filters'].append(plan['filters'][0])
|
|
with self.assertRaises(InstallError):self.apply(plan)
|
|
self.assertNotIn('open',self.backend.calls)
|
|
def test_existing_provider_collision_does_not_add_delete(self):
|
|
self.backend.objects[('provider',PROVIDER)]={'flags':1,'data':'foreign','service':SERVICE}
|
|
with self.assertRaisesRegex(InstallError,'existing-boundary-collision'):self.apply()
|
|
self.assertEqual(self.backend.calls[-2:],['abort','close']);self.assertFalse(any(c.startswith(('add-','delete-')) for c in self.backend.calls))
|
|
def test_every_mutation_or_commit_error_aborts_and_closes(self):
|
|
for failure in ['add-provider','add-sublayer','add-v4-non-tcp','add-v4-tcp-non-broker-address','add-v4-tcp-other-local-port','add-v6-all','commit']:
|
|
self.backend=Backend();self.installer=BoundaryInstaller(self.backend,is_elevated=lambda:True);self.backend.failure=failure
|
|
with self.assertRaises(InstallError):self.apply()
|
|
self.assertEqual(self.backend.calls[-2:],['abort','close']);self.assertEqual(self.backend.objects,{})
|
|
def test_readback_tampering_aborts_before_commit(self):
|
|
self.backend.tamper=True
|
|
with self.assertRaisesRegex(InstallError,'added-filter-mismatch'):self.apply()
|
|
self.assertNotIn('commit',self.backend.calls);self.assertEqual(self.backend.objects,{})
|
|
def test_postcommit_verification_failure_keeps_denies_and_reports_failure(self):
|
|
self.backend.post_failure=True
|
|
with self.assertRaisesRegex(InstallError,'postcommit-verification-failed'):self.apply()
|
|
self.assertEqual(len(self.backend.objects),8);self.assertNotIn('abort',self.backend.calls);self.assertEqual(self.backend.calls[-1],'close')
|
|
def test_rollback_removes_only_exact_owned_six_filters_and_metadata(self):
|
|
self.apply();self.backend.calls=[];self.assertTrue(self.rollback()['removedOwnedBoundary']);self.assertEqual(self.backend.objects,{})
|
|
self.assertEqual([c for c in self.backend.calls if c.startswith('delete-')],['delete-filter']*6+['delete-sublayer','delete-provider'])
|
|
def test_rollback_metadata_collision_prevents_any_delete(self):
|
|
self.apply();self.backend.objects[('sublayer',SUBLAYER)]['data']='foreign';self.backend.calls=[]
|
|
with self.assertRaises(InstallError):self.rollback()
|
|
self.assertFalse(any(c.startswith('delete-') for c in self.backend.calls));self.assertEqual(len(self.backend.objects),8)
|
|
def test_rollback_error_restores_ownership_objects(self):
|
|
self.apply();self.backend.failure='delete-sublayer'
|
|
with self.assertRaises(InstallError):self.rollback()
|
|
self.assertEqual(len(self.backend.objects),8);self.assertEqual(self.backend.calls[-2:],['abort','close'])
|
|
if __name__=='__main__':unittest.main()
|