From 444d8c94904854a3cd0cf8e08cffbc51882774ac Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 22 Jun 2026 15:27:16 +0200 Subject: [PATCH] =?UTF-8?q?feat(ruleset):=20Phase=202=20(task=203)=20?= =?UTF-8?q?=E2=80=94=20rules=20as=20governed=20data,=20severity=20from=20p?= =?UTF-8?q?olicy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Externalize detection rules to versioned TOML and remove rule-local severity, so rules detect and the central CWE policy decides severity — fixing "rules too influential / not centrally managed". Zero-dependency; detection is byte-identical (benchmark recall 44/47 = 93.6%, FP 0% unchanged) and the 90/10 gate holds. - RulesetStore (ruleset/store.py + ruleset//rules.toml): PatternRule as data (rule_id, title, languages, cwe, tags, pattern, status, min_evidence_level, precision_estimate, version) with NO severity; load_ruleset (sorted, with optional rule_policy.json ledger overlay) and a stdlib TOML write_ruleset. The 32 rules were generated from the prior PATTERN_RULES with a verified exact round-trip (pattern/cwe/languages/tags/title). - findings.py: drop the hardcoded PatternRule class + PATTERN_RULES tuple; load the default ruleset from TOML; quick_scan_findings(root, targets, rankings, ruleset, policy) resolves each finding's severity from policy via resolve_severity (SEVERITY_LABEL maps 0-5 -> string) and tags findings with their rule status. - Three-status volume control: StaticFinding gains status (default "enabled"); disabled rules never fire; shadow findings are partitioned out of the report and score in the CLI (written to shadow_findings.json) — no shadow rules by default. - Emission floors (RulesetConfig.min_emit_priority / min_emit_precision, default 0.0 = emit all) replace "every match is a finding". - CLI loads the ruleset as a stage and uses it for the fail-loud policy check, detection, hunter pool, and the score CWE map. Wheel ships the ruleset TOML. 85 tests pass; ruff + mypy clean. Spec tasks 3.1-3.4 checked off. Co-Authored-By: Claude Opus 4.8 --- .../spec.json | 2 +- .../harnessx-self-improving-rulesets/tasks.md | 10 +- src/openultrasast/cli.py | 32 +- src/openultrasast/config.py | 18 + src/openultrasast/findings.py | 362 ++---------------- src/openultrasast/hunter.py | 24 +- src/openultrasast/ruleset/__init__.py | 17 + src/openultrasast/ruleset/c/rules.toml | 71 ++++ src/openultrasast/ruleset/groovy/rules.toml | 11 + src/openultrasast/ruleset/java/rules.toml | 47 +++ .../ruleset/javascript/rules.toml | 131 +++++++ src/openultrasast/ruleset/python/rules.toml | 119 ++++++ src/openultrasast/ruleset/store.py | 142 +++++++ tests/test_findings.py | 3 +- tests/test_score_pipeline.py | 3 +- 15 files changed, 645 insertions(+), 347 deletions(-) create mode 100644 src/openultrasast/ruleset/__init__.py create mode 100644 src/openultrasast/ruleset/c/rules.toml create mode 100644 src/openultrasast/ruleset/groovy/rules.toml create mode 100644 src/openultrasast/ruleset/java/rules.toml create mode 100644 src/openultrasast/ruleset/javascript/rules.toml create mode 100644 src/openultrasast/ruleset/python/rules.toml create mode 100644 src/openultrasast/ruleset/store.py diff --git a/.kiro/specs/harnessx-self-improving-rulesets/spec.json b/.kiro/specs/harnessx-self-improving-rulesets/spec.json index 2b20d0a..4821e9e 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/spec.json +++ b/.kiro/specs/harnessx-self-improving-rulesets/spec.json @@ -1,7 +1,7 @@ { "feature_name": "harnessx-self-improving-rulesets", "created_at": "2026-06-22T10:07:18Z", - "updated_at": "2026-06-22T10:20:00Z", + "updated_at": "2026-06-22T10:45:00Z", "language": "en", "phase": "implementation", "approvals": { diff --git a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md index 7ed590f..19d23e9 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md +++ b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md @@ -59,15 +59,15 @@ This plan splits the fused regex layer into three governed owners (rules detect, ## Phase 2 — Core: Rules-as-Data and Benchmark Signals (zero-dep, byte-identical) -- [ ] 3. Convert detection rules into versioned, governed data -- [ ] 3.1 Externalize detection patterns into a versioned Ruleset Store +- [x] 3. Convert detection rules into versioned, governed data +- [x] 3.1 Externalize detection patterns into a versioned Ruleset Store - Load detection rules from versioned data files in which each rule declares a pattern, a CWE, tags, and a loop-controlled status, carrying no rule-local severity field. - Remove the rule-local severity field from the in-memory rule model so severity can only come from policy. - Observable completion: the existing pattern rules load from versioned data files with no severity attribute present, proven by a unit test asserting the absence of the field. - _Requirements: 2.1, 2.9_ - _Boundary: RulesetStore_ -- [ ] 3.2 Implement the three-status volume control +- [x] 3.2 Implement the three-status volume control - Support exactly three rule statuses — `enabled`, `shadow`, `disabled` — where enabled fires/scores/reports, shadow fires and records outcomes but is excluded from report and score, and disabled does not fire. - Exclude shadow-status findings from the reported set and from the score input while still recording their outcome for precision tracking. - Observable completion: an enabled rule appears in report and score, a shadow rule is excluded from both yet records an outcome, and a disabled rule never fires, proven by a unit test across all three statuses. @@ -75,14 +75,14 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: RulesetStore, ScoreModel_ - _Depends: 3.1, 2.1_ -- [ ] 3.3 Remap the unmapped C/C++ buffer rules to a scored CWE +- [x] 3.3 Remap the unmapped C/C++ buffer rules to a scored CWE - Remap the C/C++ buffer-handling rules that reference the unmapped CWE-120 to CWE-121 so they resolve in policy and contribute a nonzero penalty. - Observable completion: the remapped buffer rules resolve to a severity-5 policy entry and produce a nonzero score penalty, verified by a unit test. - _Requirements: 2.8_ - _Boundary: RulesetStore_ - _Depends: 3.1, 1.1_ -- [ ] 3.4 Wire findings and severity resolution through the governance plane +- [x] 3.4 Wire findings and severity resolution through the governance plane - Pass the loaded ruleset into the finding-detection stage instead of reading an implicit module-level rule list, and resolve every finding's severity through the Policy Store. - Apply emission thresholds so not every regex match becomes a reported finding, replacing the old "rule is its own volume control" behavior. - Observable completion: a scan produces findings with policy-resolved severity and shadow findings filtered out, while the default ruleset keeps benchmark output byte-identical to the pre-feature baseline. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 80b99ad..a6c845b 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -25,7 +25,7 @@ from .calibration import ( write_ranking_calibrations, ) from .config import load_config -from .findings import PATTERN_RULES, StaticFinding, quick_scan_findings, write_findings +from .findings import StaticFinding, quick_scan_findings, write_findings from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config from .hunter import run_hunter_pool, write_hunter_trajectories from .index import build_code_chunks @@ -34,6 +34,7 @@ from .policy import assert_rules_resolve, load_policy from .preprocess import preprocess_repository, write_preprocess_artifact from .rank import rank_targets, write_rankings from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report +from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .run import ScanRun, create_scan_run from .scoring import build_score_artifact from .verification import VerificationResult, verify_findings, write_verification_results @@ -104,7 +105,8 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json") runtime.start(mode=mode, target=run.target) policy = runtime.run_stage("policy_load", load_policy) - runtime.run_stage("policy_check", lambda: assert_rules_resolve(PATTERN_RULES, policy)) + ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR)) + runtime.run_stage("policy_check", lambda: assert_rules_resolve(ruleset, policy)) static_hints = runtime.run_stage("static_mapping", lambda: _load_static_hints(config.static_analysis.sarif_paths)) write_static_hints(static_hints, run.root / "mapping" / "static_hints.json") snapshot, targets = runtime.run_stage( @@ -123,12 +125,32 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_rankings(rankings, run.root / "rank" / "ranking.json") write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json") if mode == "standard": - hunter_result = runtime.run_stage("hunter_pool", lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id)) + hunter_result = runtime.run_stage( + "hunter_pool", + lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy), + ) findings = hunter_result.findings trajectories = hunter_result.trajectories else: - findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings)) + findings = runtime.run_stage( + "quick_findings", + lambda: quick_scan_findings( + run.target, + targets, + rankings, + ruleset, + policy, + min_emit_priority=config.ruleset.min_emit_priority, + min_emit_precision=config.ruleset.min_emit_precision, + ), + ) trajectories = [] + # Shadow-status rules fire but are excluded from the report and the score; + # their outcomes are preserved separately for precision tracking. + shadow_findings = [finding for finding in findings if finding.status == "shadow"] + if shadow_findings: + findings = [finding for finding in findings if finding.status != "shadow"] + write_findings(shadow_findings, run.root / "shadow_findings.json") findings_path = run.root / "findings.json" verification_path = run.root / "verification.json" markdown_path = run.root / "report.md" @@ -144,7 +166,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut "record_calibration", lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications), ) - cwe_by_rule = {rule.rule_id: rule.cwe for rule in PATTERN_RULES} + cwe_by_rule = {rule.rule_id: rule.cwe for rule in ruleset} rule_cwe_by_id = {finding.finding_id: cwe_by_rule.get(finding.finding_id.split(":", 1)[0], "") for finding in findings} score_artifact = runtime.run_stage( "score", diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 6c72d30..83e5447 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -58,6 +58,14 @@ class ScoreConfig: blocking: bool = False +@dataclass(frozen=True) +class RulesetConfig: + # Emission floors that replace "every regex match becomes a finding". The + # defaults (0.0) emit everything, keeping output byte-identical by default. + min_emit_priority: float = 0.0 + min_emit_precision: float = 0.0 + + @dataclass(frozen=True) class ResolvedConfig: models: ModelConfig = ModelConfig() @@ -67,6 +75,7 @@ class ResolvedConfig: evidence: EvidenceConfig = EvidenceConfig() static_analysis: StaticAnalysisConfig = StaticAnalysisConfig() score: ScoreConfig = ScoreConfig() + ruleset: RulesetConfig = RulesetConfig() runs_dir: str = ".openultrasast/runs" @@ -84,6 +93,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig: evidence=_load_evidence(data.get("evidence", {})), static_analysis=_load_static_analysis(data.get("static_analysis", {})), score=_load_score(data.get("score", {})), + ruleset=_load_ruleset_config(data.get("ruleset", {})), runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"), ) @@ -170,6 +180,14 @@ def _load_score(value: object) -> ScoreConfig: ) +def _load_ruleset_config(value: object) -> RulesetConfig: + data = _section(value) + return RulesetConfig( + min_emit_priority=_float_value(data.get("min_emit_priority"), 0.0), + min_emit_precision=_float_value(data.get("min_emit_precision"), 0.0), + ) + + def _float_value(value: object, default: float) -> float: if isinstance(value, bool): return default diff --git a/src/openultrasast/findings.py b/src/openultrasast/findings.py index a4db163..20fece1 100644 --- a/src/openultrasast/findings.py +++ b/src/openultrasast/findings.py @@ -5,8 +5,10 @@ import re from dataclasses import asdict, dataclass from pathlib import Path +from .policy import CwePolicy, load_policy, resolve_severity from .preprocess import FileTarget from .rank import RankingScore +from .ruleset import DEFAULT_RULESET_DIR, PatternRule, load_ruleset @dataclass(frozen=True) @@ -25,348 +27,45 @@ class StaticFinding: reachability_conditions: list[str] tags: list[str] ranking_priority: float + status: str = "enabled" -@dataclass(frozen=True) -class PatternRule: - rule_id: str - title: str - severity: str - tags: tuple[str, ...] - pattern: re.Pattern[str] - cwe: str - languages: tuple[str, ...] +SEVERITY_LABEL = {5: "critical", 4: "high", 3: "medium", 2: "low", 1: "low", 0: "info"} + +# Default ruleset loaded from versioned TOML data (rules-as-data). +PATTERN_RULES = load_ruleset(DEFAULT_RULESET_DIR) -C_LANGUAGES = ("c", "cpp") -PYTHON_LANGUAGES = ("python",) -JS_LANGUAGES = ("javascript", "typescript") -JAVA_LANGUAGES = ("java",) -GROOVY_LANGUAGES = ("groovy",) - -PATTERN_RULES = ( - # --- C / C++ ----------------------------------------------------------------- - PatternRule( - rule_id="c-unsafe-copy", - title="Unsafe C string copy needs review", - severity="high", - tags=("memory_unsafe",), - pattern=re.compile(r"\b(gets|strcpy|strcat|sprintf)\s*\("), - cwe="CWE-121", - languages=C_LANGUAGES, - ), - PatternRule( - rule_id="c-unsafe-memory-copy", - title="Unsafe C memory copy needs review", - severity="high", - tags=("memory_unsafe",), - pattern=re.compile(r"\b(memcpy|memmove|bcopy)\s*\("), - cwe="CWE-121", - languages=C_LANGUAGES, - ), - PatternRule( - rule_id="c-unsafe-format", - title="Unsafe C format string needs review", - severity="high", - tags=("memory_unsafe",), - # Flag format functions that receive a variable format string. - # Calls with string literals, such as printf("..."), are ignored. - pattern=re.compile(r"\b(printf|vprintf|vsprintf|vsnprintf|vfprintf|syslog)\s*\(\s*[A-Za-z_]"), - cwe="CWE-134", - languages=C_LANGUAGES, - ), - PatternRule( - rule_id="c-unsafe-scanf", - title="Unbounded C scanf needs review", - severity="high", - tags=("memory_unsafe",), - pattern=re.compile(r"\b(scanf|fscanf|sscanf)\s*\([^\n]*%s"), - cwe="CWE-121", - languages=C_LANGUAGES, - ), - PatternRule( - rule_id="c-shell-exec", - title="C shell execution needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(system|popen)\s*\("), - cwe="CWE-78", - languages=C_LANGUAGES, - ), - PatternRule( - rule_id="c-exec-family", - title="C exec command execution needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(execl|execlp|execle|execv|execvp|execvpe)\s*\("), - cwe="CWE-78", - languages=C_LANGUAGES, - ), - # --- Python ------------------------------------------------------------------ - PatternRule( - rule_id="python-unsafe-eval", - title="Dynamic Python execution needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(eval|exec)\s*\("), - cwe="CWE-95", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-os-command", - title="Python OS command execution needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(os\.system|os\.popen|os\.execl|os\.execlp|os\.execv|os\.execvp)\s*\("), - cwe="CWE-78", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-shell-true", - title="Subprocess shell execution needs review", - severity="medium", - tags=("syscall_entry",), - pattern=re.compile(r"subprocess\.[a-zA-Z_]+\s*\([^\n]*shell\s*=\s*True"), - cwe="CWE-78", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-unsafe-deserialization", - title="Unsafe Python deserialization needs review", - severity="high", - tags=("deserialization",), - pattern=re.compile(r"\b(pickle\.loads|pickle\.load|cPickle\.loads|yaml\.load|marshal\.loads)\s*\("), - cwe="CWE-502", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-sql-injection", - title="Python SQL string composition needs review", - severity="high", - tags=("injection",), - # Flag .execute(...) calls built with %-formatting, .format(), string - # concatenation, or an f-string. Parameterised %s placeholders pass. - pattern=re.compile(r"(\.execute)\s*\(\s*[^\n]*(?:[\"']\s*%\s*[(A-Za-z_]|\.format\s*\(|[\"']\s*\+|\bf[\"'])"), - cwe="CWE-89", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-path-traversal", - title="Python file open with external input needs review", - severity="medium", - tags=("filesystem_entry",), - pattern=re.compile(r"\b(open)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\(|params\[|GET\[|POST\[)"), - cwe="CWE-22", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-ssrf", - title="Python outbound request with external input needs review", - severity="medium", - tags=("network_entry",), - pattern=re.compile( - r"\b(requests\.(?:get|post|put|delete|head|patch)|urllib\.request\.urlopen|urlopen)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\()" - ), - cwe="CWE-918", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-weak-hash", - title="Python weak hash needs review", - severity="low", - tags=("crypto",), - pattern=re.compile(r"\bhashlib\.(md5|sha1)\s*\("), - cwe="CWE-327", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-ssti", - title="Python server-side template injection needs review", - severity="high", - tags=("injection",), - pattern=re.compile(r"\b(render_template_string)\s*\("), - cwe="CWE-94", - languages=PYTHON_LANGUAGES, - ), - PatternRule( - rule_id="python-flask-debug", - title="Flask debug mode enabled needs review", - severity="low", - tags=("insecure_default",), - pattern=re.compile(r"\.run\s*\([^\n]*(debug)\s*=\s*True"), - cwe="CWE-489", - languages=PYTHON_LANGUAGES, - ), - # --- JavaScript / TypeScript ------------------------------------------------- - PatternRule( - rule_id="js-command-exec", - title="Node child_process command execution needs review", - severity="high", - tags=("syscall_entry",), - # exec and execSync invoke a shell. execFile and spawn with an argv array - # do not, so this rule leaves them alone. - pattern=re.compile(r"\b(exec|execSync)\s*\("), - cwe="CWE-78", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-spawn-shell", - title="Node spawn with shell needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(spawn|spawnSync)\s*\([^\n]*shell\s*:\s*true"), - cwe="CWE-78", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-eval", - title="Dynamic JavaScript evaluation needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(eval)\s*\("), - cwe="CWE-95", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-function-constructor", - title="JavaScript Function constructor needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"new\s+(Function)\s*\("), - cwe="CWE-95", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-xss-response", - title="Unsanitized HTTP response reflection needs review", - severity="high", - tags=("injection",), - pattern=re.compile(r"\.(send|write|end)\s*\(\s*[^\n]*\b(?:req|request)\.(?:query|params|body)"), - cwe="CWE-79", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-dom-xss", - title="DOM sink with untrusted input needs review", - severity="high", - tags=("injection",), - pattern=re.compile(r"\.(innerHTML|outerHTML)\s*=\s*[^\n]*\b(?:req|request|location|document\.location)"), - cwe="CWE-79", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-sql-injection", - title="JavaScript SQL string composition needs review", - severity="high", - tags=("injection",), - pattern=re.compile(r"\.(query|execute)\s*\(\s*[^\n]*(?:\$\{|\+\s*[A-Za-z_$])"), - cwe="CWE-89", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-path-traversal", - title="Node filesystem access with external input needs review", - severity="medium", - tags=("filesystem_entry",), - pattern=re.compile( - r"\bfs\.(readFile|readFileSync|writeFile|writeFileSync|createReadStream|unlink|unlinkSync)\s*\(\s*[^\n]*\b(?:req|request)\." - ), - cwe="CWE-22", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-ssrf", - title="JavaScript outbound request with external input needs review", - severity="medium", - tags=("network_entry",), - pattern=re.compile(r"\b(axios|fetch|got)\s*[.(][^\n]*\b(?:req|request)\."), - cwe="CWE-918", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-weak-hash", - title="JavaScript weak hash needs review", - severity="low", - tags=("crypto",), - pattern=re.compile(r"createHash\s*\(\s*[\"'](md5|sha1)[\"']"), - cwe="CWE-327", - languages=JS_LANGUAGES, - ), - PatternRule( - rule_id="js-deserialize", - title="JavaScript unsafe deserialization needs review", - severity="high", - tags=("deserialization",), - pattern=re.compile(r"\b(unserialize)\s*\("), - cwe="CWE-502", - languages=JS_LANGUAGES, - ), - # --- Java -------------------------------------------------------------------- - PatternRule( - rule_id="java-command-exec", - title="Java runtime command execution needs review", - severity="high", - tags=("syscall_entry",), - pattern=re.compile(r"\b(exec)\s*\("), - cwe="CWE-78", - languages=JAVA_LANGUAGES, - ), - PatternRule( - rule_id="java-sql-injection", - title="Java SQL string composition needs review", - severity="high", - tags=("injection",), - # Flag SQL strings that start with a statement keyword and then concatenate - # more data. Parameterised queries with placeholders are ignored. - pattern=re.compile(r"(?i)\"\s*(select|insert|update|delete)\b[^\"]*\"\s*\+"), - cwe="CWE-89", - languages=JAVA_LANGUAGES, - ), - PatternRule( - rule_id="java-weak-hash", - title="Java weak hash needs review", - severity="low", - tags=("crypto",), - pattern=re.compile(r"MessageDigest\.getInstance\s*\(\s*\"(MD5|SHA-1)\""), - cwe="CWE-327", - languages=JAVA_LANGUAGES, - ), - PatternRule( - rule_id="java-deserialization", - title="Java unsafe deserialization needs review", - severity="high", - tags=("deserialization",), - pattern=re.compile(r"\.(readObject)\s*\("), - cwe="CWE-502", - languages=JAVA_LANGUAGES, - ), - # --- Groovy server-side templates ------------------------------------------- - PatternRule( - rule_id="groovy-xss-unescaped", - title="Groovy template unescaped output needs review", - severity="high", - tags=("injection",), - # Flag unescaped bound variables. String literals are ignored. - pattern=re.compile(r"(yieldUnescaped)\s+[A-Za-z_]"), - cwe="CWE-79", - languages=GROOVY_LANGUAGES, - ), -) - - -def quick_scan_findings(root: Path, targets: list[FileTarget], rankings: list[RankingScore]) -> list[StaticFinding]: +def quick_scan_findings( + root: Path, + targets: list[FileTarget], + rankings: list[RankingScore], + ruleset: tuple[PatternRule, ...] | None = None, + policy: dict[str, CwePolicy] | None = None, + *, + min_emit_priority: float = 0.0, + min_emit_precision: float = 0.0, +) -> list[StaticFinding]: + ruleset = PATTERN_RULES if ruleset is None else ruleset + policy = load_policy() if policy is None else policy ranking_by_path = {ranking.path: ranking for ranking in rankings} + # Compile once; disabled rules and rules below the precision floor never fire. + compiled = [ + (rule, re.compile(rule.pattern)) for rule in ruleset if rule.status != "disabled" and rule.precision_estimate >= min_emit_precision + ] findings: list[StaticFinding] = [] for target in targets: text = _read_text(root / target.path) - for rule in PATTERN_RULES: + for rule, pattern in compiled: if rule.languages and target.language not in rule.languages: continue ranking = ranking_by_path.get(target.path) - for match in rule.pattern.finditer(text): + for match in pattern.finditer(text): if _match_is_in_comment(text, match.start(), target.language): continue - findings.append(_finding_from_match(target, rule, text, match, ranking)) + finding = _finding_from_match(target, rule, text, match, ranking, policy) + if finding.ranking_priority >= min_emit_priority: + findings.append(finding) return sorted(findings, key=lambda item: (_severity_sort(item.severity), -item.ranking_priority, item.path)) @@ -393,17 +92,19 @@ def _finding_from_match( text: str, match: re.Match[str], ranking: RankingScore | None, + policy: dict[str, CwePolicy], ) -> StaticFinding: offset = match.start() line = text.count("\n", 0, offset) + 1 reachability_status, function_name, reachability_evidence = _reachability_for_line(target, line) priority = ranking.priority if ranking is not None else 1.0 matched_token = match.group(1) if match.groups() else match.group(0).split("(", 1)[0].strip() + severity = SEVERITY_LABEL.get(resolve_severity(policy, rule.cwe), "info") return StaticFinding( finding_id=f"{rule.rule_id}:{target.path}:{line}", path=target.path, title=rule.title, - severity=rule.severity, + severity=severity, confidence="medium", evidence_level="static_corroboration", rationale=( @@ -416,6 +117,7 @@ def _finding_from_match( reachability_conditions=_reachability_conditions(reachability_evidence), tags=sorted(set(target.tags) | set(rule.tags)), ranking_priority=_finding_priority(priority, reachability_status), + status=rule.status, ) diff --git a/src/openultrasast/hunter.py b/src/openultrasast/hunter.py index a1a64c9..41cdeb9 100644 --- a/src/openultrasast/hunter.py +++ b/src/openultrasast/hunter.py @@ -7,8 +7,10 @@ from pathlib import Path from typing import Literal from .findings import StaticFinding, quick_scan_findings +from .policy import CwePolicy from .preprocess import FileTarget from .rank import RankingScore +from .ruleset import PatternRule HunterTier = Literal["A", "B", "C"] @@ -98,13 +100,21 @@ def schedule_hunter_tasks( return scheduled -def run_hunter_pool(root: Path, targets: list[FileTarget], rankings: list[RankingScore], *, scan_id: str) -> HunterPoolResult: +def run_hunter_pool( + root: Path, + targets: list[FileTarget], + rankings: list[RankingScore], + *, + scan_id: str, + ruleset: tuple[PatternRule, ...] | None = None, + policy: dict[str, CwePolicy] | None = None, +) -> HunterPoolResult: tasks = schedule_hunter_tasks(targets, rankings, retrieval_context_by_path=build_retrieval_context_by_path(root, targets)) findings: list[StaticFinding] = [] trajectories: list[HunterTrajectory] = [] max_workers = min(8, max(1, len(tasks))) with ThreadPoolExecutor(max_workers=max_workers) as executor: - for task_findings, trajectory in executor.map(lambda task: _run_hunter_task(root, scan_id, task), tasks): + for task_findings, trajectory in executor.map(lambda task: _run_hunter_task(root, scan_id, task, ruleset, policy), tasks): findings.extend(task_findings) trajectories.append(trajectory) return HunterPoolResult(findings=sorted(findings, key=lambda item: item.finding_id), trajectories=trajectories) @@ -140,8 +150,14 @@ def select_skill_snippets(target: FileTarget, *, max_snippets: int = 4) -> list[ return sorted(dict.fromkeys(snippets))[:max_snippets] -def _run_hunter_task(root: Path, scan_id: str, task: HunterTask) -> tuple[list[StaticFinding], HunterTrajectory]: - findings = quick_scan_findings(root, [task.target], [task.ranking])[: task.budget.max_findings_per_target] +def _run_hunter_task( + root: Path, + scan_id: str, + task: HunterTask, + ruleset: tuple[PatternRule, ...] | None = None, + policy: dict[str, CwePolicy] | None = None, +) -> tuple[list[StaticFinding], HunterTrajectory]: + findings = quick_scan_findings(root, [task.target], [task.ranking], ruleset, policy)[: task.budget.max_findings_per_target] return findings, HunterTrajectory( scan_id=scan_id, target_path=task.target.path, diff --git a/src/openultrasast/ruleset/__init__.py b/src/openultrasast/ruleset/__init__.py new file mode 100644 index 0000000..79709fc --- /dev/null +++ b/src/openultrasast/ruleset/__init__.py @@ -0,0 +1,17 @@ +"""Central, versioned detection ruleset (rules as governed data).""" + +from .store import ( + DEFAULT_RULESET_DIR, + PatternRule, + RulesetError, + load_ruleset, + write_ruleset, +) + +__all__ = [ + "DEFAULT_RULESET_DIR", + "PatternRule", + "RulesetError", + "load_ruleset", + "write_ruleset", +] diff --git a/src/openultrasast/ruleset/c/rules.toml b/src/openultrasast/ruleset/c/rules.toml new file mode 100644 index 0000000..6cf34a9 --- /dev/null +++ b/src/openultrasast/ruleset/c/rules.toml @@ -0,0 +1,71 @@ +[[rule]] +rule_id = "c-exec-family" +title = "C exec command execution needs review" +languages = ["c", "cpp"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(execl|execlp|execle|execv|execvp|execvpe)\s*\(''' + +[[rule]] +rule_id = "c-shell-exec" +title = "C shell execution needs review" +languages = ["c", "cpp"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(system|popen)\s*\(''' + +[[rule]] +rule_id = "c-unsafe-copy" +title = "Unsafe C string copy needs review" +languages = ["c", "cpp"] +cwe = "CWE-121" +tags = ["memory_unsafe"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(gets|strcpy|strcat|sprintf)\s*\(''' + +[[rule]] +rule_id = "c-unsafe-format" +title = "Unsafe C format string needs review" +languages = ["c", "cpp"] +cwe = "CWE-134" +tags = ["memory_unsafe"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(printf|vprintf|vsprintf|vsnprintf|vfprintf|syslog)\s*\(\s*[A-Za-z_]''' + +[[rule]] +rule_id = "c-unsafe-memory-copy" +title = "Unsafe C memory copy needs review" +languages = ["c", "cpp"] +cwe = "CWE-121" +tags = ["memory_unsafe"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(memcpy|memmove|bcopy)\s*\(''' + +[[rule]] +rule_id = "c-unsafe-scanf" +title = "Unbounded C scanf needs review" +languages = ["c", "cpp"] +cwe = "CWE-121" +tags = ["memory_unsafe"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(scanf|fscanf|sscanf)\s*\([^\n]*%s''' diff --git a/src/openultrasast/ruleset/groovy/rules.toml b/src/openultrasast/ruleset/groovy/rules.toml new file mode 100644 index 0000000..73784e3 --- /dev/null +++ b/src/openultrasast/ruleset/groovy/rules.toml @@ -0,0 +1,11 @@ +[[rule]] +rule_id = "groovy-xss-unescaped" +title = "Groovy template unescaped output needs review" +languages = ["groovy"] +cwe = "CWE-79" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''(yieldUnescaped)\s+[A-Za-z_]''' diff --git a/src/openultrasast/ruleset/java/rules.toml b/src/openultrasast/ruleset/java/rules.toml new file mode 100644 index 0000000..17dbdaa --- /dev/null +++ b/src/openultrasast/ruleset/java/rules.toml @@ -0,0 +1,47 @@ +[[rule]] +rule_id = "java-command-exec" +title = "Java runtime command execution needs review" +languages = ["java"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(exec)\s*\(''' + +[[rule]] +rule_id = "java-deserialization" +title = "Java unsafe deserialization needs review" +languages = ["java"] +cwe = "CWE-502" +tags = ["deserialization"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\.(readObject)\s*\(''' + +[[rule]] +rule_id = "java-sql-injection" +title = "Java SQL string composition needs review" +languages = ["java"] +cwe = "CWE-89" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''(?i)\"\s*(select|insert|update|delete)\b[^\"]*\"\s*\+''' + +[[rule]] +rule_id = "java-weak-hash" +title = "Java weak hash needs review" +languages = ["java"] +cwe = "CWE-327" +tags = ["crypto"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''MessageDigest\.getInstance\s*\(\s*\"(MD5|SHA-1)\"''' diff --git a/src/openultrasast/ruleset/javascript/rules.toml b/src/openultrasast/ruleset/javascript/rules.toml new file mode 100644 index 0000000..c2d0c06 --- /dev/null +++ b/src/openultrasast/ruleset/javascript/rules.toml @@ -0,0 +1,131 @@ +[[rule]] +rule_id = "js-command-exec" +title = "Node child_process command execution needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(exec|execSync)\s*\(''' + +[[rule]] +rule_id = "js-deserialize" +title = "JavaScript unsafe deserialization needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-502" +tags = ["deserialization"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(unserialize)\s*\(''' + +[[rule]] +rule_id = "js-dom-xss" +title = "DOM sink with untrusted input needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-79" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\.(innerHTML|outerHTML)\s*=\s*[^\n]*\b(?:req|request|location|document\.location)''' + +[[rule]] +rule_id = "js-eval" +title = "Dynamic JavaScript evaluation needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-95" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(eval)\s*\(''' + +[[rule]] +rule_id = "js-function-constructor" +title = "JavaScript Function constructor needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-95" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''new\s+(Function)\s*\(''' + +[[rule]] +rule_id = "js-path-traversal" +title = "Node filesystem access with external input needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-22" +tags = ["filesystem_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\bfs\.(readFile|readFileSync|writeFile|writeFileSync|createReadStream|unlink|unlinkSync)\s*\(\s*[^\n]*\b(?:req|request)\.''' + +[[rule]] +rule_id = "js-spawn-shell" +title = "Node spawn with shell needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(spawn|spawnSync)\s*\([^\n]*shell\s*:\s*true''' + +[[rule]] +rule_id = "js-sql-injection" +title = "JavaScript SQL string composition needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-89" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\.(query|execute)\s*\(\s*[^\n]*(?:\$\{|\+\s*[A-Za-z_$])''' + +[[rule]] +rule_id = "js-ssrf" +title = "JavaScript outbound request with external input needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-918" +tags = ["network_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(axios|fetch|got)\s*[.(][^\n]*\b(?:req|request)\.''' + +[[rule]] +rule_id = "js-weak-hash" +title = "JavaScript weak hash needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-327" +tags = ["crypto"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''createHash\s*\(\s*[\"'](md5|sha1)[\"']''' + +[[rule]] +rule_id = "js-xss-response" +title = "Unsanitized HTTP response reflection needs review" +languages = ["javascript", "typescript"] +cwe = "CWE-79" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\.(send|write|end)\s*\(\s*[^\n]*\b(?:req|request)\.(?:query|params|body)''' diff --git a/src/openultrasast/ruleset/python/rules.toml b/src/openultrasast/ruleset/python/rules.toml new file mode 100644 index 0000000..b38fb41 --- /dev/null +++ b/src/openultrasast/ruleset/python/rules.toml @@ -0,0 +1,119 @@ +[[rule]] +rule_id = "python-flask-debug" +title = "Flask debug mode enabled needs review" +languages = ["python"] +cwe = "CWE-489" +tags = ["insecure_default"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\.run\s*\([^\n]*(debug)\s*=\s*True''' + +[[rule]] +rule_id = "python-os-command" +title = "Python OS command execution needs review" +languages = ["python"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(os\.system|os\.popen|os\.execl|os\.execlp|os\.execv|os\.execvp)\s*\(''' + +[[rule]] +rule_id = "python-path-traversal" +title = "Python file open with external input needs review" +languages = ["python"] +cwe = "CWE-22" +tags = ["filesystem_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(open)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\(|params\[|GET\[|POST\[)''' + +[[rule]] +rule_id = "python-shell-true" +title = "Subprocess shell execution needs review" +languages = ["python"] +cwe = "CWE-78" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''subprocess\.[a-zA-Z_]+\s*\([^\n]*shell\s*=\s*True''' + +[[rule]] +rule_id = "python-sql-injection" +title = "Python SQL string composition needs review" +languages = ["python"] +cwe = "CWE-89" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''(\.execute)\s*\(\s*[^\n]*(?:[\"']\s*%\s*[(A-Za-z_]|\.format\s*\(|[\"']\s*\+|\bf[\"'])''' + +[[rule]] +rule_id = "python-ssrf" +title = "Python outbound request with external input needs review" +languages = ["python"] +cwe = "CWE-918" +tags = ["network_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(requests\.(?:get|post|put|delete|head|patch)|urllib\.request\.urlopen|urlopen)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\()''' + +[[rule]] +rule_id = "python-ssti" +title = "Python server-side template injection needs review" +languages = ["python"] +cwe = "CWE-94" +tags = ["injection"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(render_template_string)\s*\(''' + +[[rule]] +rule_id = "python-unsafe-deserialization" +title = "Unsafe Python deserialization needs review" +languages = ["python"] +cwe = "CWE-502" +tags = ["deserialization"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(pickle\.loads|pickle\.load|cPickle\.loads|yaml\.load|marshal\.loads)\s*\(''' + +[[rule]] +rule_id = "python-unsafe-eval" +title = "Dynamic Python execution needs review" +languages = ["python"] +cwe = "CWE-95" +tags = ["syscall_entry"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\b(eval|exec)\s*\(''' + +[[rule]] +rule_id = "python-weak-hash" +title = "Python weak hash needs review" +languages = ["python"] +cwe = "CWE-327" +tags = ["crypto"] +status = "enabled" +min_evidence_level = "static_corroboration" +precision_estimate = 0.0 +version = "1" +pattern = '''\bhashlib\.(md5|sha1)\s*\(''' diff --git a/src/openultrasast/ruleset/store.py b/src/openultrasast/ruleset/store.py new file mode 100644 index 0000000..c0fd1a3 --- /dev/null +++ b/src/openultrasast/ruleset/store.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import json +import tomllib +from collections.abc import Iterable +from dataclasses import dataclass, replace +from pathlib import Path + +DEFAULT_RULESET_DIR = Path(__file__).parent +VALID_STATUS = ("enabled", "shadow", "disabled") + + +class RulesetError(ValueError): + """Raised when a ruleset data file is malformed.""" + + +@dataclass(frozen=True) +class PatternRule: + """A detection rule expressed as governed data (no rule-local severity). + + ``pattern`` is human-PR-only; the self-improvement loop may only change + ``status``/``min_evidence_level``/``precision_estimate`` via the ledger. + """ + + rule_id: str + title: str + languages: tuple[str, ...] + cwe: str + tags: tuple[str, ...] + pattern: str + status: str = "enabled" + min_evidence_level: str = "static_corroboration" + precision_estimate: float = 0.0 + version: str = "1" + + +def load_ruleset(directory: Path = DEFAULT_RULESET_DIR, ledger: Path | None = None) -> tuple[PatternRule, ...]: + """Load every ``*.toml`` rule file under ``directory`` and apply the loop ledger. + + Rules are returned sorted by ``rule_id`` for determinism. The optional ledger + (``rule_policy.json``) overlays loop-owned ``status``/``min_evidence_level``/ + ``precision_estimate`` per ``rule_id``. + """ + rules: dict[str, PatternRule] = {} + for path in sorted(directory.rglob("*.toml")): + payload = tomllib.loads(path.read_text()) + for item in payload.get("rule", []): + rule = _rule_from_dict(item, path) + if rule.rule_id in rules: + raise RulesetError(f"duplicate rule_id {rule.rule_id!r} in {path}") + rules[rule.rule_id] = rule + overlay = _load_ledger(ledger) + resolved = [_apply_overlay(rule, overlay.get(rule.rule_id)) for rule in rules.values()] + return tuple(sorted(resolved, key=lambda rule: rule.rule_id)) + + +def write_ruleset(path: Path, rules: Iterable[PatternRule]) -> None: + """Write rules as TOML. Patterns use literal multi-line strings (no escaping).""" + path.parent.mkdir(parents=True, exist_ok=True) + blocks: list[str] = [] + for rule in rules: + if "'''" in rule.pattern: + raise RulesetError(f"rule {rule.rule_id} pattern cannot contain a triple single-quote") + blocks.append( + "\n".join( + [ + "[[rule]]", + f"rule_id = {_s(rule.rule_id)}", + f"title = {_s(rule.title)}", + f"languages = {_arr(rule.languages)}", + f"cwe = {_s(rule.cwe)}", + f"tags = {_arr(rule.tags)}", + f"status = {_s(rule.status)}", + f"min_evidence_level = {_s(rule.min_evidence_level)}", + f"precision_estimate = {float(rule.precision_estimate)}", + f"version = {_s(rule.version)}", + f"pattern = '''{rule.pattern}'''", + ] + ) + ) + path.write_text("\n\n".join(blocks) + "\n") + + +def _rule_from_dict(item: dict[str, object], path: Path) -> PatternRule: + try: + rule_id = str(item["rule_id"]) + cwe = str(item["cwe"]) + pattern = str(item["pattern"]) + except KeyError as exc: + raise RulesetError(f"rule in {path} missing required field {exc}") from exc + status = str(item.get("status", "enabled")) + if status not in VALID_STATUS: + raise RulesetError(f"rule {rule_id} has invalid status {status!r}") + if not pattern: + raise RulesetError(f"rule {rule_id} has an empty pattern") + return PatternRule( + rule_id=rule_id, + title=str(item.get("title", rule_id)), + languages=_str_tuple(item.get("languages")), + cwe=cwe, + tags=_str_tuple(item.get("tags")), + pattern=pattern, + status=status, + min_evidence_level=str(item.get("min_evidence_level", "static_corroboration")), + precision_estimate=float(item.get("precision_estimate", 0.0)), # type: ignore[arg-type] + version=str(item.get("version", "1")), + ) + + +def _load_ledger(ledger: Path | None) -> dict[str, dict[str, object]]: + if ledger is None or not ledger.exists(): + return {} + payload = json.loads(ledger.read_text()) + return {str(key): value for key, value in payload.items() if isinstance(value, dict)} + + +def _apply_overlay(rule: PatternRule, overlay: dict[str, object] | None) -> PatternRule: + if not overlay: + return rule + status = str(overlay.get("status", rule.status)) + if status not in VALID_STATUS: + status = rule.status + return replace( + rule, + status=status, + min_evidence_level=str(overlay.get("min_evidence_level", rule.min_evidence_level)), + precision_estimate=float(overlay.get("precision_estimate", rule.precision_estimate)), # type: ignore[arg-type] + ) + + +def _str_tuple(value: object) -> tuple[str, ...]: + if isinstance(value, list): + return tuple(str(item) for item in value) + return () + + +def _s(value: str) -> str: + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def _arr(values: tuple[str, ...]) -> str: + return "[" + ", ".join(_s(value) for value in values) + "]" diff --git a/tests/test_findings.py b/tests/test_findings.py index 2c56b44..49f0902 100644 --- a/tests/test_findings.py +++ b/tests/test_findings.py @@ -56,7 +56,8 @@ def test_findings_and_report_artifacts(tmp_path: Path) -> None: write_findings(findings, findings_path) write_markdown_report(findings, report_path) - assert json.loads(findings_path.read_text())["findings"][0]["severity"] == "medium" + # Severity is now policy-governed: CWE-78 (command injection) resolves to 5 -> "critical". + assert json.loads(findings_path.read_text())["findings"][0]["severity"] == "critical" assert "Subprocess shell execution" in report_path.read_text() diff --git a/tests/test_score_pipeline.py b/tests/test_score_pipeline.py index 6e2afaf..696c20a 100644 --- a/tests/test_score_pipeline.py +++ b/tests/test_score_pipeline.py @@ -35,7 +35,8 @@ def test_scan_fails_loud_when_an_enabled_rule_cwe_is_unmapped(tmp_path: Path, mo (repo / "app.py").write_text("x = 1\n") monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") unmapped = SimpleNamespace(rule_id="bogus", cwe="CWE-99999", status="enabled") - monkeypatch.setattr(cli, "PATTERN_RULES", (*cli.PATTERN_RULES, unmapped)) + real_loader = cli.load_ruleset + monkeypatch.setattr(cli, "load_ruleset", lambda *a, **k: (*real_loader(*a, **k), unmapped)) with pytest.raises(PolicyError): main(["scan", str(repo), "--mode", "quick"])