mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 01:48:17 +00:00
feat(detection-core): package-rooted entry packs wired into JS recognition; fact digest covers packs (8.2 part 2)
This commit is contained in:
@@ -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
|
||||
pilot cases: dbhub, matrix). Script: /tmp/opencode/recognition_table_71.py; raw:
|
||||
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
|
||||
@@ -44,11 +44,17 @@ def _sha256_bytes(payload: bytes) -> 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] = {}
|
||||
for path in sorted(directory.rglob("*.toml")):
|
||||
if path.is_file():
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -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"]
|
||||
@@ -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(
|
||||
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user