59 lines
3.2 KiB
Python
59 lines
3.2 KiB
Python
|
import ctypes as C
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps/network-broker'))
|
||
|
|
from windows_wfp_receiver import service_sddl, service_descriptor
|
||
|
|
from windows_wfp_install import NativeBackend
|
||
|
|
from windows_wfp_probe import Filter, Blob
|
||
|
|
|
||
|
|
class ReceiverTests(unittest.TestCase):
|
||
|
|
def test_service_sid_only_and_minimum_match_right(self):
|
||
|
|
self.assertEqual(service_sddl('S-1-5-80-1-2-3-4-5'), 'D:(A;;CC;;;S-1-5-80-1-2-3-4-5)')
|
||
|
|
for sid in ['S-1-5-19', 'S-1-5-80-01-2-3-4-5', 'S-1-5-80-4294967296-2-3-4-5']:
|
||
|
|
with self.assertRaises(Exception): service_sddl(sid)
|
||
|
|
|
||
|
|
@unittest.skipUnless(os.name == 'nt', 'Native Windows SD validation')
|
||
|
|
def test_actual_native_self_relative_sd_and_exact_ace_mask(self):
|
||
|
|
api = C.WinDLL('advapi32')
|
||
|
|
api.IsValidSecurityDescriptor.argtypes = [C.c_void_p]
|
||
|
|
api.IsValidSecurityDescriptor.restype = C.c_int
|
||
|
|
with service_descriptor('S-1-5-80-1-2-3-4-5') as blob:
|
||
|
|
self.assertTrue(api.IsValidSecurityDescriptor(blob.data))
|
||
|
|
raw = C.string_at(blob.data, blob.size)
|
||
|
|
self.assertTrue(int.from_bytes(raw[2:4], 'little') & 0x8000)
|
||
|
|
dacl = int.from_bytes(raw[16:20], 'little')
|
||
|
|
self.assertEqual(int.from_bytes(raw[dacl+4:dacl+6], 'little'), 1)
|
||
|
|
self.assertEqual(int.from_bytes(raw[dacl+12:dacl+16], 'little'), 1)
|
||
|
|
|
||
|
|
@unittest.skipUnless(os.name == 'nt', 'Native Windows SD marshalling')
|
||
|
|
def test_actual_sd_lifetime_and_uint64_soft_permit_marshalling_without_add(self):
|
||
|
|
backend = NativeBackend.__new__(NativeBackend)
|
||
|
|
backend.engine = C.c_void_p(1)
|
||
|
|
captured = []
|
||
|
|
def capture(name, _engine, pointer, _security, _id):
|
||
|
|
self.assertEqual(name, 'FwpmFilterAdd0')
|
||
|
|
value = C.cast(pointer, C.POINTER(Filter)).contents
|
||
|
|
self.assertEqual(value.flags, 1) # PERSISTENT only; no clear-action.
|
||
|
|
self.assertEqual(value.action.type, 0x1002)
|
||
|
|
self.assertEqual(value.weight.type, 4)
|
||
|
|
self.assertEqual(C.cast(value.weight.pointer, C.POINTER(C.c_uint64)).contents.value, 2)
|
||
|
|
self.assertEqual(value.count, 5)
|
||
|
|
condition = value.conditions[4]
|
||
|
|
self.assertEqual(condition.value.type, 14)
|
||
|
|
blob = C.cast(condition.value.pointer, C.POINTER(Blob)).contents
|
||
|
|
captured.append(C.string_at(blob.data, blob.size))
|
||
|
|
backend._call = capture
|
||
|
|
script = "const {planServiceReceiver}=require('./apps/network-broker/wfp-topology.cjs');console.log(JSON.stringify(planServiceReceiver({appId:Buffer.from(process.argv[1],'hex'),brokerPort:43120,serviceSid:'S-1-5-80-1-2-3-4-5'}).filters[0]));"
|
||
|
|
identity = '\\device\\harddiskvolume3\\safetynet\\node.exe\0'.encode('utf-16le').hex()
|
||
|
|
result = subprocess.run(['node','-e',script,identity],cwd=Path(__file__).resolve().parents[1],capture_output=True,text=True,check=True,timeout=10)
|
||
|
|
rule = json.loads(result.stdout)
|
||
|
|
backend.add_filter(rule)
|
||
|
|
rule['conditions'] = rule['conditions'][-1:]
|
||
|
|
with self.assertRaises(Exception): backend.add_filter(rule)
|
||
|
|
self.assertEqual(len(captured), 1)
|
||
|
|
|
||
|
|
if __name__ == '__main__': unittest.main()
|