127 lines
8.0 KiB
Python
127 lines
8.0 KiB
Python
"""Read-only native Windows WFP observations; never adds/deletes filters."""
|
|
import ctypes as C
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
U32=C.c_uint32
|
|
class GUID(C.Structure):
|
|
_fields_=[('data1',U32),('data2',C.c_uint16),('data3',C.c_uint16),('data4',C.c_ubyte*8)]
|
|
def text(self): return str(uuid.UUID(bytes_le=bytes(self)))
|
|
class Blob(C.Structure):
|
|
_fields_=[('size',U32),('data',C.POINTER(C.c_ubyte))]
|
|
class Data(C.Union):
|
|
_fields_=[('u8',C.c_uint8),('u16',C.c_uint16),('u32',U32),('pointer',C.c_void_p)]
|
|
class Value(C.Structure):
|
|
_anonymous_=['data']
|
|
_fields_=[('type',U32),('data',Data)]
|
|
class Condition(C.Structure):
|
|
_fields_=[('field',GUID),('match',U32),('value',Value)]
|
|
class Display(C.Structure):
|
|
_fields_=[('name',C.c_wchar_p),('description',C.c_wchar_p)]
|
|
class Action(C.Structure):
|
|
_fields_=[('type',U32),('key',GUID)]
|
|
class Context(C.Union):
|
|
_fields_=[('raw',C.c_uint64),('provider',GUID)]
|
|
class Filter(C.Structure):
|
|
_fields_=[('key',GUID),('display',Display),('flags',U32),('provider',C.POINTER(GUID)),('provider_data',Blob),('layer',GUID),('sublayer',GUID),('weight',Value),('count',U32),('conditions',C.POINTER(Condition)),('action',Action),('context',Context),('reserved',C.POINTER(GUID)),('id',C.c_uint64),('effective_weight',Value)]
|
|
|
|
TYPES={1:'FWP_UINT8',2:'FWP_UINT16',3:'FWP_UINT32',12:'FWP_BYTE_BLOB_TYPE',14:'FWP_SECURITY_DESCRIPTOR_TYPE'}
|
|
MATCHES={0:'equal',10:'not-equal'}
|
|
def _blob(blob):
|
|
if not blob.data or not 0<blob.size<=65536:raise ValueError('Invalid bounded blob')
|
|
return C.string_at(blob.data,blob.size)
|
|
def _decode(native):
|
|
if native.count>16 or (native.count and not native.conditions):raise ValueError('Invalid bounded conditions')
|
|
conditions=[]
|
|
for index in range(native.count):
|
|
condition=native.conditions[index];value=condition.value
|
|
if value.type==1: parsed=value.u8
|
|
elif value.type==2: parsed=value.u16
|
|
elif value.type==3: parsed=value.u32
|
|
elif value.type in (12,14):
|
|
if not value.pointer:raise ValueError('Missing blob')
|
|
parsed=_blob(C.cast(value.pointer,C.POINTER(Blob)).contents).hex()
|
|
else:raise ValueError('Unsupported condition type')
|
|
conditions.append({'field':condition.field.text(),'match':MATCHES.get(condition.match,'unsupported'),'type':TYPES[value.type],'value':parsed})
|
|
weight=C.cast(native.weight.pointer,C.POINTER(C.c_uint64)).contents.value if native.weight.type==4 and native.weight.pointer else None
|
|
return {'key':native.key.text(),'provider':native.provider.contents.text() if native.provider else None,'sublayer':native.sublayer.text(),'layer':native.layer.text(),'flags':native.flags,'action':native.action.type,'conditions':conditions,'weight':weight}
|
|
|
|
def verify_observed(plan,observed,provider,sublayer):
|
|
"""Compare complete native observations to trusted deployment metadata."""
|
|
provider=str(uuid.UUID(provider));sublayer=str(uuid.UUID(sublayer))
|
|
selected=[item for item in observed if item.get('provider')==provider or item.get('sublayer')==sublayer]
|
|
expected=plan.get('filters',[])
|
|
if len(expected) not in (6,8) or len(selected)!=len(expected):return {'verified':False,'reason':'filter-count-mismatch'}
|
|
def signature(conditions):return sorted((c['field'],c['match'],c['type'],str(c['value'])) for c in conditions)
|
|
pending=list(expected)
|
|
for item in selected:
|
|
if item['provider']!=provider or item['sublayer']!=sublayer or not item['flags']&1 or item['flags']&(8|32) or (len(expected)==6 and item['action']!=0x1001):return {'verified':False,'reason':'filter-metadata-mismatch'}
|
|
def expected_conditions(rule):
|
|
result=[]
|
|
for condition in rule['conditions']:
|
|
condition=dict(condition)
|
|
if condition['type']=='FWP_SECURITY_DESCRIPTOR_TYPE':
|
|
from windows_wfp_receiver import service_descriptor
|
|
with service_descriptor(condition['value'][len('D:(A;;CC;;;'):-1]) as blob:condition['value']=_blob(blob).hex()
|
|
result.append(condition)
|
|
return result
|
|
matches=[rule for rule in pending if item['action']=={'FWP_ACTION_BLOCK':0x1001,'FWP_ACTION_PERMIT':0x1002}.get(rule['action']) and rule['clearActionRight'] is False and rule['layer']==item['layer'] and item.get('weight')==rule.get('weight') and signature(expected_conditions(rule))==signature(item['conditions'])]
|
|
if len(matches)!=1:return {'verified':False,'reason':'filter-condition-mismatch'}
|
|
pending.remove(matches[0])
|
|
return {'verified':not pending,'reason':'verified' if not pending else 'filter-missing'}
|
|
|
|
class ReadOnlyWfp:
|
|
def __init__(self):
|
|
if os.name!='nt':raise RuntimeError('Windows required')
|
|
self.api=C.WinDLL('fwpuclnt',use_last_error=True)
|
|
definitions={
|
|
'FwpmEngineOpen0':([C.c_wchar_p,U32,C.c_void_p,C.c_void_p,C.POINTER(C.c_void_p)],U32),
|
|
'FwpmEngineClose0':([C.c_void_p],U32),
|
|
'FwpmGetAppIdFromFileName0':([C.c_wchar_p,C.POINTER(C.POINTER(Blob))],U32),
|
|
'FwpmFreeMemory0':([C.POINTER(C.c_void_p)],None),
|
|
'FwpmFilterCreateEnumHandle0':([C.c_void_p,C.c_void_p,C.POINTER(C.c_void_p)],U32),
|
|
'FwpmFilterDestroyEnumHandle0':([C.c_void_p,C.c_void_p],U32),
|
|
'FwpmFilterEnum0':([C.c_void_p,C.c_void_p,U32,C.POINTER(C.POINTER(C.POINTER(Filter))),C.POINTER(U32)],U32)}
|
|
for name,(arguments,result) in definitions.items():
|
|
method=getattr(self.api,name);method.argtypes=arguments;method.restype=result
|
|
def app_id(self,executable):
|
|
target=Path(executable).resolve(strict=True)
|
|
if not target.is_file() or target.suffix.lower()!='.exe':raise ValueError('Executable required')
|
|
pointer=C.POINTER(Blob)();code=self.api.FwpmGetAppIdFromFileName0(str(target),C.byref(pointer))
|
|
if code:return {'ok':False,'reason':'app-id-unavailable','code':int(code)}
|
|
try:
|
|
data=_blob(pointer.contents)
|
|
return {'ok':True,'blobHex':data.hex()}
|
|
finally:
|
|
memory=C.cast(pointer,C.c_void_p);self.api.FwpmFreeMemory0(C.byref(memory))
|
|
def inspect(self,plan,provider,sublayer,max_entries=4096):
|
|
provider=str(uuid.UUID(provider));sublayer=str(uuid.UUID(sublayer))
|
|
if not isinstance(max_entries,int) or not 6<=max_entries<=4096:raise ValueError('Invalid enumeration bound')
|
|
engine=C.c_void_p();code=self.api.FwpmEngineOpen0(None,10,None,None,C.byref(engine))
|
|
if code:return {'verified':False,'reason':'engine-unavailable','code':int(code)}
|
|
enumeration=C.c_void_p()
|
|
try:
|
|
code=self.api.FwpmFilterCreateEnumHandle0(engine,None,C.byref(enumeration))
|
|
if code:return {'verified':False,'reason':'enumeration-unavailable','code':int(code)}
|
|
selected=[];total=0
|
|
while True:
|
|
entries=C.POINTER(C.POINTER(Filter))();count=U32()
|
|
code=self.api.FwpmFilterEnum0(engine,enumeration,128,C.byref(entries),C.byref(count))
|
|
if code:return {'verified':False,'reason':'enumeration-failed','code':int(code)}
|
|
try:
|
|
if count.value>128 or total+count.value>max_entries:return {'verified':False,'reason':'enumeration-limit'}
|
|
total+=count.value
|
|
for index in range(count.value):
|
|
native=entries[index].contents
|
|
if (native.provider and native.provider.contents.text()==provider) or native.sublayer.text()==sublayer:selected.append(_decode(native))
|
|
if not count.value:break
|
|
finally:
|
|
if entries:
|
|
memory=C.cast(entries,C.c_void_p);self.api.FwpmFreeMemory0(C.byref(memory))
|
|
return verify_observed(plan,selected,provider,sublayer)
|
|
except (ValueError,KeyError,TypeError):return {'verified':False,'reason':'native-contract-invalid'}
|
|
finally:
|
|
if enumeration:self.api.FwpmFilterDestroyEnumHandle0(engine,enumeration)
|
|
self.api.FwpmEngineClose0(engine)
|