242 lines
17 KiB
Python
242 lines
17 KiB
Python
"""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()
|