feat(detection-core): package-rooted entry packs wired into JS recognition; fact digest covers packs (8.2 part 2)

This commit is contained in:
Marius Ciepluch
2026-10-30 19:25:30 +01:00
parent 1d5a509424
commit 51971cda7e
6 changed files with 163 additions and 6 deletions
@@ -24,3 +24,14 @@ jaxrs, symfony, livewire, laravel, remix, tornado. **0/12 pilot repos covered.**
First fact-pack target: **express** (5 repos; covers 2 of the 3 in-family access_control First fact-pack target: **express** (5 repos; covers 2 of the 3 in-family access_control
pilot cases: dbhub, matrix). Script: /tmp/opencode/recognition_table_71.py; raw: pilot cases: dbhub, matrix). Script: /tmp/opencode/recognition_table_71.py; raw:
table.json. table.json.
## Probe 2026-10-30 (engine wiring): handler extraction is the deeper gap
The probe ran with built-in tokens only (pack wiring is package-rooted since this commit); route statements are found (matrix: 10, dbhub: several),
but every one has `defines_handler=False` and a misidentified handler name (`res`,
`None`) on `this.app.get('/health', this.getHealth.bind(this))`-style registrations.
The prover consumes only entries with `defines_handler` + named handler + line, so all
are ignored -- this is why the admission records showed zero frameworks for these repos.
Raw probe: `probe-handler-extraction.txt`. Next increment: `_js_registration` /
`_js_handler_name` must resolve `.bind(this)` and arrow handlers. The express pack is
necessary (patterns) but not sufficient (handler identity).
@@ -0,0 +1,14 @@
bytebase__dbhub: raw route entries=10
src/server.ts:208 name=None defines_handler=False fw=None
src/server.ts:216 name=next defines_handler=False fw=None
src/server.ts:247 name=arrow_handler defines_handler=False fw=None
src/server.ts:254 name=res defines_handler=False fw=None
src/server.ts:265 name=None defines_handler=False fw=None
src/server.ts:268 name=listSources defines_handler=False fw=None
matrix-org__matrix-appservice-bridge: raw route entries=10
src/bridge.ts:1543 name=None defines_handler=False fw=None
src/components/media-proxy.ts:49 name=_res defines_handler=False fw=None
src/components/media-proxy.ts:53 name=None defines_handler=False fw=None
src/components/media-proxy.ts:54 name=res defines_handler=False fw=None
src/components/media-proxy.ts:57 name=None defines_handler=False fw=None
src/components/media-proxy.ts:62 name=None defines_handler=False fw=None
+7 -1
View File
@@ -44,11 +44,17 @@ def _sha256_bytes(payload: bytes) -> str:
def fact_manifest_disk(directory: Path) -> dict[str, str]: def fact_manifest_disk(directory: Path) -> dict[str, str]:
"""Digest every TOML fact artifact under ``directory`` as it exists on disk.""" """Digest every fact artifact under ``directory`` as it exists on disk.
Covers ruleset TOMLs plus admitted entry-pack JSON files under ``entries/``, so a
pack is part of the frozen fact-set digest (2026-10-30 detection-core 8.2).
"""
manifest: dict[str, str] = {} manifest: dict[str, str] = {}
for path in sorted(directory.rglob("*.toml")): for path in sorted(directory.rglob("*.toml")):
if path.is_file(): if path.is_file():
manifest[path.relative_to(directory).as_posix()] = _sha256_bytes(path.read_bytes()) manifest[path.relative_to(directory).as_posix()] = _sha256_bytes(path.read_bytes())
for path in sorted((directory / "entries").glob("*.json")) if (directory / "entries").is_dir() else []:
manifest[path.relative_to(directory).as_posix()] = _sha256_bytes(path.read_bytes())
return manifest return manifest
+51
View File
@@ -0,0 +1,51 @@
"""Entry-fact packs loaded through the core fact-pack contract.
Packs live in ``ruleset/entries/*.json`` and are admitted only with a self-consistent
review attestation (`core.factpack.load_pack_file`). Languages map pack languages to
source languages; route patterns become extra recognition regexes.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
from openultrasast.core.factpack import load_pack_file
_LANGUAGE_ALIASES = {"js": "javascript", "ts": "typescript"}
_cache: dict[tuple[str, tuple[tuple[str, float], ...]], dict[str, list[re.Pattern[str]]]] = {}
def load_entry_pack_patterns(repo_root: Path | None = None) -> dict[str, list[re.Pattern[str]]]:
"""Compiled route patterns per source language, from admitted entry packs.
Packs resolve from the installed package's ``ruleset/entries`` (not the analyzed
repository), so recognition is active on external repos; ``repo_root`` is accepted
only for tests that stage a fake ruleset.
"""
import openultrasast
base = repo_root if repo_root is not None else Path(openultrasast.__file__).resolve().parent
entries = base / "ruleset" / "entries"
if not entries.is_dir():
return {}
files = sorted(entries.glob("*.json"))
stamp = tuple((p.name, p.stat().st_mtime) for p in files)
key = (str(entries.resolve()), stamp)
if key in _cache:
return _cache[key]
patterns: dict[str, list[re.Pattern[str]]] = {}
for pack_file in files:
payload = json.loads(pack_file.read_text(encoding="utf-8"))
pack = load_pack_file(pack_file)
for raw_language in payload.get("languages", []):
language = _LANGUAGE_ALIASES.get(str(raw_language), str(raw_language))
for fact in pack.facts:
if fact.get("handler_kind") == "route" and fact.get("pattern"):
patterns.setdefault(language, []).append(re.compile(str(fact["pattern"])))
_cache[key] = patterns
return patterns
__all__ = ["load_entry_pack_patterns"]
+17 -5
View File
@@ -9,6 +9,7 @@ from pathlib import Path
from typing import Any, Literal, cast from typing import Any, Literal, cast
from ..preprocess import FileTarget from ..preprocess import FileTarget
from .entry_packs import load_entry_pack_patterns
from .registry import registry_entries from .registry import registry_entries
MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"] MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"]
@@ -107,7 +108,7 @@ def analyze_entry_points(
source_texts[config] = (root / config).read_text(errors="ignore") source_texts[config] = (root / config).read_text(errors="ignore")
for target in targets: for target in targets:
text = source_texts[target.path] text = source_texts[target.path]
records.extend(_language_entry_points(target, text)) records.extend(_language_entry_points(target, text, load_entry_pack_patterns()))
records.extend(registry_entries(target, text)) records.extend(registry_entries(target, text))
if target.language == "php": if target.language == "php":
framework_entries = _php_framework_entries(target, text, source_texts) framework_entries = _php_framework_entries(target, text, source_texts)
@@ -184,13 +185,16 @@ def write_entry_points(entry_points: list[EntryPointRecord], path: Path) -> None
path.write_text(json.dumps({"entry_points": [asdict(entry_point) for entry_point in entry_points]}, indent=2, sort_keys=True) + "\n") path.write_text(json.dumps({"entry_points": [asdict(entry_point) for entry_point in entry_points]}, indent=2, sort_keys=True) + "\n")
def _language_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]: def _language_entry_points(
target: FileTarget, text: str, pack_patterns: Mapping[str, list[re.Pattern[str]]] | None = None
) -> list[EntryPointRecord]:
packs = pack_patterns or {}
if target.language == "java": if target.language == "java":
return _jetty_entries(target, text) return _jetty_entries(target, text)
if target.language == "python": if target.language == "python":
return _python_entry_points(target, text) + _tornado_entries(target, text) return _python_entry_points(target, text) + _tornado_entries(target, text)
if target.language in {"javascript", "typescript"}: if target.language in {"javascript", "typescript"}:
return _js_entry_points(target, text) + _remix_exports(target, text) return _js_entry_points(target, text, packs.get(target.language, [])) + _remix_exports(target, text)
if target.language in {"c", "cpp"}: if target.language in {"c", "cpp"}:
return _c_entry_points(target, text) return _c_entry_points(target, text)
if target.language == "solidity": if target.language == "solidity":
@@ -484,6 +488,7 @@ def entry_framework(entry: object) -> str | None:
("Laravel", "laravel"), ("Laravel", "laravel"),
("Remix", "remix"), ("Remix", "remix"),
("Tornado", "tornado"), ("Tornado", "tornado"),
("Express", "express"),
): ):
if name in evidence: if name in evidence:
return identifier return identifier
@@ -770,15 +775,22 @@ def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord
return records return records
def _js_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]: def _js_entry_points(target: FileTarget, text: str, pack_patterns: Sequence[re.Pattern[str]] | None = None) -> list[EntryPointRecord]:
records: list[EntryPointRecord] = [] records: list[EntryPointRecord] = []
route_tokens = (".get(", ".post(", ".put(", ".delete(", ".patch(", ".use(") route_tokens = (".get(", ".post(", ".put(", ".delete(", ".patch(", ".use(")
extra = list(pack_patterns or [])
lines = text.splitlines() lines = text.splitlines()
bounds = _js_function_bounds(lines) bounds = _js_function_bounds(lines)
for number, stripped in _js_statements(lines): for number, stripped in _js_statements(lines):
if any(token in stripped for token in route_tokens) and any(prefix in stripped.lower() for prefix in ("app", "router", "server")): token_route = any(token in stripped for token in route_tokens) and any(
prefix in stripped.lower() for prefix in ("app", "router", "server")
)
pack_route = any(pattern.search(stripped) for pattern in extra)
if token_route or pack_route:
handler, middleware = _js_registration(stripped, bounds) handler, middleware = _js_registration(stripped, bounds)
access, evidence = _js_route_access(middleware, stripped) access, evidence = _js_route_access(middleware, stripped)
if pack_route and not token_route:
evidence = [*evidence, "Express"]
start, end = bounds.get(handler or "", (number, number)) start, end = bounds.get(handler or "", (number, number))
named = handler or _js_handler_name(stripped) named = handler or _js_handler_name(stripped)
records.append( records.append(
+63
View File
@@ -0,0 +1,63 @@
"""Entry-pack wiring tests for detection-core 8.2 part 2 (Req 10, 3)."""
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from openultrasast.mapping.entry_packs import load_entry_pack_patterns # noqa: E402
from openultrasast.mapping.legacy import _js_entry_points, analyze_entry_points, entry_framework # noqa: E402
from openultrasast.preprocess import FileTarget, detect_language # noqa: E402
REPO = Path(__file__).resolve().parents[1]
TS_SOURCE = """import express from "express";
const app = express();
app.all("/admin", adminHandler);
function adminHandler(req, res) { res.send("ok"); }
"""
def _target(root: Path, name: str, text: str) -> FileTarget:
path = root / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
return FileTarget(name, str(path), detect_language(path), 5, [], False, [], [])
def test_pack_patterns_loaded_for_typescript() -> None:
patterns = load_entry_pack_patterns()
assert patterns.get("typescript") and any(p.search("app.all(") for p in patterns["typescript"])
def test_pack_patterns_resolve_from_package_not_target_repo() -> None:
# default call reads the installed package's ruleset, independent of any analyzed repo
assert load_entry_pack_patterns().get("typescript")
# an explicit root is a test-staging override only
with tempfile.TemporaryDirectory() as tmp:
assert load_entry_pack_patterns(Path(tmp)) == {}
def test_pack_only_pattern_recognised_as_express_route() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
target = _target(root, "src/routes.ts", TS_SOURCE)
without = _js_entry_points(target, TS_SOURCE, [])
assert not any(e.kind == "route" for e in without)
packs = load_entry_pack_patterns()
with_pack = _js_entry_points(target, TS_SOURCE, packs.get("typescript", []))
routes = [e for e in with_pack if e.kind == "route"]
assert routes, "pack-only pattern app.all( must be recognised"
assert any("Express" in " ".join(e.access_evidence) for e in routes)
assert entry_framework(routes[0]) == "express"
def test_analyze_entry_points_uses_packs() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
target = _target(root, "src/routes.ts", TS_SOURCE)
entries = analyze_entry_points(root, [target], texts={target.path: TS_SOURCE}, include_manifests=False)
routes = [e for e in entries if e.kind == "route" and e.defines_handler]
assert routes
assert entry_framework(routes[0]) == "express"