"""Read-only BFE topology/notifications. No installer or packet-enforcement claim. Notifications exclude inaccessible objects and are asynchronous. Successful subscription does NOT establish atomic protection against privileged changes. """ import ctypes as C import threading from windows_wfp_probe import GUID, Display, Blob, Filter, U32, ReadOnlyWfp class Sublayer(C.Structure): _fields_ = [('key', GUID), ('display', Display), ('flags', U32), ('provider', C.POINTER(GUID)), ('data', Blob), ('weight', C.c_uint16)] class Subscription(C.Structure): _fields_ = [('template', C.c_void_p), ('flags', U32), ('session', GUID)] CALLBACK = getattr(C, 'WINFUNCTYPE', C.CFUNCTYPE)(None, C.c_void_p, C.c_void_p) def bfe_status(): """SCM_QUERY_STATUS only; never start/stop/change service configuration.""" api = C.WinDLL('advapi32', use_last_error=True) api.OpenSCManagerW.argtypes = [C.c_wchar_p, C.c_wchar_p, U32] api.OpenSCManagerW.restype = C.c_void_p api.OpenServiceW.argtypes = [C.c_void_p, C.c_wchar_p, U32] api.OpenServiceW.restype = C.c_void_p api.CloseServiceHandle.argtypes = [C.c_void_p] api.QueryServiceStatusEx.argtypes = [C.c_void_p, U32, C.c_void_p, U32, C.POINTER(U32)] manager = api.OpenSCManagerW(None, None, 1) if not manager:return {'running':False,'code':C.get_last_error()} service = None try: service = api.OpenServiceW(manager, 'BFE', 4) if not service:return {'running':False,'code':C.get_last_error()} status = (U32 * 9)() needed = U32() if not api.QueryServiceStatusEx(service, 0, status, C.sizeof(status), C.byref(needed)): return {'running':False,'code':C.get_last_error()} return {'running':status[1] == 4,'state':int(status[1])} finally: if service:api.CloseServiceHandle(service) api.CloseServiceHandle(manager) class TopologyObserver: def __init__(self): self.reader = ReadOnlyWfp() self.api = self.reader.api self.engine = C.c_void_p() self.handles = [] self.changed = threading.Event() self.callback = CALLBACK(lambda _context, _change: self.changed.set()) for kind in ('Filter', 'SubLayer'): subscribe = getattr(self.api, 'Fwpm' + kind + 'SubscribeChanges0') subscribe.argtypes = [C.c_void_p, C.POINTER(Subscription), CALLBACK, C.c_void_p, C.POINTER(C.c_void_p)] subscribe.restype = U32 unsubscribe = getattr(self.api, 'Fwpm' + kind + 'UnsubscribeChanges0') unsubscribe.argtypes = [C.c_void_p, C.c_void_p] unsubscribe.restype = U32 self.api.FwpmSubLayerGetByKey0.argtypes = [C.c_void_p, C.POINTER(GUID), C.POINTER(C.POINTER(Sublayer))] self.api.FwpmSubLayerGetByKey0.restype = U32 def start(self): if self.engine.value: raise RuntimeError('Observer already started') code = self.api.FwpmEngineOpen0(None, 10, None, None, C.byref(self.engine)) if code: self.changed.set() return {'subscribed': False, 'reason': 'engine-unavailable', 'code': int(code)} for kind in ('Filter', 'SubLayer'): handle = C.c_void_p() subscription = Subscription() # FWPM_SUBSCRIPTION_FLAG_NOTIFY_ON_ADD | NOTIFY_ON_DELETE. subscription.flags = 3 code = getattr(self.api, 'Fwpm' + kind + 'SubscribeChanges0')( self.engine, C.byref(subscription), self.callback, None, C.byref(handle)) if code: self.changed.set() self.close() return {'subscribed': False, 'reason': 'subscription-unavailable', 'code': int(code)} self.handles.append((kind, handle)) return {'subscribed': True, 'allObjectsReadable': False, 'zeroChangeWindowProven': False} def snapshot(self, relevant_layers, limit=4096): if not self.engine.value or self.changed.is_set(): return {'complete': False, 'reason': 'observer-unavailable'} if not isinstance(limit, int) or not 1 <= limit <= 65536: raise ValueError('Invalid snapshot bound') enum = C.c_void_p() layers = set(relevant_layers) filters, sublayers, total = [], {}, 0 try: code = self.api.FwpmFilterCreateEnumHandle0(self.engine, None, C.byref(enum)) if code: return {'complete': False, 'reason': 'enumeration-unavailable', 'code': int(code)} while True: entries = C.POINTER(C.POINTER(Filter))() count = U32() code = self.api.FwpmFilterEnum0(self.engine, enum, 128, C.byref(entries), C.byref(count)) if code: return {'complete': False, 'reason': 'enumeration-failed', 'code': int(code)} try: total += count.value if count.value > 128 or total > limit: return {'complete': False, 'reason': 'enumeration-limit'} for index in range(count.value): value = entries[index].contents if value.layer.text() not in layers: continue key = value.sublayer.text() if key not in sublayers: pointer = C.POINTER(Sublayer)() code = self.api.FwpmSubLayerGetByKey0(self.engine, C.byref(value.sublayer), C.byref(pointer)) if code: return {'complete': False, 'reason': 'sublayer-unavailable', 'code': int(code)} try: sublayers[key] = {'key': key, 'weight': int(pointer.contents.weight)} finally: memory = C.cast(pointer, C.c_void_p) self.api.FwpmFreeMemory0(C.byref(memory)) filters.append({'key':value.key.text(),'provider':value.provider.contents.text() if value.provider else None, 'layer': value.layer.text(), 'sublayer': key, 'action': int(value.action.type), 'flags': int(value.flags)}) if not count.value: break finally: if entries: memory = C.cast(entries, C.c_void_p) self.api.FwpmFreeMemory0(C.byref(memory)) return {'complete': not self.changed.is_set(), 'allObjectsReadable': False, 'filters': filters, 'sublayers': list(sublayers.values()), 'zeroChangeWindowProven': False} finally: if enum: self.api.FwpmFilterDestroyEnumHandle0(self.engine, enum) def revalidate(self, relevant_layers): # Notifications are unreliable. Every release gate must additionally # revalidate; a service caller can periodically invoke this method. # No result permits an automatic recovery after previous invalidation. if self.changed.is_set():return {'complete':False,'reason':'observer-invalidated'} if not bfe_status().get('running'): self.changed.set() return {'complete':False,'reason':'bfe-unavailable'} result = self.snapshot(relevant_layers) if not result.get('complete'):self.changed.set() return result def authorized_snapshot(self, trusted_plan, provider, sublayer, relevant_layers): """Exact protected-plan readback, never a caller-selected visibility flag. Integration must derive trusted_plan from protected signed installation and deployment material. No public CLI or untrusted JSON plan loader. """ result = self.revalidate(relevant_layers) if not result.get('complete'):return result verified = self.reader.inspect(trusted_plan, provider, sublayer) if not verified.get('verified') or self.changed.is_set(): self.changed.set() return {'complete':False,'reason':'owned-readback-unavailable'} selected=[row for row in result['filters'] if row['sublayer']==sublayer or row['provider']==provider] if len(selected)!=8 or any(row['sublayer']!=sublayer or row['provider']!=provider for row in selected): self.changed.set() return {'complete':False,'reason':'owned-readback-count'} for row in selected:row['ownedVerified']=True result.update(bfeRunning=True,ownedSublayer=sublayer,enumerationComplete=True) # Remains false until a separate, authoritative protected deployment # prerequisite proves visibility. EOF and token elevation cannot do so. return result def close(self): self.changed.set() for kind, handle in reversed(self.handles): getattr(self.api, 'Fwpm' + kind + 'UnsubscribeChanges0')(self.engine, handle) self.handles.clear() if self.engine.value: self.api.FwpmEngineClose0(self.engine) self.engine = C.c_void_p()