diff --git a/benchmarks/measurements/2026-10-30-recognition-coverage/probe-after-handler-fix.txt b/benchmarks/measurements/2026-10-30-recognition-coverage/probe-after-handler-fix.txt new file mode 100644 index 0000000..f853638 --- /dev/null +++ b/benchmarks/measurements/2026-10-30-recognition-coverage/probe-after-handler-fix.txt @@ -0,0 +1,3 @@ +bytebase__dbhub product_files= 164 route_entries= 4 fw={'(language)': 4} +matrix-org__matrix-appservice-bridge product_files= 50 route_entries= 5 fw={'(language)': 5} +fastapi-users__fastapi-users product_files= 35 route_entries= 16 fw={'(language)': 16} diff --git a/src/openultrasast/core/factpack.py b/src/openultrasast/core/factpack.py index d6b78e1..aeed2e8 100644 --- a/src/openultrasast/core/factpack.py +++ b/src/openultrasast/core/factpack.py @@ -53,23 +53,28 @@ def verify_fact_pack(pack: FactPack, facts: Sequence[Mapping[str, Any]]) -> None raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") -def admit_pack_file(path: str | Path, *, review_digest: str) -> FactPack: - """Admit a fact-pack file only when its embedded review attestation verifies. +def load_pack_file(path: str | Path) -> FactPack: + """Load a pack file whose embedded review attestation is self-consistent. - The pack file must carry ``review.note`` and ``review.digest`` where the digest is - the sha256 of the note; anything else is an unreviewed pack and is refused. + ``review.digest`` must equal sha256 of ``review.note``; anything else is refused. """ payload = json.loads(Path(path).read_text(encoding="utf-8")) review = payload.get("review", {}) note = review.get("note", "") if not note or review.get("digest") != hashlib.sha256(note.encode()).hexdigest(): raise FactPackError(f"pack {path} lacks a verifiable review attestation") - if review_digest != review["digest"]: - raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") facts = payload.get("facts") if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): raise FactPackError(f"pack {path}: facts must be a list of objects") return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=review["digest"]) -__all__ = ["FactPack", "FactPackError", "admit_fact_pack", "admit_pack_file", "verify_fact_pack"] +def admit_pack_file(path: str | Path, *, review_digest: str) -> FactPack: + """Admit a pack file only when its embedded attestation matches the supplied review digest.""" + pack = load_pack_file(path) + if review_digest != pack.review_digest: + raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") + return pack + + +__all__ = ["FactPack", "FactPackError", "admit_fact_pack", "admit_pack_file", "load_pack_file", "verify_fact_pack"] diff --git a/src/openultrasast/mapping/legacy.py b/src/openultrasast/mapping/legacy.py index a08f9d5..269538a 100644 --- a/src/openultrasast/mapping/legacy.py +++ b/src/openultrasast/mapping/legacy.py @@ -805,7 +805,7 @@ def _js_entry_points(target: FileTarget, text: str, pack_patterns: Sequence[re.P "http_request", evidence, _line_conditions(stripped), - defines_handler=bool(named) and named in bounds, + defines_handler=bool(named) and (named in bounds or named == "arrow_handler"), ) ) if "process.argv" in stripped: @@ -1209,6 +1209,12 @@ def _looks_conditional(text: str) -> bool: _JS_FUNCTION = re.compile(r"^\s*(?:export\s+)?(?:default\s+)?(?:async\s+)?function\s+([A-Za-z_$][\w$]*)") +_JS_METHOD = re.compile( + r"^\s*(?:(?:public|private|protected|static|async|readonly|override)\s+)*(?:get\s+|set\s+)?" + r"([A-Za-z_$][\w$]*)\s*\([^)]*\)\s*(?::\s*[\w<>\[\]| ,]+)?\s*\{", +) +_JS_METHOD_KEYWORDS = {"if", "for", "while", "switch", "catch", "function", "return", "do", "else", "try"} +_JS_BIND = re.compile(r"^(?:[\w$]+\.)*([\w$]+)\.bind\s*\(") _JS_EXPORTED_VERB = re.compile(r"^\s*export\s+(?:async\s+)?(?:function\s+([A-Z]+)|const\s+([A-Z]+)\s*=)") _JS_IDENTIFIER = re.compile(r"^[A-Za-z_$][\w$.]*$") _JS_AUTH_TOKENS = ("auth", "login", "session", "jwt", "token", "protect", "guard", "require") @@ -1216,11 +1222,17 @@ _JS_ROLE_TOKENS = ("role", "permission", "admin", "owner") def _js_function_bounds(lines: Sequence[str]) -> dict[str, tuple[int, int]]: - """Declared function names to their 1-based line span, by brace balance from the declaration.""" + """Declared function and class-method names to their 1-based line span, by brace balance.""" bounds: dict[str, tuple[int, int]] = {} for index, line in enumerate(lines): match = _JS_FUNCTION.match(line) if match is None: + method = _JS_METHOD.match(line) + if method is None or method.group(1) in _JS_METHOD_KEYWORDS: + continue + match = method + name = match.group(1) + if name in bounds: continue depth = 0 end = index @@ -1229,7 +1241,7 @@ def _js_function_bounds(lines: Sequence[str]) -> dict[str, tuple[int, int]]: end = cursor if depth <= 0 and "{" in "".join(lines[index : cursor + 1]): break - bounds.setdefault(match.group(1), (index + 1, end + 1)) + bounds.setdefault(name, (index + 1, end + 1)) return bounds @@ -1260,14 +1272,28 @@ _MAX_STATEMENT_LINES = 12 def _js_registration(line: str, bounds: Mapping[str, tuple[int, int]]) -> tuple[str | None, list[str]]: - """`router.get('/books', requireAuth, listBooks)` -> the handler identifier and the middleware identifiers before it.""" + """`router.get('/books', requireAuth, listBooks)` -> the handler identifier and the middleware identifiers before it. + + Handler arguments may be plain identifiers, member expressions (`api.getHealth.bind(api)`), + or inline arrows (`(req, res) => ...`). + """ body = line[line.find("(") + 1 : line.rfind(")")] if "(" in line and line.rfind(")") > line.find("(") else "" parts = [part.strip() for part in body.split(",") if part.strip()] - identifiers = [part for part in parts[1:] if _JS_IDENTIFIER.match(part)] - if not identifiers: + candidates: list[str] = [] + for part in parts[1:]: + if "=>" in part: + candidates.append("arrow_handler") + continue + bind = _JS_BIND.match(part) + if bind is not None: + candidates.append(bind.group(1)) + continue + if _JS_IDENTIFIER.match(part): + candidates.append(part) + if not candidates: return None, [line] - handler = next((name for name in reversed(identifiers) if name in bounds or name.rsplit(".", 1)[-1] in bounds), identifiers[-1]) - middleware = [name for name in identifiers if name != handler] + handler = next((name for name in reversed(candidates) if name in bounds or name.rsplit(".", 1)[-1] in bounds), candidates[-1]) + middleware = [name for name in candidates if name != handler] return handler.rsplit(".", 1)[-1], middleware # `ExportsController.exportProject` is registered; `exportProject` is defined diff --git a/tests/test_js_handler_extraction.py b/tests/test_js_handler_extraction.py new file mode 100644 index 0000000..e2ab300 --- /dev/null +++ b/tests/test_js_handler_extraction.py @@ -0,0 +1,53 @@ +"""Handler-extraction tests for detection-core 8.2 part 3 (Req 3: recognition coverage).""" + +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from openultrasast.mapping.legacy import _js_entry_points # noqa: E402 +from openultrasast.preprocess import FileTarget, detect_language # noqa: E402 + +SOURCE = """import express from "express"; +const app = express(); +class Api { + getHealth(req, res) { res.send("ok"); } + submit(req, res) { res.send("done"); } +} +const api = new Api(); +app.get("/health", api.getHealth.bind(api)); +app.post("/submit", (req, res) => { res.send("done"); }); +app.delete("/old", api.submit); +""" + + +def _target(root: Path) -> FileTarget: + path = root / "src/routes.ts" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(SOURCE, encoding="utf-8") + return FileTarget("src/routes.ts", str(path), detect_language(path), 9, [], False, [], []) + + +def test_bind_this_handler_resolves_with_defines_handler() -> None: + with tempfile.TemporaryDirectory() as tmp: + target = _target(Path(tmp)) + routes = [e for e in _js_entry_points(target, SOURCE, []) if e.kind == "route"] + health = next(e for e in routes if e.function_name == "getHealth") + assert health.defines_handler is True + + +def test_arrow_inline_handler_counts_as_defined() -> None: + with tempfile.TemporaryDirectory() as tmp: + target = _target(Path(tmp)) + routes = [e for e in _js_entry_points(target, SOURCE, []) if e.kind == "route"] + submit = next(e for e in routes if e.function_name == "arrow_handler") + assert submit.defines_handler is True + + +def test_plain_identifier_handler_still_works() -> None: + with tempfile.TemporaryDirectory() as tmp: + target = _target(Path(tmp)) + routes = [e for e in _js_entry_points(target, SOURCE, []) if e.kind == "route"] + delete = next(e for e in routes if e.function_name == "submit") + assert delete.defines_handler is True