#!/usr/bin/env python3 """Check lexical guard lifetimes, lock order, and waits in Wi-Fi production code. The ownership mutex protects the pure machine. A separate transition mutex serializes decisions with driver/store effects, but must not span connection waits or HTTP shutdown. The profile and ownership locks must never nest. Negative mutations run with the gate so comments or an expired inner guard cannot accidentally satisfy it. This is a source-shape gate, not a C++ parser. """ from __future__ import annotations import pathlib import re import sys REPO = pathlib.Path(__file__).resolve().parents[2] WIFI_MANAGER = REPO / 'firmware/main/wifi_manager.cpp' BLOCKING_CALLS = [r'\bvTaskDelay\b', r'\bxEventGroup\w+', r'\besp_wifi_', r'\b(?:Guard|TransitionGuard|PortalGuard)\s+\w+', r'\bs_store\.', r'\bESP_LOG', r'\bboard_', r'\bhttpd_'] def mask_noncode(code: str) -> str: pattern = re.compile(r'//[^\n]*|/\*.*?\*/|R"(?P[^\s()\\]{0,16})\(.*?\)(?P=delimiter)"|"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'', re.S) return pattern.sub(lambda m: ''.join('\n' if c == '\n' else ' ' for c in m.group()), code) def read_code() -> str: return WIFI_MANAGER.read_text(encoding='utf-8') def brace_end(code: str, start: int) -> int: depth = 0 for i in range(start, len(code)): if code[i] == '{': depth += 1 elif code[i] == '}': depth -= 1 if depth == 0: return i + 1 raise ValueError('unbalanced source braces') def helper_spans(code: str) -> list[tuple[str, int, int]]: code = mask_noncode(code) pattern = re.compile(r'^[ \t]*[\w:<>&*]+\s+(own_\w+)\s*\([^;{}]*\)\s*\{', re.M) return [(m.group(1), m.end()-1, brace_end(code, m.end()-1)) for m in pattern.finditer(code)] def guard_scopes(code: str, kind: str) -> list[tuple[int, int]]: """From declaration to the closing brace of its actual enclosing scope.""" result = [] for m in re.finditer(r'\b'+kind+r'\s+\w+\s*;', code): stack = [] for i, c in enumerate(code[:m.start()]): if c == '{': stack.append(i) elif c == '}' and stack: stack.pop() if stack: result.append((m.start(), brace_end(code, stack[-1]))) return result def check_all_s_own_accesses_are_locked(code: str) -> list[str]: code = mask_noncode(code) spans = helper_spans(code) failures = [] for m in re.finditer(r'\bs_own\b', code): if re.match(r'\s*;', code[m.end():]) and 'ConnectionOwnership' in code[max(0,m.start()-40):m.start()]: continue if not any(start <= m.start() < end for _,start,end in spans): failures.append(f'line {code[:m.start()].count(chr(10))+1}: bare s_own access outside helper') return failures def check_helpers_use_ownguard(code: str) -> list[str]: code = mask_noncode(code) failures = [] for name,start,end in helper_spans(code): body = code[start:end] guards = re.findall(r'\bOwnGuard\s+\w+\s*;',body) if len(guards) != 1 or not re.match(r'\{\s*OwnGuard\s+\w+\s*;',body): failures.append(f'{name}: OwnGuard must be the single outer-scope first statement') if not re.search(r'\bs_own\s*\.',body): failures.append(f'{name}: no ownership member access') return failures def check_no_blocking_under_ownguard(code: str) -> list[str]: code = mask_noncode(code) failures = [] for start,end in guard_scopes(code,'OwnGuard'): for pattern in BLOCKING_CALLS: if re.search(pattern,code[start:end]): failures.append(f'OwnGuard spans forbidden operation {pattern}') # Reverse nesting is equally invalid, even outside an ownership helper. for start,end in guard_scopes(code,'Guard'): if re.search(r'\b(?:OwnGuard|TransitionGuard)\s+\w+|\bown_\w+\s*\(',code[start:end]): failures.append('store Guard spans ownership/transition acquisition') for start,end in guard_scopes(code,'TransitionGuard'): if re.search(r'\bvTaskDelay\b|\bxEventGroupWaitBits\b|\bboard_\w+\s*\(|\bhttpd_\w+\s*\(|\b(?:start_portal|stop_portal|stop_access_point|wait_station_disconnected|validate_candidate)\s*\(',code[start:end]): failures.append('TransitionGuard spans a connection wait or HTTP lifecycle operation') return failures def check_lock_created_before_tasks(code: str) -> list[str]: code = mask_noncode(code) failures = [] for name in ('s_own_lock','s_transition_lock','s_portal_lock'): create = re.search(r'\b'+name+r'\s*=\s*xSemaphoreCreate(?:Recursive)?Mutex\(\)',code) first_task = code.find('xTaskCreate(') if create is None or (first_task != -1 and create.start() > first_task): failures.append(f'{name}: mutex must be created before tasks') return failures def check(code: str) -> list[str]: return sum((fn(code) for fn in (check_all_s_own_accesses_are_locked,check_helpers_use_ownguard, check_no_blocking_under_ownguard,check_lock_created_before_tasks)),[]) def negative_checks(code: str) -> list[str]: original = 'OwnGuard guard;\n return s_own.station_steady();' assert original in code variants = { 'bare access': code+'\nvoid unsafe() { s_own.cancel_owner(); }\n', 'missing guard': code.replace(original,'return s_own.station_steady();',1), 'expired guard': code.replace(original,'{ OwnGuard guard; }\n return s_own.station_steady();',1), 'ownership to store': code.replace(original,'OwnGuard guard;\n Guard store;\n return s_own.station_steady();',1), 'store to ownership': code+'\nvoid unsafe() { Guard store; own_station_steady(); }\n', 'wait under transition': code+'\nvoid unsafe() { TransitionGuard transaction; vTaskDelay(10); }\n', 'comment fake guard': code.replace(original,'// OwnGuard guard;\n return s_own.station_steady();',1), } return [f'negative mutation missed: {name}' for name,mutant in variants.items() if not check(mutant)] def main() -> int: code = read_code() failures = check(code)+negative_checks(code) if failures: for failure in failures: print('FAIL:',failure,file=sys.stderr) return 1 print('Ownership guard lifetimes / both lock orders / transition wait boundaries / 7 negative mutations PASS') return 0 if __name__ == '__main__': raise SystemExit(main())