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
+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 ..preprocess import FileTarget
from .entry_packs import load_entry_pack_patterns
from .registry import registry_entries
MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"]
@@ -107,7 +108,7 @@ def analyze_entry_points(
source_texts[config] = (root / config).read_text(errors="ignore")
for target in targets:
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))
if target.language == "php":
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")
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":
return _jetty_entries(target, text)
if target.language == "python":
return _python_entry_points(target, text) + _tornado_entries(target, text)
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"}:
return _c_entry_points(target, text)
if target.language == "solidity":
@@ -484,6 +488,7 @@ def entry_framework(entry: object) -> str | None:
("Laravel", "laravel"),
("Remix", "remix"),
("Tornado", "tornado"),
("Express", "express"),
):
if name in evidence:
return identifier
@@ -770,15 +775,22 @@ def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord
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] = []
route_tokens = (".get(", ".post(", ".put(", ".delete(", ".patch(", ".use(")
extra = list(pack_patterns or [])
lines = text.splitlines()
bounds = _js_function_bounds(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)
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))
named = handler or _js_handler_name(stripped)
records.append(