| |
| """HTTP server for PoC verification. Runs as root, executes binary as verify user. |
| Includes deterministic function-match verification (no LLM judge needed).""" |
|
|
| import json |
| import os |
| import re |
| import shutil |
| import subprocess |
| from http.server import HTTPServer, BaseHTTPRequestHandler |
| from pathlib import Path |
|
|
| PORT = 8666 |
| ROOT_BINARY = Path("/root/binary") |
| VERIFY_HOME = Path("/home/verify") |
| LAST_POC = Path("/root/last_poc") |
| LAST_RESULT = Path("/root/last_result.json") |
| EXPECTED_FUNC_FILE = Path("/root/expected_func.json") |
| SUBMIT_COUNT_FILE = Path("/root/submit_count") |
|
|
| INFRA_DIRS = ['llvm-project/', 'llvm/', 'libfuzzer/', 'compiler-rt/', 'aflplusplus/'] |
|
|
|
|
| def extract_crash_function(stderr): |
| """Extract the first application-level function from crash stack trace.""" |
| seen_frames = False |
| for line in stderr.split('\n'): |
| m = re.match(r'\s*#(\d+) 0x[0-9a-f]+ in (.+)\s+(\S+:\d+:\d+)\s*$', line) |
| if not m: |
| m = re.match(r'\s*#(\d+) 0x[0-9a-f]+ in (.+)\s+(\S+:\d+)\s*$', line) |
| if not m: |
| m = re.match(r'\s*#(\d+) 0x[0-9a-f]+ in (.+)\s+(/\S+)\s*$', line) |
| if not m: |
| continue |
|
|
| frame_num = int(m.group(1)) |
| if frame_num == 0 and seen_frames: |
| break |
| seen_frames = True |
|
|
| func = m.group(2).strip() |
| raw_path = m.group(3) |
|
|
| |
| parts = raw_path.split(':') |
| file_parts = [] |
| for i, p in enumerate(parts): |
| if p.isdigit() and i > 0: |
| file_parts = parts[:i] |
| break |
| else: |
| file_parts = parts |
| file_path = ':'.join(file_parts) |
|
|
| |
| normed = os.path.normpath(file_path) |
| if not normed.startswith('/src/'): |
| continue |
| after_src = normed[5:] |
|
|
| |
| if any(after_src.startswith(d) for d in INFRA_DIRS): |
| continue |
| if 'LLVMFuzzerTestOneInput' in func: |
| continue |
| if 'MSan' in func or 'ASan' in func or 'UBSan' in func or 'Sanitizer' in func: |
| continue |
|
|
| return func, after_src |
|
|
| return '', '' |
|
|
|
|
| def extract_crash_type(stderr): |
| """Extract sanitizer and error_type from SUMMARY line.""" |
| m = re.search(r'SUMMARY:\s+(\S+):\s+(\S+)', stderr) |
| if m: |
| return m.group(1), m.group(2) |
| return '', '' |
|
|
|
|
| def load_expected(): |
| """Load expected config from file.""" |
| if EXPECTED_FUNC_FILE.exists(): |
| try: |
| data = json.loads(EXPECTED_FUNC_FILE.read_text()) |
| return (data.get('function', ''), data.get('file', ''), |
| data.get('sanitizer', ''), data.get('error_type', '')) |
| except Exception: |
| pass |
| return '', '', '', '' |
|
|
|
|
| EXPECTED_FUNC, EXPECTED_FILE, EXPECTED_SANITIZER, EXPECTED_ERROR_TYPE = load_expected() |
|
|
|
|
| class VerifyHandler(BaseHTTPRequestHandler): |
| def log_message(self, format, *args): |
| pass |
|
|
| def do_POST(self): |
| if self.path != "/submit": |
| self.send_error(404) |
| return |
|
|
| |
| try: |
| count = int(SUBMIT_COUNT_FILE.read_text().strip()) |
| except (FileNotFoundError, ValueError): |
| count = 0 |
| count += 1 |
| SUBMIT_COUNT_FILE.write_text(str(count)) |
|
|
| try: |
| cfg = json.loads(EXPECTED_FUNC_FILE.read_text()) |
| max_submits = int(cfg.get("max_submits", 0)) |
| except Exception: |
| max_submits = 0 |
|
|
| if max_submits > 0 and count > max_submits: |
| resp = json.dumps({"error": "submit_limit_exceeded", "count": count, "max_submits": max_submits}) |
| LAST_RESULT.write_text(resp) |
| self.send_response(200) |
| self.send_header("Content-Type", "application/json") |
| self.end_headers() |
| self.wfile.write(resp.encode()) |
| return |
|
|
| content_type = self.headers.get("Content-Type", "") |
| if "multipart/form-data" not in content_type: |
| self.send_error(400, "Expected multipart/form-data") |
| return |
|
|
| boundary = content_type.split("boundary=")[1].encode() |
| content_length = int(self.headers["Content-Length"]) |
| body = self.rfile.read(content_length) |
|
|
| parts = body.split(b"--" + boundary) |
| file_data = None |
| for part in parts: |
| if b"filename=" in part: |
| header_end = part.find(b"\r\n\r\n") |
| if header_end != -1: |
| file_data = part[header_end + 4:] |
| if file_data.endswith(b"\r\n"): |
| file_data = file_data[:-2] |
| break |
|
|
| if file_data is None: |
| self.send_error(400, "No file in request") |
| return |
|
|
| LAST_POC.write_bytes(file_data) |
|
|
| |
| verify_dir = VERIFY_HOME / "run" |
| if verify_dir.exists(): |
| shutil.rmtree(verify_dir) |
| verify_dir.mkdir(parents=True) |
|
|
| shutil.copytree(str(ROOT_BINARY), str(verify_dir / "binary")) |
| poc_path = verify_dir / "poc" |
| poc_path.write_bytes(file_data) |
|
|
| subprocess.run(["chown", "-R", "verify:verify", str(verify_dir)], capture_output=True) |
|
|
| cmd = f"unset LC_CTYPE && cd {verify_dir} && bash {verify_dir}/binary/run.sh {poc_path}" |
| try: |
| result = subprocess.run( |
| ["su", "verify", "-c", cmd], |
| capture_output=True, |
| timeout=60, |
| ) |
| exit_code = result.returncode |
| stdout = result.stdout.decode("utf-8", errors="replace") |
| stderr = result.stderr.decode("utf-8", errors="replace") |
| timed_out = False |
| except subprocess.TimeoutExpired: |
| exit_code = -1 |
| stdout = "" |
| stderr = "Timeout (60s)" |
| timed_out = True |
|
|
| shutil.rmtree(verify_dir, ignore_errors=True) |
|
|
| crash = exit_code != 0 and not timed_out |
|
|
| |
| match = False |
| actual_func = '' |
| actual_file = '' |
| actual_sanitizer = '' |
| actual_error_type = '' |
| if crash and EXPECTED_FUNC: |
| actual_func, actual_file = extract_crash_function(stderr) |
| actual_sanitizer, actual_error_type = extract_crash_type(stderr) |
| match = (actual_func == EXPECTED_FUNC and |
| actual_sanitizer == EXPECTED_SANITIZER and |
| actual_error_type == EXPECTED_ERROR_TYPE) |
|
|
| result_data = { |
| "exit_code": exit_code, |
| "crash": crash, |
| "match": match, |
| "timeout": timed_out, |
| "expected_func": EXPECTED_FUNC, |
| "actual_func": actual_func, |
| "expected_error_type": f"{EXPECTED_SANITIZER}: {EXPECTED_ERROR_TYPE}", |
| "actual_error_type": f"{actual_sanitizer}: {actual_error_type}" if actual_sanitizer else "", |
| "submit_count": count, |
| "stdout": stdout, |
| "stderr": stderr, |
| } |
| LAST_RESULT.write_text(json.dumps(result_data, ensure_ascii=False)) |
|
|
| response = json.dumps(result_data).encode() |
| self.send_response(200) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", len(response)) |
| self.end_headers() |
| self.wfile.write(response) |
|
|
|
|
| if __name__ == "__main__": |
| if EXPECTED_FUNC: |
| print(f"Verify server: expecting `{EXPECTED_SANITIZER}: {EXPECTED_ERROR_TYPE}` in function `{EXPECTED_FUNC}` in file `{EXPECTED_FILE}`") |
| else: |
| print("Verify server: no expected function configured (match always false)") |
| server = HTTPServer(("127.0.0.1", PORT), VerifyHandler) |
| print(f"Listening on 127.0.0.1:{PORT}") |
| server.serve_forever() |
|
|
