56 lines
3.3 KiB
Python
56 lines
3.3 KiB
Python
"""Exact service-SID receiver filter marshalling; installer integration pending.
|
|
|
|
No CLI or activation. Only FWP_ACTION_PERMIT without CLEAR_ACTION_RIGHT is used
|
|
for the exact service identity, followed by a lower-weight tuple BLOCK.
|
|
"""
|
|
import ctypes as C
|
|
import re
|
|
from contextlib import contextmanager
|
|
from windows_wfp_install import InstallError
|
|
from windows_wfp_probe import Blob
|
|
|
|
def validate_receiver_rule(rule):
|
|
expected = [('3971ef2b-623e-4f9a-8cb1-6e79b806b9a7','FWP_UINT8',6),
|
|
('d9ee00de-c1ef-4617-bfe3-ffd8f5a08957','FWP_UINT32',0x7f000001)]
|
|
conditions = rule.get('conditions', [])
|
|
permit = rule.get('id') == 'broker-service-receiver'
|
|
if rule.get('layer') != 'e1cd9fe7-f4b5-4273-96c0-592e487b8650' or rule.get('clearActionRight') is not False or len(conditions) != (5 if permit else 3):
|
|
raise InstallError('unsafe-receiver-plan')
|
|
for item, (field, type_, value) in zip(conditions, expected):
|
|
if item != {'field':field,'match':'equal','type':type_,'value':value}:raise InstallError('unsafe-receiver-plan')
|
|
port = conditions[2]
|
|
if port.get('field') != '0c1ba1af-5765-453f-af22-a8f791ac775b' or port.get('match') != 'equal' or port.get('type') != 'FWP_UINT16' or not isinstance(port.get('value'),int) or not 1 <= port['value'] <= 65535:
|
|
raise InstallError('unsafe-receiver-plan')
|
|
if rule.get('action') != ('FWP_ACTION_PERMIT' if permit else 'FWP_ACTION_BLOCK') or rule.get('weight') != (2 if permit else 1):raise InstallError('unsafe-receiver-plan')
|
|
if permit:
|
|
app, sd = conditions[3:]
|
|
if app.get('field') != 'd78e1e87-8644-4ea5-9437-d809ecefc971' or app.get('type') != 'FWP_BYTE_BLOB_TYPE' or app.get('match') != 'equal' or not isinstance(app.get('value'),str) or not re.fullmatch('[0-9a-f]{16,131072}',app['value']):raise InstallError('unsafe-receiver-plan')
|
|
if sd.get('field') != 'af043a0a-b34d-4f86-979c-c90371af6e66' or sd.get('type') != 'FWP_SECURITY_DESCRIPTOR_TYPE' or sd.get('match') != 'equal':raise InstallError('unsafe-receiver-plan')
|
|
|
|
def service_sddl(sid):
|
|
if not isinstance(sid, str) or not re.fullmatch(r'S-1-5-80-(?:\d+-){4}\d+', sid):
|
|
raise InstallError('invalid-service-sid')
|
|
if any(str(int(part)) != part or int(part) > 4294967295 for part in sid.split('-')[4:]):
|
|
raise InstallError('invalid-service-sid')
|
|
# FWP_ACTRL_MATCH_FILTER=1, represented by CC in SDDL. Not GENERIC_ALL.
|
|
return 'D:(A;;CC;;;' + sid + ')'
|
|
|
|
@contextmanager
|
|
def service_descriptor(sid):
|
|
advapi = C.WinDLL('advapi32', use_last_error=True)
|
|
kernel = C.WinDLL('kernel32', use_last_error=True)
|
|
convert = advapi.ConvertStringSecurityDescriptorToSecurityDescriptorW
|
|
convert.argtypes = [C.c_wchar_p, C.c_uint32, C.POINTER(C.c_void_p), C.POINTER(C.c_uint32)]
|
|
convert.restype = C.c_int
|
|
kernel.LocalFree.argtypes = [C.c_void_p]
|
|
kernel.LocalFree.restype = C.c_void_p
|
|
pointer, size = C.c_void_p(), C.c_uint32()
|
|
if not convert(service_sddl(sid), 1, C.byref(pointer), C.byref(size)):
|
|
raise InstallError('service-descriptor-invalid', C.get_last_error())
|
|
try:
|
|
if not 20 <= size.value <= 4096:
|
|
raise InstallError('service-descriptor-size')
|
|
yield Blob(size.value, C.cast(pointer, C.POINTER(C.c_ubyte)))
|
|
finally:
|
|
kernel.LocalFree(pointer)
|