"""Trusted signed-installer library. No CLI, automatic elevation or live apply. The installer must authenticate its manifest, verify installation ACLs and service provenance, stop browser processes, and own rollback. This module does not prove those prerequisites or replace physical receiver tests. Filters never permit. """ import ctypes as C import copy import hashlib import os import uuid from contextlib import ExitStack from pathlib import Path from windows_wfp_probe import GUID,Blob,Display,Value,Condition,Filter,U32,ReadOnlyWfp,_blob,_decode,verify_observed PROVIDER='2a7bcbf2-e4f5-4e94-bd33-d5442a3bef50' SUBLAYER='403a69b8-fc7c-460a-a247-72e0d72e6bf8' SERVICE='SafetyNetNetworkBroker' MARKER=b'SafetyNet.NetworkBoundary.v1' OWNED_PRIORITY=65535 L4='c38d57d1-05a7-4c33-904f-7fbceee60e82' L6='4a72393b-319f-44bc-84c3-ba54dcb3b6b4' R4='e1cd9fe7-f4b5-4273-96c0-592e487b8650' R6='a3b42c97-9f04-4672-b87e-cee9c483257f' APP='d78e1e87-8644-4ea5-9437-d809ecefc971' PROTO='3971ef2b-623e-4f9a-8cb1-6e79b806b9a7' ADDR='b235ae9a-1d64-49b8-a44c-5ff3d9095045' PORT='c35a604d-d22b-4e1a-91b4-68f674ee674b' RULE_IDS=['v4-non-tcp','v4-tcp-non-broker-address','v4-tcp-other-local-port','v6-all','browser-v4-incoming','browser-v6-incoming'] _MANAGED_TOKEN=object() class InstallError(RuntimeError): def __init__(self,reason,code=0):super().__init__(reason);self.reason=reason;self.code=int(code) def _guid(text):return GUID.from_buffer_copy(uuid.UUID(text).bytes_le) def _key(rule):return str(uuid.uuid5(uuid.UUID(PROVIDER),rule)) def _owned_metadata(value,kind): if not value:return False return value.get('flags')==1 and value.get('data')==MARKER.hex() and (value.get('service')==SERVICE if kind=='provider' else value.get('provider')==PROVIDER and value.get('weight')==OWNED_PRIORITY) def _validate(plan,identity,port): if not isinstance(port,int) or isinstance(port,bool) or not 1<=port<=65535:raise InstallError('invalid-broker-port') def c(field,match,type_,value):return {'field':field,'match':match,'type':type_,'value':value} app=c(APP,'equal','FWP_BYTE_BLOB_TYPE',identity) tcp=c(PROTO,'equal','FWP_UINT8',6) specifications=[(L4,[app,c(PROTO,'not-equal','FWP_UINT8',6)]),(L4,[app,tcp,c(ADDR,'not-equal','FWP_UINT32',0x7f000001)]),(L4,[app,tcp,c(ADDR,'equal','FWP_UINT32',0x7f000001),c(PORT,'not-equal','FWP_UINT16',port)]),(L6,[app]),(R4,[app]),(R6,[app])] managed=plan.get('version')==3 if not isinstance(plan,dict) or plan.get('applicationId')!=identity or plan.get('version') not in (2,3) or len(plan.get('filters',[]))!=(8 if managed else 6):raise InstallError('plan-identity-mismatch') for rule,rule_id,(layer,conditions) in zip(plan['filters'],RULE_IDS,specifications): if rule.get('id')!=rule_id or rule.get('layer')!=layer or rule.get('action')!='FWP_ACTION_BLOCK' or rule.get('clearActionRight') is not False or rule.get('conditions')!=conditions:raise InstallError('unsafe-plan') if managed: from windows_wfp_receiver import validate_receiver_rule,service_sddl pair=plan['filters'][6:] if [rule.get('id') for rule in pair]!=['broker-service-receiver','broker-receiver-deny']:raise InstallError('unsafe-receiver-plan') for rule in pair:validate_receiver_rule(rule) if any(rule['conditions'][2]['value']!=port for rule in pair) or pair[0]['conditions'][3]['value']!=plan.get('brokerApplicationId') or pair[0]['conditions'][4]['value']!=service_sddl(plan.get('serviceSid')):raise InstallError('unsafe-receiver-plan') return plan def elevated(): if os.name!='nt':return False k=C.WinDLL('kernel32',use_last_error=True);a=C.WinDLL('advapi32',use_last_error=True) k.GetCurrentProcess.restype=C.c_void_p;k.CloseHandle.argtypes=[C.c_void_p] a.OpenProcessToken.argtypes=[C.c_void_p,U32,C.POINTER(C.c_void_p)] a.GetTokenInformation.argtypes=[C.c_void_p,U32,C.c_void_p,U32,C.POINTER(U32)] token=C.c_void_p();value=U32();size=U32() if not a.OpenProcessToken(k.GetCurrentProcess(),8,C.byref(token)):return False try:return bool(a.GetTokenInformation(token,20,C.byref(value),4,C.byref(size)) and value.value) finally:k.CloseHandle(token) class Provider(C.Structure): _fields_=[('key',GUID),('display',Display),('flags',U32),('data',Blob),('service',C.c_wchar_p)] class Sublayer(C.Structure): _fields_=[('key',GUID),('display',Display),('flags',U32),('provider',C.POINTER(GUID)),('data',Blob),('weight',C.c_uint16)] class NativeBackend: def __init__(self): self.reader=ReadOnlyWfp();self.api=self.reader.api;self.engine=C.c_void_p() for name,args in { 'FwpmTransactionBegin0':[C.c_void_p,U32], 'FwpmTransactionCommit0':[C.c_void_p], 'FwpmTransactionAbort0':[C.c_void_p], 'FwpmProviderGetByKey0':[C.c_void_p,C.POINTER(GUID),C.POINTER(C.POINTER(Provider))], 'FwpmSubLayerGetByKey0':[C.c_void_p,C.POINTER(GUID),C.POINTER(C.POINTER(Sublayer))], 'FwpmFilterGetByKey0':[C.c_void_p,C.POINTER(GUID),C.POINTER(C.POINTER(Filter))], 'FwpmProviderAdd0':[C.c_void_p,C.POINTER(Provider),C.c_void_p], 'FwpmSubLayerAdd0':[C.c_void_p,C.POINTER(Sublayer),C.c_void_p], 'FwpmFilterAdd0':[C.c_void_p,C.POINTER(Filter),C.c_void_p,C.POINTER(C.c_uint64)], 'FwpmFilterDeleteByKey0':[C.c_void_p,C.POINTER(GUID)], 'FwpmSubLayerDeleteByKey0':[C.c_void_p,C.POINTER(GUID)], 'FwpmProviderDeleteByKey0':[C.c_void_p,C.POINTER(GUID)]}.items(): function=getattr(self.api,name);function.argtypes=args;function.restype=U32 def _call(self,name,*args): code=getattr(self.api,name)(*args) if code:raise InstallError('native-operation-failed',code) def app_id(self,executable): result=self.reader.app_id(executable) if not result['ok']:raise InstallError(result['reason'],result['code']) return result['blobHex'] def open(self):self._call('FwpmEngineOpen0',None,10,None,None,C.byref(self.engine)) def prepare_topology(self): from windows_wfp_topology import TopologyObserver self.topology=TopologyObserver() result=self.topology.start() if not result.get('subscribed'):return False snapshot=self.topology.snapshot([L4,L6,R4,R6]) # The current observer cannot attest READ visibility over every object. # Do not substitute successful enumeration for that deployment gate. return bool(snapshot.get('complete') and snapshot.get('allObjectsReadable')) def close(self): if getattr(self,'topology',None):self.topology.close();self.topology=None if self.engine:self.api.FwpmEngineClose0(self.engine);self.engine=C.c_void_p() def begin(self):self._call('FwpmTransactionBegin0',self.engine,0) def commit(self):self._call('FwpmTransactionCommit0',self.engine) def abort(self):self._call('FwpmTransactionAbort0',self.engine) def lookup(self,kind,key): cls,method,not_found={'provider':(Provider,'FwpmProviderGetByKey0',0x80320005),'sublayer':(Sublayer,'FwpmSubLayerGetByKey0',0x80320007),'filter':(Filter,'FwpmFilterGetByKey0',0x80320003)}[kind] pointer=C.POINTER(cls)();guid=_guid(key);code=getattr(self.api,method)(self.engine,C.byref(guid),C.byref(pointer)) if code==not_found:return None if code:raise InstallError('native-lookup-failed',code) try: value=pointer.contents if kind=='filter':return _decode(value) result={'flags':value.flags,'data':_blob(value.data).hex()} if kind=='provider':result['service']=value.service else:result.update(provider=value.provider.contents.text() if value.provider else None,weight=value.weight) return result finally: memory=C.cast(pointer,C.c_void_p);self.api.FwpmFreeMemory0(C.byref(memory)) def add_provider(self): data=(C.c_ubyte*len(MARKER)).from_buffer_copy(MARKER);value=Provider();value.key=_guid(PROVIDER);value.flags=1;value.display=Display('SafetyNet network boundary','Owned browser egress restrictions');value.data=Blob(len(MARKER),data);value.service=SERVICE self._call('FwpmProviderAdd0',self.engine,C.byref(value),None) def add_sublayer(self): data=(C.c_ubyte*len(MARKER)).from_buffer_copy(MARKER);provider=_guid(PROVIDER);value=Sublayer();value.key=_guid(SUBLAYER);value.flags=1;value.display=Display('SafetyNet network boundary','Owned priority; collisions rejected');value.provider=C.pointer(provider);value.data=Blob(len(MARKER),data);value.weight=OWNED_PRIORITY self._call('FwpmSubLayerAdd0',self.engine,C.byref(value),None) def add_filter(self,rule): if rule.get('id') in ('broker-service-receiver','broker-receiver-deny'): from windows_wfp_receiver import validate_receiver_rule validate_receiver_rule(rule) if any(item['type']=='FWP_SECURITY_DESCRIPTOR_TYPE' for item in rule['conditions']): with ExitStack() as stack: from windows_wfp_receiver import service_descriptor descriptors={} for index,item in enumerate(rule['conditions']): if item['type']=='FWP_SECURITY_DESCRIPTOR_TYPE': value=item['value'] if not value.startswith('D:(A;;CC;;;') or not value.endswith(')'):raise InstallError('unsafe-service-descriptor') descriptors[index]=stack.enter_context(service_descriptor(value[len('D:(A;;CC;;;'):-1])) return self._add_filter(rule,descriptors) return self._add_filter(rule,{}) def _add_filter(self,rule,descriptors): conditions=(Condition*len(rule['conditions']))();retained=[] for index,item in enumerate(rule['conditions']): target=conditions[index];target.field=_guid(item['field']);target.match={'equal':0,'not-equal':10}[item['match']] target.value.type={'FWP_UINT8':1,'FWP_UINT16':2,'FWP_UINT32':3,'FWP_BYTE_BLOB_TYPE':12,'FWP_SECURITY_DESCRIPTOR_TYPE':14}[item['type']] if target.value.type==14: blob=descriptors[index];retained.append(blob);target.value.pointer=C.cast(C.pointer(blob),C.c_void_p) continue if target.value.type==12: payload=bytes.fromhex(item['value']);data=(C.c_ubyte*len(payload)).from_buffer_copy(payload);blob=Blob(len(payload),data);retained.extend([data,blob]);target.value.pointer=C.cast(C.pointer(blob),C.c_void_p) else:setattr(target.value,{1:'u8',2:'u16',3:'u32'}[target.value.type],item['value']) provider=_guid(PROVIDER);value=Filter();value.key=_guid(_key(rule['id']));value.flags=1;value.display=Display('SafetyNet '+rule['id'],'Deny only');value.provider=C.pointer(provider);value.layer=_guid(rule['layer']);value.sublayer=_guid(SUBLAYER);value.weight.type=0;value.count=len(conditions);value.conditions=conditions;value.action.type=0x1001 if rule.get('weight') is not None: if rule.get('id') not in ('broker-service-receiver','broker-receiver-deny') or rule['weight'] not in (1,2):raise InstallError('unsafe-receiver-weight') weight=C.c_uint64(rule['weight']);retained.append(weight);value.weight.type=4;value.weight.pointer=C.cast(C.pointer(weight),C.c_void_p) if rule['action']=='FWP_ACTION_PERMIT': if rule.get('id')!='broker-service-receiver' or rule.get('weight')!=2 or not descriptors:raise InstallError('unsafe-permit') value.action.type=0x1002 elif rule['action']!='FWP_ACTION_BLOCK':raise InstallError('unsafe-action') filter_id=C.c_uint64();self._call('FwpmFilterAdd0',self.engine,C.byref(value),None,C.byref(filter_id)) def verify(self,plan):return self.reader.inspect(plan,PROVIDER,SUBLAYER) def delete(self,kind,key): guid=_guid(key);method={'filter':'FwpmFilterDeleteByKey0','sublayer':'FwpmSubLayerDeleteByKey0','provider':'FwpmProviderDeleteByKey0'}[kind] self._call(method,self.engine,C.byref(guid)) class BoundaryInstaller: def __init__(self,backend=None,is_elevated=elevated):self.backend=backend;self.is_elevated=is_elevated def _prepare(self,installation_root,executable,expected_sha256,port,plan): if not self.is_elevated():raise InstallError('administrator-required') raw_root=Path(installation_root).absolute();raw_target=Path(executable).absolute() try: for component in {raw_root,raw_target,*raw_root.parents,*raw_target.parents}: if getattr(component.lstat(),'st_file_attributes',0)&0x400 or component.is_symlink():raise InstallError('installation-reparse-point') root=raw_root.resolve(strict=True);target=raw_target.resolve(strict=True) except OSError:raise InstallError('installation-path-invalid') from None if not root.is_dir() or not target.is_relative_to(root) or target==root or not target.is_file() or target.suffix.lower()!='.exe':raise InstallError('installation-path-invalid') if not isinstance(expected_sha256,str) or len(expected_sha256)!=64 or any(c not in '0123456789abcdef' for c in expected_sha256):raise InstallError('manifest-hash-invalid') digest=hashlib.sha256() with target.open('rb') as stream: for chunk in iter(lambda:stream.read(1024*1024),b''):digest.update(chunk) if digest.hexdigest()!=expected_sha256:raise InstallError('installation-hash-mismatch') backend=self.backend or NativeBackend();identity=backend.app_id(target);_validate(plan,identity,port) return backend def _existing(self,backend,plan): provider=backend.lookup('provider',PROVIDER);sublayer=backend.lookup('sublayer',SUBLAYER) filters=[backend.lookup('filter',_key(rule['id'])) for rule in plan['filters']] if provider is None and sublayer is None and all(item is None for item in filters):return False if not _owned_metadata(provider,'provider') or not _owned_metadata(sublayer,'sublayer') or any(item is None for item in filters):raise InstallError('existing-boundary-collision') result=verify_observed(plan,filters,PROVIDER,SUBLAYER) if not result['verified'] or not backend.verify(plan)['verified']:raise InstallError('existing-boundary-mismatch') return True def apply_managed(self,installation_root,executable,expected_sha256,broker_executable,broker_sha256,port,plan): plan=copy.deepcopy(plan) if plan.get('version')!=3:raise InstallError('managed-plan-required') # Independently bind the receiver app ID to an exact verified installed # broker executable, using the same containment/reparse/hash checks. broker_plan=copy.deepcopy(plan);broker_plan['version']=2;broker_plan['filters']=broker_plan['filters'][:6] broker_plan['applicationId']=plan.get('brokerApplicationId') for rule in broker_plan['filters']:rule['conditions'][0]['value']=broker_plan['applicationId'] self._prepare(installation_root,broker_executable,broker_sha256,port,broker_plan) return self.apply(installation_root,executable,expected_sha256,port,plan,_managed=_MANAGED_TOKEN) def apply(self,installation_root,executable,expected_sha256,port,plan,_managed=None): plan=copy.deepcopy(plan) if plan.get('version')==3 and _managed is not _MANAGED_TOKEN:raise InstallError('receiver-install-requires-verified-broker') backend=self._prepare(installation_root,executable,expected_sha256,port,plan);active=False;backend.open() try: if not backend.prepare_topology():raise InstallError('topology-admission-unavailable') backend.begin();active=True existing=self._existing(backend,plan) if not existing: backend.add_provider();backend.add_sublayer() for rule in plan['filters']:backend.add_filter(rule) # Read back transaction-owned objects before committing. if not _owned_metadata(backend.lookup('provider',PROVIDER),'provider') or not _owned_metadata(backend.lookup('sublayer',SUBLAYER),'sublayer'):raise InstallError('added-metadata-mismatch') observed=[backend.lookup('filter',_key(rule['id'])) for rule in plan['filters']] if any(item is None for item in observed) or not verify_observed(plan,observed,PROVIDER,SUBLAYER)['verified']:raise InstallError('added-filter-mismatch') backend.commit();active=False result=backend.verify(plan) if not result['verified']:raise InstallError('postcommit-verification-failed') return {'installed':True,'existing':existing,'nativeVerified':True,'receiverTestsVerified':False} finally: try: if active:backend.abort() finally:backend.close() def rollback(self,installation_root,executable,expected_sha256,port,plan): plan=copy.deepcopy(plan) backend=self._prepare(installation_root,executable,expected_sha256,port,plan);active=False;backend.open() try: backend.begin();active=True if not self._existing(backend,plan):raise InstallError('owned-boundary-not-present') for rule in plan['filters']:backend.delete('filter',_key(rule['id'])) backend.delete('sublayer',SUBLAYER);backend.delete('provider',PROVIDER) backend.commit();active=False return {'removedOwnedBoundary':True} finally: try: if active:backend.abort() finally:backend.close()