76 lines
5.7 KiB
Python
76 lines
5.7 KiB
Python
"""Read-only WFP contract and cleanup tests; no filter mutations."""
|
|
import copy
|
|
import ctypes as C
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
sys.path.insert(0,str(Path(__file__).resolve().parent.parent/'apps/network-broker'))
|
|
from windows_wfp_probe import ReadOnlyWfp,verify_observed,Filter
|
|
PROVIDER='2a7bcbf2-e4f5-4e94-bd33-d5442a3bef50'
|
|
SUBLAYER='403a69b8-fc7c-460a-a247-72e0d72e6bf8'
|
|
def contract():
|
|
expected=[];observed=[]
|
|
for index in range(6):
|
|
conditions=[{'field':'d78e1e87-8644-4ea5-9437-d809ecefc971','match':'equal','type':'FWP_BYTE_BLOB_TYPE','value':'01020304'}, {'field':'3971ef2b-623e-4f9a-8cb1-6e79b806b9a7','match':'equal','type':'FWP_UINT8','value':index}]
|
|
expected.append({'layer':'layer-'+str(index),'action':'FWP_ACTION_BLOCK','clearActionRight':False,'conditions':conditions})
|
|
observed.append({'provider':PROVIDER,'sublayer':SUBLAYER,'layer':'layer-'+str(index),'flags':1,'action':0x1001,'conditions':copy.deepcopy(conditions)})
|
|
return {'filters':expected},observed
|
|
class WfpProbeTests(unittest.TestCase):
|
|
def test_exact_contract_and_reordered_conditions(self):
|
|
plan,items=contract();items.reverse()
|
|
for item in items:item['conditions'].reverse()
|
|
self.assertTrue(verify_observed(plan,items,PROVIDER,SUBLAYER)['verified'])
|
|
def test_missing_duplicate_or_extra_filters(self):
|
|
plan,items=contract()
|
|
for bad in [items[:-1],items+items[:1],items[:3]+items[:1]]:
|
|
self.assertFalse(verify_observed(plan,bad,PROVIDER,SUBLAYER)['verified'])
|
|
def test_blob_type_match_layer_and_numeric_mismatch(self):
|
|
plan,items=contract()
|
|
for key,value in [('type','FWP_UINT32'),('value','deadbeef'),('match','not-equal')]:
|
|
bad=copy.deepcopy(items);bad[0]['conditions'][0][key]=value
|
|
self.assertEqual(verify_observed(plan,bad,PROVIDER,SUBLAYER)['reason'],'filter-condition-mismatch')
|
|
bad=copy.deepcopy(items);bad[0]['layer']='other-layer';self.assertFalse(verify_observed(plan,bad,PROVIDER,SUBLAYER)['verified'])
|
|
def test_disabled_nonpersistent_permit_softblock_or_wrong_owner(self):
|
|
plan,items=contract()
|
|
for key,value in [('flags',0),('flags',33),('flags',9),('action',0x1002),('provider','11111111-1111-1111-1111-111111111111'),('sublayer','22222222-2222-2222-2222-222222222222')]:
|
|
bad=copy.deepcopy(items);bad[0][key]=value
|
|
self.assertEqual(verify_observed(plan,bad,PROVIDER,SUBLAYER)['reason'],'filter-metadata-mismatch')
|
|
def test_engine_denial_is_unverified_and_not_closed_or_enumerated(self):
|
|
probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock();probe.api.FwpmEngineOpen0.return_value=5
|
|
self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER),{'verified':False,'reason':'engine-unavailable','code':5})
|
|
probe.api.FwpmFilterCreateEnumHandle0.assert_not_called();probe.api.FwpmEngineClose0.assert_not_called()
|
|
def test_enum_handle_denial_closes_opened_engine(self):
|
|
probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
|
|
def opened(_name,_auth,_identity,_session,handle):handle._obj.value=123;return 0
|
|
probe.api.FwpmEngineOpen0.side_effect=opened;probe.api.FwpmFilterCreateEnumHandle0.return_value=5
|
|
self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['code'],5)
|
|
probe.api.FwpmEngineClose0.assert_called_once();probe.api.FwpmFilterDestroyEnumHandle0.assert_not_called()
|
|
def test_empty_enumeration_cleans_handle_and_engine(self):
|
|
probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
|
|
def opened(_name,_auth,_identity,_session,handle):handle._obj.value=123;return 0
|
|
def enumeration(_engine,_template,handle):handle._obj.value=456;return 0
|
|
def empty(_engine,_handle,_limit,_entries,count):count._obj.value=0;return 0
|
|
probe.api.FwpmEngineOpen0.side_effect=opened;probe.api.FwpmFilterCreateEnumHandle0.side_effect=enumeration;probe.api.FwpmFilterEnum0.side_effect=empty
|
|
self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'filter-count-mismatch')
|
|
probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
|
|
def test_oversized_native_page_frees_memory_and_closes_handles(self):
|
|
probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
|
|
probe.api.FwpmEngineOpen0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',123) or 0)
|
|
probe.api.FwpmFilterCreateEnumHandle0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',456) or 0)
|
|
storage=(C.POINTER(Filter)*129)()
|
|
def oversized(_engine,_handle,_limit,entries,count):
|
|
C.cast(entries,C.POINTER(C.POINTER(C.POINTER(Filter))))[0]=C.cast(storage,C.POINTER(C.POINTER(Filter)))
|
|
count._obj.value=129;return 0
|
|
probe.api.FwpmFilterEnum0.side_effect=oversized
|
|
self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'enumeration-limit')
|
|
probe.api.FwpmFreeMemory0.assert_called_once();probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
|
|
def test_enumeration_error_cleans_handles(self):
|
|
probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
|
|
probe.api.FwpmEngineOpen0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',123) or 0)
|
|
probe.api.FwpmFilterCreateEnumHandle0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',456) or 0)
|
|
probe.api.FwpmFilterEnum0.return_value=5
|
|
self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'enumeration-failed')
|
|
probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
|
|
if __name__=='__main__':unittest.main()
|