mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 11:49:28 +00:00
feat(learn): labels from ground truth with provenance (decision-engine task 4)
learn/sources.toml is the fail-closed source list: every read goes through one guard that refuses an unlisted path or an [[excluded]] file name before opening it, and a source whose evaluation record is missing cannot label. Population v3 is named only as excluded, by file name, and is never opened (tested with a v3-named fixture and open/read_text tracked). learn/labels.py and `ousast learn labels` apply the design's rules: v2 declared sites and v1 fix-range functions positive, the same function on the fixed pin negative only when the fix changed it, introduce/repair delta units, benign control pushes as conditional negatives; pairs vulnerable/fixed (unscorable and title tier out, title as a sensitivity arm); dev-php recipe known functions; recorded adjudications with their reasons. v2 rows are labels marked design_informed, as the design says. Repository groups: normalised owner/name, merged by shared CVE/GHSA, shared fix commits and [[alias]] rows. assumed_benign (maintainer decision): ordinary commits of the population repositories, >= 90 days from any fix, no later security commit touching their files, weight 0.5, conditional, reported separately and never merged with verified negatives. The security filter is multilingual (CJK, Korean, Russian, European and other languages) plus CVE/GHSA/CWE references and dependency bumps; widened after reading accepted pushes (validation, passwords, guards). First snapshot, counts only (benchmarks/measurements/2026-09-30-decision-engine- labels/labels.json): verified pin labels in positive/negative repository groups injection 44/41, untrusted_destination 35/34, deserialization 20/20 (per-family), output_encoding 15/15, path 15/14, access_control 12/12 (pooled), config_secrets 5/5, memory 5/5 (insufficient). assumed_benign: 969 pushes in 27 groups from 36,000 commits scanned. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
884092e62c
commit
e6ed7854bb
@@ -22,7 +22,7 @@ the maintainer confirms its budget.
|
||||
approach), the vocabulary overlay for the engine; fixture tests with no model calls.
|
||||
- _Requirements: 8.1_
|
||||
|
||||
- [ ] 4. Labels
|
||||
- [x] 4. Labels
|
||||
- `learn/sources.toml` (fail-closed allowed sources, v3 excluded unread), `learn/labels.py`, `ousast learn
|
||||
labels`, repository-group normalisation and CVE merge, the `assumed_benign` source mined from non-security
|
||||
commits (separate, never merged); a first label snapshot committed as counts only.
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"modules": 142,
|
||||
"modules": 143,
|
||||
"counts": {
|
||||
"load_bearing": 135,
|
||||
"load_bearing": 136,
|
||||
"standalone_capability": 7
|
||||
},
|
||||
"orphaned": [],
|
||||
@@ -351,6 +351,7 @@
|
||||
"classification": "load_bearing",
|
||||
"reason": "reachable from gate, map_gate, pair_gate or the CLI",
|
||||
"importers": [
|
||||
"cli",
|
||||
"learn.features",
|
||||
"learn.roles",
|
||||
"plane.tasks.features",
|
||||
@@ -366,6 +367,14 @@
|
||||
"plane.tasks.features"
|
||||
]
|
||||
},
|
||||
{
|
||||
"module": "learn.labels",
|
||||
"classification": "load_bearing",
|
||||
"reason": "reachable from gate, map_gate, pair_gate or the CLI",
|
||||
"importers": [
|
||||
"cli"
|
||||
]
|
||||
},
|
||||
{
|
||||
"module": "learn.roles",
|
||||
"classification": "load_bearing",
|
||||
@@ -429,6 +438,7 @@
|
||||
"cpg.backend",
|
||||
"cpg.session",
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"model.candidates",
|
||||
"model.config_value",
|
||||
"model.dominance",
|
||||
@@ -671,6 +681,7 @@
|
||||
"importers": [
|
||||
"cli",
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"model.candidates",
|
||||
"model.report",
|
||||
"model.scan",
|
||||
@@ -695,6 +706,7 @@
|
||||
"improve",
|
||||
"improve.evolve",
|
||||
"improve.memory",
|
||||
"learn.labels",
|
||||
"model.candidates",
|
||||
"pair_gate",
|
||||
"plane.tasks.loop"
|
||||
@@ -817,6 +829,7 @@
|
||||
"importers": [
|
||||
"cli",
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"learn.roles",
|
||||
"plane.engine_alerts",
|
||||
"plane.generate",
|
||||
@@ -872,6 +885,7 @@
|
||||
"reason": "ai-service-plane Req 5.1/5.2: source-only functions and callers per file; runner integration pending",
|
||||
"importers": [
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"learn.roles",
|
||||
"plane.generate",
|
||||
"plane.tasks.alerts",
|
||||
@@ -944,6 +958,7 @@
|
||||
"improve.evolve",
|
||||
"index",
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"learn.roles",
|
||||
"map_gate",
|
||||
"mapping",
|
||||
@@ -1223,6 +1238,7 @@
|
||||
"improve.memory",
|
||||
"improve.validator",
|
||||
"learn.features",
|
||||
"learn.labels",
|
||||
"learn.roles",
|
||||
"pairs",
|
||||
"plane.tasks.alerts",
|
||||
@@ -1238,6 +1254,7 @@
|
||||
"classification": "load_bearing",
|
||||
"reason": "reachable from gate, map_gate, pair_gate or the CLI",
|
||||
"importers": [
|
||||
"learn.labels",
|
||||
"learn.roles",
|
||||
"ruleset.store",
|
||||
"semantic.facts"
|
||||
|
||||
@@ -0,0 +1,473 @@
|
||||
{
|
||||
"note": "First label snapshot of the learned decision engine (task 4), COUNTS ONLY: rows and repository groups per source, family, language, framework, split and unit. No candidate, path, function, commit or repository name. Built by `ousast learn labels` from src/openultrasast/learn/sources.toml (fail-closed; population v3 excluded unread). verified_pin_labels exclude conditional rows (benign_control, assumed_benign), which need a candidate signal at the tip and none at the base. `model` applies the design's floors to repository groups with positive and with negative pin labels. The full rows (labels_sha256) stay local: they name functions.",
|
||||
"assumed_benign_history": {
|
||||
"accepted": 969,
|
||||
"commits_scanned": 36000,
|
||||
"rejected": {
|
||||
"bulk change": 278,
|
||||
"dependency": 2206,
|
||||
"english": 4251,
|
||||
"later fix touches its files": 3168,
|
||||
"near_fix": 4911,
|
||||
"no changed function": 50,
|
||||
"no product source": 2368,
|
||||
"other_language": 2,
|
||||
"protocol pin": 55,
|
||||
"reference": 77
|
||||
},
|
||||
"repositories": 28
|
||||
},
|
||||
"conditional_negatives": {
|
||||
"assumed_benign": {
|
||||
"by_language": {
|
||||
"go": 18,
|
||||
"javascript": 2136,
|
||||
"php": 832,
|
||||
"python": 1049,
|
||||
"typescript": 636
|
||||
},
|
||||
"groups": 27,
|
||||
"pushes": 969,
|
||||
"rows": 4671
|
||||
},
|
||||
"benign_control": {
|
||||
"by_language": {
|
||||
"javascript": 7,
|
||||
"php": 27,
|
||||
"python": 23,
|
||||
"typescript": 15
|
||||
},
|
||||
"groups": 24,
|
||||
"pushes": 24,
|
||||
"rows": 72
|
||||
}
|
||||
},
|
||||
"created": "2026-09-30",
|
||||
"excluded_unread": [
|
||||
"candidates-v3-php.json",
|
||||
"freeze-v3-php.json",
|
||||
"population-v3-php.toml",
|
||||
"protocol-v3.md",
|
||||
"research-2026-09-30-php.md"
|
||||
],
|
||||
"floors": {
|
||||
"per_family": 20,
|
||||
"pooled": 10,
|
||||
"unit": "repository groups with positive and with negative pin labels"
|
||||
},
|
||||
"group_merges": {
|
||||
"advisory": 0,
|
||||
"alias": 0,
|
||||
"commit": 0
|
||||
},
|
||||
"groups": 105,
|
||||
"labels_sha256": "18337014d8a5c8df67e06f4e9d8e72c1fb8f82b1ed1b9a18836ab37e6cf4fdc9",
|
||||
"rows": 6417,
|
||||
"schema": 1,
|
||||
"skipped": {
|
||||
"pairs: label names no function": 44,
|
||||
"pairs: title tier (sensitivity arm)": 58,
|
||||
"pairs: unscorable pair": 7,
|
||||
"population: positive the fix did not change (no negative)": 22
|
||||
},
|
||||
"sources": [
|
||||
"adjudications",
|
||||
"assumed_benign",
|
||||
"dev-php",
|
||||
"pairs",
|
||||
"population-v1",
|
||||
"population-v2"
|
||||
],
|
||||
"verified_delta_labels": {
|
||||
"access_control:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 12,
|
||||
"positive_rows": 13
|
||||
},
|
||||
"access_control:repair": {
|
||||
"negative_groups": 12,
|
||||
"negative_rows": 13,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"config_secrets:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 18
|
||||
},
|
||||
"config_secrets:repair": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 16,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"deserialization:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 20,
|
||||
"positive_rows": 20
|
||||
},
|
||||
"deserialization:repair": {
|
||||
"negative_groups": 20,
|
||||
"negative_rows": 20,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"injection:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 42,
|
||||
"positive_rows": 77
|
||||
},
|
||||
"injection:repair": {
|
||||
"negative_groups": 41,
|
||||
"negative_rows": 66,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"memory:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 173
|
||||
},
|
||||
"memory:repair": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 173,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"output_encoding:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 15,
|
||||
"positive_rows": 56
|
||||
},
|
||||
"output_encoding:repair": {
|
||||
"negative_groups": 15,
|
||||
"negative_rows": 56,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"path:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 14,
|
||||
"positive_rows": 14
|
||||
},
|
||||
"path:repair": {
|
||||
"negative_groups": 14,
|
||||
"negative_rows": 14,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"unknown:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 2,
|
||||
"positive_rows": 2
|
||||
},
|
||||
"unknown:repair": {
|
||||
"negative_groups": 2,
|
||||
"negative_rows": 2,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
},
|
||||
"untrusted_destination:introduce": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 35,
|
||||
"positive_rows": 45
|
||||
},
|
||||
"untrusted_destination:repair": {
|
||||
"negative_groups": 33,
|
||||
"negative_rows": 36,
|
||||
"positive_groups": 0,
|
||||
"positive_rows": 0
|
||||
}
|
||||
},
|
||||
"verified_pin_labels": {
|
||||
"by_family": {
|
||||
"access_control": {
|
||||
"model": "pooled",
|
||||
"negative_groups": 12,
|
||||
"negative_rows": 13,
|
||||
"positive_groups": 12,
|
||||
"positive_rows": 13
|
||||
},
|
||||
"config_secrets": {
|
||||
"model": "insufficient",
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 17,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 18
|
||||
},
|
||||
"deserialization": {
|
||||
"model": "per_family",
|
||||
"negative_groups": 20,
|
||||
"negative_rows": 20,
|
||||
"positive_groups": 20,
|
||||
"positive_rows": 20
|
||||
},
|
||||
"injection": {
|
||||
"model": "per_family",
|
||||
"negative_groups": 41,
|
||||
"negative_rows": 89,
|
||||
"positive_groups": 44,
|
||||
"positive_rows": 81
|
||||
},
|
||||
"memory": {
|
||||
"model": "insufficient",
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 173,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 173
|
||||
},
|
||||
"output_encoding": {
|
||||
"model": "pooled",
|
||||
"negative_groups": 15,
|
||||
"negative_rows": 56,
|
||||
"positive_groups": 15,
|
||||
"positive_rows": 56
|
||||
},
|
||||
"path": {
|
||||
"model": "pooled",
|
||||
"negative_groups": 14,
|
||||
"negative_rows": 14,
|
||||
"positive_groups": 15,
|
||||
"positive_rows": 15
|
||||
},
|
||||
"unknown": {
|
||||
"model": "insufficient",
|
||||
"negative_groups": 2,
|
||||
"negative_rows": 2,
|
||||
"positive_groups": 2,
|
||||
"positive_rows": 2
|
||||
},
|
||||
"untrusted_destination": {
|
||||
"model": "per_family",
|
||||
"negative_groups": 34,
|
||||
"negative_rows": 52,
|
||||
"positive_groups": 35,
|
||||
"positive_rows": 46
|
||||
}
|
||||
},
|
||||
"by_framework": {
|
||||
"django": {
|
||||
"negative_groups": 1,
|
||||
"negative_rows": 2,
|
||||
"positive_groups": 1,
|
||||
"positive_rows": 2
|
||||
},
|
||||
"express": {
|
||||
"negative_groups": 3,
|
||||
"negative_rows": 6,
|
||||
"positive_groups": 4,
|
||||
"positive_rows": 12
|
||||
},
|
||||
"flask": {
|
||||
"negative_groups": 2,
|
||||
"negative_rows": 3,
|
||||
"positive_groups": 2,
|
||||
"positive_rows": 3
|
||||
},
|
||||
"httpx": {
|
||||
"negative_groups": 3,
|
||||
"negative_rows": 4,
|
||||
"positive_groups": 3,
|
||||
"positive_rows": 6
|
||||
},
|
||||
"lodash": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 9,
|
||||
"positive_groups": 6,
|
||||
"positive_rows": 15
|
||||
},
|
||||
"node-http-clients": {
|
||||
"negative_groups": 4,
|
||||
"negative_rows": 8,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 9
|
||||
},
|
||||
"none": {
|
||||
"negative_groups": 8,
|
||||
"negative_rows": 25,
|
||||
"positive_groups": 9,
|
||||
"positive_rows": 36
|
||||
},
|
||||
"pyyaml": {
|
||||
"negative_groups": 3,
|
||||
"negative_rows": 5,
|
||||
"positive_groups": 3,
|
||||
"positive_rows": 6
|
||||
},
|
||||
"react": {
|
||||
"negative_groups": 4,
|
||||
"negative_rows": 4,
|
||||
"positive_groups": 4,
|
||||
"positive_rows": 5
|
||||
},
|
||||
"requests": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 8,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 10
|
||||
},
|
||||
"sanitize-html": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 1,
|
||||
"positive_rows": 1
|
||||
},
|
||||
"sqlalchemy": {
|
||||
"negative_groups": 2,
|
||||
"negative_rows": 2,
|
||||
"positive_groups": 2,
|
||||
"positive_rows": 3
|
||||
},
|
||||
"unknown": {
|
||||
"negative_groups": 89,
|
||||
"negative_rows": 383,
|
||||
"positive_groups": 77,
|
||||
"positive_rows": 346
|
||||
},
|
||||
"vue": {
|
||||
"negative_groups": 3,
|
||||
"negative_rows": 5,
|
||||
"positive_groups": 4,
|
||||
"positive_rows": 6
|
||||
},
|
||||
"wordpress": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 8,
|
||||
"positive_groups": 9,
|
||||
"positive_rows": 13
|
||||
}
|
||||
},
|
||||
"by_language": {
|
||||
"c_cpp": {
|
||||
"negative_groups": 5,
|
||||
"negative_rows": 173,
|
||||
"positive_groups": 5,
|
||||
"positive_rows": 173
|
||||
},
|
||||
"java": {
|
||||
"negative_groups": 1,
|
||||
"negative_rows": 40,
|
||||
"positive_groups": 1,
|
||||
"positive_rows": 40
|
||||
},
|
||||
"javascript": {
|
||||
"negative_groups": 25,
|
||||
"negative_rows": 30,
|
||||
"positive_groups": 22,
|
||||
"positive_rows": 31
|
||||
},
|
||||
"php": {
|
||||
"negative_groups": 9,
|
||||
"negative_rows": 26,
|
||||
"positive_groups": 13,
|
||||
"positive_rows": 22
|
||||
},
|
||||
"python": {
|
||||
"negative_groups": 59,
|
||||
"negative_rows": 131,
|
||||
"positive_groups": 59,
|
||||
"positive_rows": 132
|
||||
},
|
||||
"typescript": {
|
||||
"negative_groups": 7,
|
||||
"negative_rows": 33,
|
||||
"positive_groups": 7,
|
||||
"positive_rows": 23
|
||||
},
|
||||
"unknown": {
|
||||
"negative_groups": 1,
|
||||
"negative_rows": 3,
|
||||
"positive_groups": 1,
|
||||
"positive_rows": 3
|
||||
}
|
||||
},
|
||||
"by_source": {
|
||||
"adjudications": {
|
||||
"negative_groups": 15,
|
||||
"negative_rows": 40,
|
||||
"positive_groups": 3,
|
||||
"positive_rows": 3
|
||||
},
|
||||
"dev-php": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 3,
|
||||
"positive_rows": 3
|
||||
},
|
||||
"pairs": {
|
||||
"negative_groups": 74,
|
||||
"negative_rows": 343,
|
||||
"positive_groups": 74,
|
||||
"positive_rows": 343
|
||||
},
|
||||
"population-v1": {
|
||||
"negative_groups": 11,
|
||||
"negative_rows": 35,
|
||||
"positive_groups": 11,
|
||||
"positive_rows": 50
|
||||
},
|
||||
"population-v2": {
|
||||
"negative_groups": 14,
|
||||
"negative_rows": 18,
|
||||
"positive_groups": 17,
|
||||
"positive_rows": 25
|
||||
}
|
||||
},
|
||||
"by_split": {
|
||||
"dev-php": {
|
||||
"negative_groups": 0,
|
||||
"negative_rows": 0,
|
||||
"positive_groups": 3,
|
||||
"positive_rows": 3
|
||||
},
|
||||
"holdout": {
|
||||
"negative_groups": 63,
|
||||
"negative_rows": 189,
|
||||
"positive_groups": 63,
|
||||
"positive_rows": 189
|
||||
},
|
||||
"population-v1": {
|
||||
"negative_groups": 11,
|
||||
"negative_rows": 46,
|
||||
"positive_groups": 11,
|
||||
"positive_rows": 51
|
||||
},
|
||||
"population-v2": {
|
||||
"negative_groups": 15,
|
||||
"negative_rows": 47,
|
||||
"positive_groups": 17,
|
||||
"positive_rows": 27
|
||||
},
|
||||
"train": {
|
||||
"negative_groups": 65,
|
||||
"negative_rows": 154,
|
||||
"positive_groups": 65,
|
||||
"positive_rows": 154
|
||||
}
|
||||
},
|
||||
"design_informed": {
|
||||
"adjudications": {
|
||||
"negative_groups": 13,
|
||||
"negative_rows": 29,
|
||||
"positive_groups": 2,
|
||||
"positive_rows": 2
|
||||
},
|
||||
"population-v2": {
|
||||
"negative_groups": 14,
|
||||
"negative_rows": 18,
|
||||
"positive_groups": 17,
|
||||
"positive_rows": 25
|
||||
}
|
||||
},
|
||||
"privileged_positives": 2
|
||||
}
|
||||
}
|
||||
@@ -259,6 +259,16 @@ def _main(argv: list[str] | None) -> int:
|
||||
candidates.add_argument("--generator", choices=("ir", "ruleset"), default="ir", help="which pass enumerates the candidate sites")
|
||||
candidates.add_argument("--json", action="store_true")
|
||||
|
||||
learn = subparsers.add_parser("learn", help="maintainer: the learned decision engine's data (labels)")
|
||||
learn_sub = learn.add_subparsers(dest="learn_command", required=True)
|
||||
learn_labels = learn_sub.add_parser("labels", help="labels from the fail-closed sources in learn/sources.toml (host only)")
|
||||
learn_labels.add_argument("--sources", type=Path, help="the source list (default: the packaged learn/sources.toml)")
|
||||
learn_labels.add_argument("--root", type=Path, default=Path("."), help="the repository root the source paths are relative to")
|
||||
learn_labels.add_argument("--cache", type=Path, help="clones and checkouts (default ~/.cache/openultrasast)")
|
||||
learn_labels.add_argument("--out", type=Path, help="write the full rows to OUT/labels-<sha>.jsonl (local: they name functions)")
|
||||
learn_labels.add_argument("--snapshot", type=Path, help="write the counts-only snapshot (the committable record) here")
|
||||
learn_labels.add_argument("--include-title", action="store_true", help="the title-tier pairs as well (a sensitivity arm)")
|
||||
learn_labels.add_argument("--no-assumed-benign", action="store_true", help="skip mining the assumed-benign history source")
|
||||
plane = subparsers.add_parser("plane", help="maintainer: the ax-backed service plane on this host's kind cluster")
|
||||
plane_sub = plane.add_subparsers(dest="plane_command", required=True)
|
||||
plane_run = plane_sub.add_parser("run", help="execute a Run manifest on ax; a rerun skips done tasks")
|
||||
@@ -429,6 +439,8 @@ def _main(argv: list[str] | None) -> int:
|
||||
return _model_candidates(args)
|
||||
if args.command == "plane":
|
||||
return _plane(args)
|
||||
if args.command == "learn":
|
||||
return _learn_labels(args)
|
||||
if args.command == "mcp":
|
||||
from .mcp import serve # lazy: keeps the import cycle (mcp -> cli) one-directional
|
||||
|
||||
@@ -1315,6 +1327,33 @@ def _load_scan_findings(run_dir: Path) -> list[StaticFinding]:
|
||||
return load_findings(run_dir / "findings.json")
|
||||
|
||||
|
||||
def _learn_labels(args: argparse.Namespace) -> int:
|
||||
"""``ousast learn labels``: exit 2 on a source outside the list, an excluded file or an unreadable clone."""
|
||||
from .learn.labels import DEFAULT_CACHE, DEFAULT_SOURCES, LabelSourceError, build_labels, write_labels
|
||||
|
||||
try:
|
||||
build, record = build_labels(
|
||||
args.root, sources_path=args.sources or DEFAULT_SOURCES, cache=args.cache or DEFAULT_CACHE, include_title=args.include_title,
|
||||
assumed_benign=not args.no_assumed_benign,
|
||||
) # fmt: skip
|
||||
except LabelSourceError as exc:
|
||||
print(f"learn labels: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
if args.out is not None:
|
||||
digest = write_labels(build, args.out / "labels.jsonl")
|
||||
(args.out / "labels.jsonl").rename(args.out / f"labels-{digest[:16]}.jsonl")
|
||||
record["labels_sha256"] = digest
|
||||
text = json.dumps(record, indent=2, sort_keys=True) + "\n"
|
||||
if args.snapshot is not None:
|
||||
args.snapshot.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.snapshot.write_text(text, encoding="utf-8")
|
||||
pins = record["verified_pin_labels"]["by_family"]
|
||||
print(f"labels: {record['rows']} rows in {record['groups']} repository groups; per family (positive/negative groups, model):")
|
||||
for family, entry in pins.items():
|
||||
print(f" {family}: {entry['positive_groups']}/{entry['negative_groups']} {entry['model']}")
|
||||
return 0
|
||||
|
||||
|
||||
def _plane(args: argparse.Namespace) -> int:
|
||||
from .plane import reconciler # lazy: the plane is a maintainer surface, not the scan path
|
||||
|
||||
|
||||
@@ -0,0 +1,849 @@
|
||||
"""Labels from ground truth, with provenance (learned-decision-engine Req 2, 7.1; design section 3).
|
||||
|
||||
Host only, never in a sandbox: ``ousast learn labels``. The sources are an allow-list, **fail-closed**
|
||||
(``sources.toml``): the builder reads a file only through :class:`SourceGuard`, which refuses any path the list does
|
||||
not name and every ``[[excluded]]`` file name *before* opening it -- population v3 is excluded unread.
|
||||
|
||||
Rules per source (design section 3):
|
||||
|
||||
- *Populations* (v1, v2): positive -- a declared site (v2) or a function the fix changed on the vulnerable side
|
||||
(v1, which declares no sites); negative -- the same function on the fixed pin when the fix changed it (a function
|
||||
the fix did not touch is the same code at both pins and is not a negative). Delta units: fixed..vulnerable
|
||||
introduces (positive), vulnerable..fixed repairs (negative). The benign base..tip functions are negative by
|
||||
protocol assumption, **conditional** on a candidate there having a signal at the tip and none at the base
|
||||
(``benign_control``). A vulnerable-pin candidate matching no site is unlabelled, never negative.
|
||||
- *Pairs*: the labelled function on the vulnerable side (positive) and on the fixed side (negative); unscorable
|
||||
pairs and the ``title`` tier are excluded (the tier is a sensitivity arm, ``--include-title``).
|
||||
- *Recipes* (dev-php): the reviewed ``[[known]]`` function at the recipe commit (positive).
|
||||
- *Adjudications*: recorded ``True``/``TP`` verdicts positive (``privileged`` kept), ``False``/``FP`` negative, with
|
||||
the recorded reason as evidence.
|
||||
- *Assumed benign* (maintainer decision 2026-09-30): ordinary non-security commits of the training repositories,
|
||||
at least 90 days from any fix, with no later security commit touching the same files. Their own source
|
||||
(``assumed_benign``, weight 0.5), conditional like ``benign_control``, reported separately, never merged with the
|
||||
verified negatives. The security filter is multilingual (:func:`security_reason`).
|
||||
|
||||
A plane verdict (agreed/rejected/disputed) is never a label (Req 2.2). Every row carries its source, reference, split,
|
||||
creation date and ``group``: the normalised ``owner/name`` repository, merged across URLs and forks by shared advisory
|
||||
ids (CVE/GHSA), shared fix commits and ``[[alias]]`` rows, so a repository cannot straddle folds (Req 7.1).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import tomllib
|
||||
import unicodedata
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from dataclasses import asdict, dataclass, field, replace
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from ..model.taxonomy import load_families
|
||||
from ..plane.tasks.repo_facts import _EXCLUDED_DIRS, _NOT_PRODUCT, DECLARATION, GLOBAL, enclosing
|
||||
from ..preprocess import detect_language
|
||||
from ..ruleset.frameworks import load_frameworks
|
||||
|
||||
DEFAULT_SOURCES = Path(__file__).resolve().parent / "sources.toml"
|
||||
DEFAULT_CACHE = Path.home() / ".cache" / "openultrasast"
|
||||
KINDS = ("population", "pairs", "recipes", "adjudications", "history")
|
||||
ANY_FAMILY = "*" # a benign push is negative for whatever family a candidate there has
|
||||
CONDITIONAL = "signal_at_tip_not_base"
|
||||
PER_FAMILY_FLOOR = 20 # positive and negative repository groups for a per-family model (design section 4)
|
||||
POOLED_FLOOR = 10
|
||||
MANIFESTS = (
|
||||
"composer.json", "package.json", "requirements.txt", "pyproject.toml", "setup.py", "setup.cfg", "Pipfile", "pom.xml",
|
||||
"build.gradle", "build.gradle.kts",
|
||||
) # fmt: skip
|
||||
_ADVISORY = re.compile(r"\b(CVE-\d{4}-\d{3,}|GHSA(?:-[23456789cfghjmpqrvwx]{4}){3})\b", re.IGNORECASE)
|
||||
_SITE = re.compile(r"^(?P<path>[^:]+):(?P<line>\d+):?(?P<function>.*)$")
|
||||
_HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")
|
||||
DAY = 86_400
|
||||
|
||||
|
||||
class LabelSourceError(RuntimeError):
|
||||
"""A source outside ``sources.toml``, an excluded file, a missing evaluation record or an unreadable cache."""
|
||||
|
||||
|
||||
# --- the source list and its guard -----------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Source:
|
||||
id: str
|
||||
kind: str
|
||||
files: tuple[str, ...]
|
||||
evaluation: tuple[str, ...]
|
||||
recorded: str
|
||||
status: str
|
||||
options: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Sources:
|
||||
sources: tuple[Source, ...]
|
||||
excluded: frozenset[str]
|
||||
aliases: Mapping[str, str]
|
||||
path: Path
|
||||
|
||||
def get(self, source_id: str) -> Source:
|
||||
for source in self.sources:
|
||||
if source.id == source_id:
|
||||
return source
|
||||
raise LabelSourceError(f"{source_id!r} is not a label source in {self.path}: it cannot label")
|
||||
|
||||
|
||||
def load_sources(path: Path = DEFAULT_SOURCES) -> Sources:
|
||||
"""``sources.toml``, checked: known kinds, unique ids, and no source naming an excluded file."""
|
||||
data = tomllib.loads(Path(path).read_text(encoding="utf-8"))
|
||||
excluded = frozenset(str(row["file"]) for row in data.get("excluded", []))
|
||||
sources: list[Source] = []
|
||||
for row in data.get("source", []):
|
||||
if row.get("kind") not in KINDS:
|
||||
raise LabelSourceError(f"{path}: source {row.get('id')!r} has kind {row.get('kind')!r}, not one of {KINDS}")
|
||||
files = tuple(str(f) for f in ([row["file"]] if "file" in row else row.get("files", [])))
|
||||
named = {Path(f.split("#", 1)[0]).name for f in files}
|
||||
if named & excluded:
|
||||
raise LabelSourceError(f"{path}: source {row['id']!r} names an excluded file {sorted(named & excluded)}")
|
||||
options = {k: v for k, v in row.items() if k not in ("id", "kind", "file", "files", "evaluation", "recorded", "status")}
|
||||
sources.append(
|
||||
Source(
|
||||
str(row["id"]), str(row["kind"]), files, tuple(row.get("evaluation", [])), str(row["recorded"]), str(row["status"]), options
|
||||
)
|
||||
)
|
||||
ids = [s.id for s in sources]
|
||||
if len(ids) != len(set(ids)):
|
||||
raise LabelSourceError(f"{path}: a source id is listed twice")
|
||||
aliases = {str(row["repo"]).lower(): str(row["group"]).lower() for row in data.get("alias", [])}
|
||||
return Sources(tuple(sources), excluded, aliases, Path(path))
|
||||
|
||||
|
||||
class SourceGuard:
|
||||
"""The only way the builder reads a label file: a path must be one a listed source names (or, for ``pairs``, lie
|
||||
under that catalog's directory), and its name must not be excluded. Refused before the file is opened."""
|
||||
|
||||
def __init__(self, sources: Sources, root: Path) -> None:
|
||||
self.root = Path(root).resolve()
|
||||
self.excluded = sources.excluded
|
||||
self.files: set[Path] = set()
|
||||
self.trees: set[Path] = set()
|
||||
for source in sources.sources:
|
||||
for name in source.files:
|
||||
resolved = (self.root / name.split("#", 1)[0]).resolve()
|
||||
self.files.add(resolved)
|
||||
if source.kind == "pairs":
|
||||
self.trees.add(resolved.parent)
|
||||
self.opened: list[Path] = []
|
||||
|
||||
def check(self, path: Path) -> Path:
|
||||
resolved = (self.root / path).resolve() if not Path(path).is_absolute() else Path(path).resolve()
|
||||
if resolved.name in self.excluded:
|
||||
raise LabelSourceError(f"{resolved.name} is excluded from labels (sources.toml [[excluded]]); it is never opened")
|
||||
if resolved not in self.files and not any(tree in resolved.parents for tree in self.trees):
|
||||
raise LabelSourceError(f"{path} is not a listed label source (sources.toml): refused unread")
|
||||
return resolved
|
||||
|
||||
def text(self, path: Path | str) -> str:
|
||||
resolved = self.check(Path(path))
|
||||
self.opened.append(resolved)
|
||||
return resolved.read_text(encoding="utf-8")
|
||||
|
||||
def json(self, path: Path | str) -> Any:
|
||||
return json.loads(self.text(path))
|
||||
|
||||
def toml(self, path: Path | str) -> dict[str, Any]:
|
||||
return tomllib.loads(self.text(path))
|
||||
|
||||
|
||||
# --- repository groups -----------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def repo_name(repo: str) -> str:
|
||||
"""``owner/name`` of a repository URL or slug, lower-cased: scheme, host, user, ``.git`` and slashes dropped."""
|
||||
text = repo.strip().lower()
|
||||
text = re.sub(r"^[a-z+]+://", "", text)
|
||||
text = re.sub(r"^[^/@]+@([^:/]+)[:/]", r"\1/", text)
|
||||
text = text.removesuffix("/").removesuffix(".git").removesuffix("/")
|
||||
parts = [p for p in text.split("/") if p]
|
||||
return "/".join(parts[-2:]) if len(parts) >= 2 else text
|
||||
|
||||
|
||||
def advisories(*texts: str | Iterable[str] | None) -> set[str]:
|
||||
"""CVE and GHSA ids named anywhere in ``texts`` (upper-cased)."""
|
||||
found: set[str] = set()
|
||||
for text in texts:
|
||||
values = [text] if isinstance(text, str) else list(text or ())
|
||||
for value in values:
|
||||
found |= {m.upper() for m in _ADVISORY.findall(str(value))}
|
||||
return found
|
||||
|
||||
|
||||
class Groups:
|
||||
"""Union-find over repository names: one group per repository, merged across names that share an advisory id or
|
||||
a fix commit (a fork carries its upstream's commits), or that an ``[[alias]]`` joins."""
|
||||
|
||||
def __init__(self, aliases: Mapping[str, str] | None = None) -> None:
|
||||
self.parent: dict[str, str] = {}
|
||||
self.aliases = dict(aliases or {})
|
||||
self.merged: dict[str, int] = {"advisory": 0, "commit": 0, "alias": 0}
|
||||
self._by_key: dict[str, str] = {}
|
||||
|
||||
def _find(self, name: str) -> str:
|
||||
self.parent.setdefault(name, name)
|
||||
while self.parent[name] != name:
|
||||
self.parent[name] = self.parent[self.parent[name]]
|
||||
name = self.parent[name]
|
||||
return name
|
||||
|
||||
def _union(self, a: str, b: str, why: str) -> None:
|
||||
ra, rb = self._find(a), self._find(b)
|
||||
if ra != rb:
|
||||
self.parent[max(ra, rb)] = min(ra, rb)
|
||||
self.merged[why] += 1
|
||||
|
||||
def add(self, repo: str, *, advisory_ids: Iterable[str] = (), commits: Iterable[str] = ()) -> str:
|
||||
name = repo_name(repo) if repo else ""
|
||||
if not name:
|
||||
return ""
|
||||
self._find(name)
|
||||
if name in self.aliases:
|
||||
self._union(name, self.aliases[name], "alias")
|
||||
for key, why in [(f"adv:{a}", "advisory") for a in advisory_ids] + [
|
||||
(f"sha:{c.lower()}", "commit") for c in commits if len(c) >= 12
|
||||
]:
|
||||
if key in self._by_key:
|
||||
self._union(name, self._by_key[key], why)
|
||||
else:
|
||||
self._by_key[key] = name
|
||||
return name
|
||||
|
||||
def group(self, repo: str) -> str:
|
||||
return self._find(repo_name(repo)) if repo else ""
|
||||
|
||||
def count(self) -> int:
|
||||
return len({self._find(n) for n in self.parent})
|
||||
|
||||
|
||||
# --- the security filter ---------------------------------------------------------------------------------------------
|
||||
|
||||
# English stems (matched at a word start) and whole words; then other languages, matched as substrings after NFKC
|
||||
# case-folding (CJK has no word boundaries). A commit naming any of these, or a CVE/GHSA/CWE reference, or a
|
||||
# dependency bump (which may be a security update), is not an assumed-benign push.
|
||||
ENGLISH_STEMS = (
|
||||
"secur", "vuln", "exploit", "attack", "malicious", "inject", "sanitiz", "sanitis", "escap", "unescap", "travers",
|
||||
"deserializ", "deserialis", "unserializ", "overflow", "underflow", "privileg", "permission", "unauthori",
|
||||
"unauthenti", "authoriz", "authoris", "authenticat", "bypass", "forger", "hijack", "spoof", "leak", "disclos",
|
||||
"exposure", "taint", "harden", "clickjack", "brute", "redos", "denial of service", "out of bounds", "out-of-bounds",
|
||||
"use after free", "use-after-free", "open redirect", "prototype pollution", "access control", "csrf", "xsrf", "ssrf",
|
||||
"sqli", "xss", "xxe", "idor", "insecure", "unsafe", "advisory", "hackerone", "huntr", "snyk", "cve", "cwe", "ghsa",
|
||||
# checks and secrets: a push that adds or loosens one is not assumed benign (found reading accepted pushes)
|
||||
"validat", "password", "passwd", "credential", "secret", "allowlist", "allow-list", "whitelist", "blacklist",
|
||||
"denylist", "deny-list", "guard", "reject", "restrict", "captcha", "encrypt", "decrypt", "certificat", "tls", "ssl",
|
||||
"sandbox", "safe", "protect", "mask",
|
||||
) # fmt: skip
|
||||
ENGLISH_WORDS = ("rce", "lfi", "rfi", "dos", "oob", "csp", "cors", "nonce", "acl", "jwt", "crlf", "poc", "sec")
|
||||
OTHER_LANGUAGES = (
|
||||
# Chinese
|
||||
"安全", "漏洞", "注入", "越权", "跨站", "提权", "绕过", "繞過", "鉴权", "鑒權", "权限", "權限", "未授权", "未授權", "越界",
|
||||
"溢出", "反序列化", "敏感", "泄露", "洩漏", "攻击", "攻擊", "转义", "轉義", "过滤", "過濾", "任意文件", "路径穿越",
|
||||
"目录遍历", "命令执行", "代码执行", "远程代码", "重定向", "伪造", "偽造", "劫持", "修复漏洞", "弱口令",
|
||||
"校验", "验证", "驗證", "密码", "密碼", "白名单", "黑名单", "限制", "加密",
|
||||
# Japanese
|
||||
"脆弱性", "セキュリティ", "攻撃", "インジェクション", "権限", "認証", "改ざん", "漏洩", "エスケープ", "サニタイズ", "検証",
|
||||
"パスワード",
|
||||
# Korean
|
||||
"보안", "취약점", "취약성", "인젝션", "권한", "인증", "우회", "공격", "유출", "이스케이프", "검증", "비밀번호",
|
||||
# Russian / Ukrainian
|
||||
"уязвим", "безопасн", "инъекц", "атак", "эксплойт", "обход", "привилег", "утечк", "санитиз", "экраниров", "валидац", "пароль",
|
||||
"вразлив", "безпек",
|
||||
# German, Dutch
|
||||
"sicherheit", "schwachstelle", "lücke", "angriff", "umgehung", "einschleus", "berechtigung", "beveilig", "kwetsba",
|
||||
"passwort", "validier",
|
||||
# French, Spanish, Portuguese, Italian, Romanian
|
||||
"sécurité", "securite", "vulnérab", "vulnerab", "faille", "attaque", "contournement", "seguridad", "segurança",
|
||||
"seguranca", "inyecci", "injeç", "injec", "ataque", "permiso", "permissõ", "sicurezza", "securitate",
|
||||
# Polish, Czech, Turkish, Vietnamese, Indonesian
|
||||
"bezpiecz", "podatno", "zranitel", "bezpečnost", "güvenlik", "zafiyet", "bảo mật", "lỗ hổng", "keamanan", "kerentanan",
|
||||
# Persian, Arabic, Hebrew, Hindi
|
||||
"امنیت", "آسیب", "أمان", "ثغرة", "حماية", "אבטח", "פגיע", "सुरक्षा", "भेद्यता",
|
||||
) # fmt: skip
|
||||
_REFERENCE = re.compile(r"\b(?:CVE-\d{4}-\d+|GHSA(?:-\w{4}){3}|CWE-\d+)\b", re.IGNORECASE)
|
||||
_DEPENDENCY = re.compile(r"\b(?:bump|bumps|dependabot|renovate)\b|\bupgrade [\w@/.-]+ (?:from|to)\b", re.IGNORECASE)
|
||||
_STEM = re.compile(r"(?<![a-z0-9])(?:" + "|".join(re.escape(s) for s in ENGLISH_STEMS) + ")")
|
||||
_WORD = re.compile(r"(?<![a-z0-9])(?:" + "|".join(re.escape(w) for w in ENGLISH_WORDS) + r")(?![a-z0-9])")
|
||||
|
||||
|
||||
def security_reason(message: str) -> str | None:
|
||||
"""Why a commit message is security-relevant -- ``reference``, ``english``, ``other_language``, ``dependency`` --
|
||||
or ``None`` for an ordinary commit. Not English-only: a Chinese "修复越权" (fix IDOR) is caught with no English."""
|
||||
if _REFERENCE.search(message):
|
||||
return "reference"
|
||||
folded = unicodedata.normalize("NFKC", message).casefold()
|
||||
if _STEM.search(folded) or _WORD.search(folded):
|
||||
return "english"
|
||||
if any(term.casefold() in folded for term in OTHER_LANGUAGES):
|
||||
return "other_language"
|
||||
if _DEPENDENCY.search(message):
|
||||
return "dependency"
|
||||
return None
|
||||
|
||||
|
||||
# --- labels ----------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Label:
|
||||
"""One label row (design, Data Models). ``candidate`` is ``path::function`` and, with ``repo``/``pin``, stays in
|
||||
the envelope; ``conditional`` names the signal a candidate needs for a conditional (benign) row to apply."""
|
||||
|
||||
candidate: str
|
||||
family: str
|
||||
label: int
|
||||
source: str
|
||||
source_ref: str
|
||||
unit: str
|
||||
weight: float
|
||||
split: str
|
||||
group: str
|
||||
repo: str
|
||||
language: str
|
||||
created: str
|
||||
evidence: str = ""
|
||||
direction: str | None = None
|
||||
pin_role: str = "vulnerable"
|
||||
pin: str = ""
|
||||
frameworks: tuple[str, ...] | None = None
|
||||
conditional: str | None = None
|
||||
design_informed: bool = False
|
||||
privileged: bool = False
|
||||
|
||||
|
||||
def _family_of(cwe: str | None, family: str | None) -> str:
|
||||
if family:
|
||||
return family
|
||||
found = load_families().family_of_cwe(cwe or "") if cwe else None
|
||||
return found.id if found is not None else "unknown"
|
||||
|
||||
|
||||
def _language(path: str) -> str:
|
||||
return detect_language(Path(path)) or "other"
|
||||
|
||||
|
||||
def _product(path: str) -> bool:
|
||||
parts = path.split("/")
|
||||
return not _NOT_PRODUCT.search(path) and not any(p in _EXCLUDED_DIRS or p.startswith(".") for p in parts[:-1])
|
||||
|
||||
|
||||
def frameworks_from(manifests: Mapping[str, str], headers: Iterable[str] = ()) -> tuple[str, ...]:
|
||||
"""The ``frameworks.toml`` ids a repository's dependency manifests name (``composer.json``, ``package.json``, ...),
|
||||
or whose markers a root file header carries (a WordPress plugin header). Stratification only, never a feature."""
|
||||
text = "\n".join(_dependency_text(name, body) for name, body in manifests.items())
|
||||
head = "\n".join(headers)
|
||||
found = []
|
||||
for framework in load_frameworks():
|
||||
names = {p for pkg in framework.packages for p in (pkg, pkg.rsplit(":", 1)[-1])}
|
||||
named = any(re.search(r"(?<![\w.\-/])" + re.escape(n) + r"(?![\w\-])", text, re.IGNORECASE) for n in names)
|
||||
if named or any(marker in head for marker in framework.markers):
|
||||
found.append(framework.id)
|
||||
return tuple(sorted(found))
|
||||
|
||||
|
||||
def _dependency_text(name: str, body: str) -> str:
|
||||
"""The dependency names of a manifest: a JSON manifest's dependency keys only (its description can say anything);
|
||||
any other manifest as text."""
|
||||
if not name.endswith(".json"):
|
||||
return body
|
||||
try:
|
||||
payload = json.loads(body)
|
||||
except ValueError:
|
||||
return ""
|
||||
keys = ("dependencies", "devDependencies", "peerDependencies", "optionalDependencies", "require", "require-dev")
|
||||
return "\n".join(str(k) for key in keys if isinstance(payload, dict) and isinstance(payload.get(key), dict) for k in payload[key])
|
||||
|
||||
|
||||
class Clone:
|
||||
"""Read-only git access to one case's clone (``git -C``); a failure is loud."""
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
if not (path / ".git").exists() and not (path / "HEAD").exists():
|
||||
raise LabelSourceError(f"no clone at {path}")
|
||||
self.path = path
|
||||
self._shows: dict[tuple[str, str], list[str] | None] = {}
|
||||
|
||||
def git(self, *args: str) -> str:
|
||||
done = subprocess.run(["git", "-C", str(self.path), *args], capture_output=True, text=True, errors="replace")
|
||||
if done.returncode != 0:
|
||||
raise LabelSourceError(f"git {' '.join(args[:3])} failed in {self.path.name}: {done.stderr.strip()[:200]}")
|
||||
return done.stdout
|
||||
|
||||
def lines(self, pin: str, path: str) -> list[str] | None:
|
||||
key = (pin, path)
|
||||
if key not in self._shows:
|
||||
done = subprocess.run(["git", "-C", str(self.path), "show", f"{pin}:{path}"], capture_output=True, text=True, errors="replace")
|
||||
self._shows[key] = done.stdout.splitlines() if done.returncode == 0 else None
|
||||
return self._shows[key]
|
||||
|
||||
def ranges(self, base: str, tip: str) -> tuple[dict[str, list[tuple[int, int]]], dict[str, list[tuple[int, int]]]]:
|
||||
"""(base-side, tip-side) changed line ranges per path of ``git diff -U0 base tip``."""
|
||||
return parse_diff(self.git("diff", "-U0", "--no-renames", base, tip))
|
||||
|
||||
def frameworks(self, pin: str) -> tuple[str, ...]:
|
||||
root = self.git("ls-tree", "--name-only", pin).split()
|
||||
manifests = {name: "\n".join(self.lines(pin, name) or ()) for name in root if name in MANIFESTS}
|
||||
headers = ["\n".join((self.lines(pin, name) or [])[:30]) for name in root if name.endswith(".php")][:10]
|
||||
return frameworks_from(manifests, headers)
|
||||
|
||||
def time(self, commit: str) -> int:
|
||||
return int(self.git("show", "-s", "--format=%ct", commit).strip())
|
||||
|
||||
|
||||
def parse_diff(diff: str) -> tuple[dict[str, list[tuple[int, int]]], dict[str, list[tuple[int, int]]]]:
|
||||
old: dict[str, list[tuple[int, int]]] = {}
|
||||
new: dict[str, list[tuple[int, int]]] = {}
|
||||
old_path = new_path = ""
|
||||
for line in diff.splitlines():
|
||||
if line.startswith("--- "):
|
||||
old_path = line[6:] if line.startswith("--- a/") else ""
|
||||
elif line.startswith("+++ "):
|
||||
new_path = line[6:] if line.startswith("+++ b/") else ""
|
||||
elif match := _HUNK.match(line):
|
||||
o_start, o_len, n_start, n_len = (int(match.group(i)) if match.group(i) is not None else 1 for i in range(1, 5))
|
||||
if old_path and o_len:
|
||||
old.setdefault(old_path, []).append((o_start, o_start + o_len - 1))
|
||||
if new_path and n_len:
|
||||
new.setdefault(new_path, []).append((n_start, n_start + n_len - 1))
|
||||
return old, new
|
||||
|
||||
|
||||
def changed_functions(clone: Clone, pin: str, ranges: Mapping[str, Sequence[tuple[int, int]]]) -> set[tuple[str, str]]:
|
||||
"""``(path, function)`` enclosing a changed line at ``pin``, product source files of a declared language only."""
|
||||
out: set[tuple[str, str]] = set()
|
||||
for path, spans in ranges.items():
|
||||
language = detect_language(Path(path))
|
||||
if language not in DECLARATION or not _product(path):
|
||||
continue
|
||||
lines = clone.lines(pin, path)
|
||||
if not lines:
|
||||
continue
|
||||
for lo, hi in spans:
|
||||
for number in range(max(lo, 1), min(hi, len(lines)) + 1):
|
||||
out.add((path, enclosing(lines, number - 1, language)))
|
||||
return out
|
||||
|
||||
|
||||
@dataclass
|
||||
class Build:
|
||||
"""The rows of one build and what it could not label (reported, never silent)."""
|
||||
|
||||
labels: list[Label] = field(default_factory=list)
|
||||
skipped: dict[str, int] = field(default_factory=dict)
|
||||
history: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def skip(self, reason: str, n: int = 1) -> None:
|
||||
self.skipped[reason] = self.skipped.get(reason, 0) + n
|
||||
|
||||
|
||||
class LabelBuilder:
|
||||
def __init__(
|
||||
self,
|
||||
sources: Sources,
|
||||
root: Path,
|
||||
*,
|
||||
cache: Path = DEFAULT_CACHE,
|
||||
created: str | None = None,
|
||||
include_title: bool = False,
|
||||
assumed_benign: bool = True,
|
||||
) -> None:
|
||||
self.sources = sources
|
||||
self.guard = SourceGuard(sources, root)
|
||||
self.root = Path(root)
|
||||
self.cache = Path(cache)
|
||||
self.created = created or date.today().isoformat()
|
||||
self.include_title = include_title
|
||||
self.assumed_benign = assumed_benign
|
||||
self.groups = Groups(sources.aliases)
|
||||
self.build = Build()
|
||||
self._cases: dict[str, tuple[Source, dict[str, Any]]] = {}
|
||||
|
||||
# -- entry --------------------------------------------------------------------------------------------------------
|
||||
|
||||
def run(self) -> Build:
|
||||
for source in self.sources.sources:
|
||||
missing = [e for e in source.evaluation if not (self.root / e).is_file()]
|
||||
if missing:
|
||||
raise LabelSourceError(f"source {source.id}: its evaluation record is not recorded ({missing}); it cannot label")
|
||||
for source in self.sources.sources:
|
||||
if source.kind == "population":
|
||||
self._population(source)
|
||||
for source in self.sources.sources:
|
||||
if source.kind == "pairs":
|
||||
self._pairs(source)
|
||||
elif source.kind == "recipes":
|
||||
self._recipes(source)
|
||||
elif source.kind == "adjudications":
|
||||
self._adjudications(source)
|
||||
elif source.kind == "history" and self.assumed_benign:
|
||||
self._history(source)
|
||||
self.build.labels = [replace(label, group=self.groups.group(label.repo) or label.group) for label in self.build.labels]
|
||||
return self.build
|
||||
|
||||
def _row(self, **kw: Any) -> None:
|
||||
self.build.labels.append(Label(created=self.created, **kw))
|
||||
|
||||
# -- populations --------------------------------------------------------------------------------------------------
|
||||
|
||||
def _clone(self, source: Source, case_id: str) -> Clone:
|
||||
return Clone(self.cache / str(source.options.get("cache", "independent")) / case_id)
|
||||
|
||||
def _population(self, source: Source) -> None:
|
||||
data = self.guard.toml(source.files[0])
|
||||
design_informed = bool(source.options.get("design_informed"))
|
||||
for case in data.get("case", []):
|
||||
self._cases[str(case["id"])] = (source, case)
|
||||
self.groups.add(case["repo"], advisory_ids=advisories(case.get("advisory")), commits=(case["fixed"],))
|
||||
for case in data.get("case", []):
|
||||
clone = self._clone(source, str(case["id"]))
|
||||
vulnerable, fixed = str(case["vulnerable"]), str(case["fixed"])
|
||||
old, new = clone.ranges(vulnerable, fixed)
|
||||
if source.options.get("matching") == "declared_sites":
|
||||
positives = {(site.partition("::")[0], site.partition("::")[2]) for site in case.get("sites", []) if "::" in site}
|
||||
else:
|
||||
positives = changed_functions(clone, vulnerable, old)
|
||||
fixed_changed = changed_functions(clone, fixed, new)
|
||||
frameworks = clone.frameworks(vulnerable)
|
||||
common = {
|
||||
"source": source.id, "source_ref": str(case["id"]), "split": source.id, "group": repo_name(case["repo"]),
|
||||
"repo": str(case["repo"]), "frameworks": frameworks, "design_informed": design_informed,
|
||||
} # fmt: skip
|
||||
family = str(case["family"])
|
||||
if not positives:
|
||||
self.build.skip(f"population: case with no {source.options.get('matching')} positive")
|
||||
for path, function in sorted(positives):
|
||||
language = _language(path)
|
||||
evidence = "declared site" if source.options.get("matching") == "declared_sites" else "fix range (vulnerable side)"
|
||||
self._row(
|
||||
candidate=f"{path}::{function}",
|
||||
family=family,
|
||||
label=1,
|
||||
unit="pin",
|
||||
weight=1.0,
|
||||
language=language,
|
||||
evidence=evidence,
|
||||
pin=vulnerable,
|
||||
**common,
|
||||
)
|
||||
self._row(
|
||||
candidate=f"{path}::{function}", family=family, label=1, unit="delta", weight=1.0, language=language, evidence=evidence,
|
||||
direction="introduce", pin=vulnerable, **common,
|
||||
) # fmt: skip
|
||||
if (path, function) in fixed_changed:
|
||||
for unit, direction in (("pin", None), ("delta", "repair")):
|
||||
self._row(
|
||||
candidate=f"{path}::{function}", family=family, label=0, unit=unit, weight=1.0, language=language,
|
||||
evidence="the fix changed this function", direction=direction, pin_role="fixed", pin=fixed, **common,
|
||||
) # fmt: skip
|
||||
else:
|
||||
self.build.skip("population: positive the fix did not change (no negative)")
|
||||
benign = case.get("benign") or {}
|
||||
if benign.get("base") and benign.get("tip"):
|
||||
_, tip_side = clone.ranges(str(benign["base"]), str(benign["tip"]))
|
||||
for path, function in sorted(changed_functions(clone, str(benign["tip"]), tip_side)):
|
||||
self._row(
|
||||
candidate=f"{path}::{function}", family=ANY_FAMILY, label=0, unit="delta", weight=1.0, language=_language(path),
|
||||
evidence="benign control push (protocol assumption)", direction="benign", pin_role="benign", pin=str(benign["tip"]),
|
||||
conditional=CONDITIONAL, **{**common, "source": "benign_control"},
|
||||
) # fmt: skip
|
||||
|
||||
# -- pairs --------------------------------------------------------------------------------------------------------
|
||||
|
||||
def _pairs(self, source: Source) -> None:
|
||||
from ..pairs import load_pair_catalog
|
||||
|
||||
catalog = self.guard.check(Path(source.files[0]))
|
||||
excluded_tiers = set() if self.include_title else set(source.options.get("exclude_tiers", []))
|
||||
for case in load_pair_catalog(catalog):
|
||||
repo = case.repo or f"{case.slice}:{case.name}"
|
||||
if case.repo:
|
||||
self.groups.add(case.repo, advisory_ids=advisories(case.cve, case.name), commits=(case.commit,))
|
||||
if case.unscorable:
|
||||
self.build.skip("pairs: unscorable pair")
|
||||
continue
|
||||
if case.review_tier in excluded_tiers:
|
||||
self.build.skip(f"pairs: {case.review_tier} tier (sensitivity arm)")
|
||||
continue
|
||||
for expected in case.expected:
|
||||
if not expected.function:
|
||||
self.build.skip("pairs: label names no function")
|
||||
continue
|
||||
path = expected.path or case.relpath
|
||||
common = {
|
||||
"candidate": f"{path}::{expected.function}", "family": _family_of(expected.cwe, expected.family), "source": "pairs",
|
||||
"source_ref": case.name, "weight": 1.0, "split": case.split, "group": repo_name(repo) if case.repo else repo,
|
||||
"repo": case.repo or repo, "language": case.language, "evidence": f"{case.review_tier} tier, {case.provenance}",
|
||||
} # fmt: skip
|
||||
self._row(label=1, unit="pin", **common)
|
||||
self._row(label=1, unit="delta", direction="introduce", **common)
|
||||
self._row(label=0, unit="pin", pin_role="fixed", **common)
|
||||
self._row(label=0, unit="delta", direction="repair", pin_role="fixed", **common)
|
||||
|
||||
# -- development recipes ------------------------------------------------------------------------------------------
|
||||
|
||||
def _recipes(self, source: Source) -> None:
|
||||
for name in source.files:
|
||||
recipe = self.guard.toml(name)
|
||||
url, commit = str(recipe.get("url", "")), str(recipe.get("commit", ""))
|
||||
self.groups.add(url, advisory_ids=advisories([k.get("id", "") for k in recipe.get("known", [])]))
|
||||
checkout = self.cache / str(source.options.get("checkouts", "repos")) / str(recipe.get("name", "")) / commit[:12]
|
||||
frameworks: tuple[str, ...] | None = None
|
||||
if checkout.is_dir():
|
||||
manifests = {m: (checkout / m).read_text(errors="replace") for m in MANIFESTS if (checkout / m).is_file()}
|
||||
headers = [p.read_text(errors="replace")[:3000] for p in sorted(checkout.glob("*.php"))[:10]]
|
||||
frameworks = frameworks_from(manifests, headers)
|
||||
for known in recipe.get("known", []):
|
||||
if not known.get("in_scope", True) or not known.get("function") or not known.get("file"):
|
||||
self.build.skip("recipes: known entry without an in-scope function")
|
||||
continue
|
||||
self._row(
|
||||
candidate=f"{known['file']}::{known['function']}", family=str(known.get("family", "unknown")), label=1,
|
||||
source=source.id,
|
||||
source_ref=f"{recipe.get('name')}:{known.get('id', '')}", unit="pin", weight=1.0, split=source.id, group=repo_name(url),
|
||||
repo=url, language=_language(str(known["file"])), evidence="reviewed known vulnerability", pin=commit,
|
||||
frameworks=frameworks,
|
||||
) # fmt: skip
|
||||
|
||||
# -- adjudications ------------------------------------------------------------------------------------------------
|
||||
|
||||
def _adjudications(self, source: Source) -> None:
|
||||
for name in source.files:
|
||||
path, _, key = name.partition("#")
|
||||
for index, row in enumerate(self.guard.json(path).get(key) or []):
|
||||
verdict = str(row.get("verdict") or "")
|
||||
label = 1 if verdict.startswith(("True", "TP")) else 0 if verdict.startswith(("False", "FP")) else None
|
||||
found = self._cases.get(str(row.get("case")))
|
||||
match = _SITE.match(str(row.get("site") or ""))
|
||||
if label is None or found is None or match is None:
|
||||
self.build.skip("adjudications: verdict, case or site unreadable")
|
||||
continue
|
||||
population, case = found
|
||||
site_path, line, function = match.group("path"), int(match.group("line")), match.group("function").strip()
|
||||
if not function:
|
||||
lines = self._clone(population, str(case["id"])).lines(str(case["vulnerable"]), site_path)
|
||||
language = detect_language(Path(site_path))
|
||||
function = enclosing(lines, line - 1, language) if lines and language in DECLARATION and line <= len(lines) else GLOBAL
|
||||
self._row(
|
||||
candidate=f"{site_path}::{function}", family=str(row.get("family") or case["family"]), label=label, source=source.id,
|
||||
source_ref=f"{Path(path).name}#{key}[{index}]", unit="pin", weight=1.0, split=population.id,
|
||||
group=repo_name(case["repo"]),
|
||||
repo=str(case["repo"]), language=_language(site_path), pin=str(case["vulnerable"]),
|
||||
evidence=f"recorded review: {verdict}: {str(row.get('reason') or '')[:300]}", privileged="privileged" in verdict,
|
||||
design_informed=bool(population.options.get("design_informed")),
|
||||
) # fmt: skip
|
||||
|
||||
# -- assumed-benign history ---------------------------------------------------------------------------------------
|
||||
|
||||
def _history(self, source: Source) -> None:
|
||||
opts = source.options
|
||||
wanted = set(opts.get("repositories_from", []))
|
||||
by_group: dict[str, list[tuple[Source, dict[str, Any]]]] = {}
|
||||
for population, case in self._cases.values():
|
||||
if population.id in wanted:
|
||||
by_group.setdefault(repo_name(case["repo"]), []).append((population, case))
|
||||
stats: dict[str, Any] = {"repositories": 0, "commits_scanned": 0, "accepted": 0, "rejected": {}}
|
||||
for group, members in sorted(by_group.items()):
|
||||
members.sort(key=lambda m: str(m[1]["id"]))
|
||||
population, case = members[0]
|
||||
clone = self._clone(population, str(case["id"]))
|
||||
pins = {str(c[k]) for _, c in members for k in ("vulnerable", "fixed")} | {
|
||||
str((c.get("benign") or {}).get(k)) for _, c in members for k in ("base", "tip") if (c.get("benign") or {}).get(k)
|
||||
}
|
||||
fixes = [clone.time(str(c["fixed"])) for _, c in members]
|
||||
fix_files = set().union(*(set(clone.ranges(str(c["vulnerable"]), str(c["fixed"]))[1]) for _, c in members))
|
||||
log = clone.git(
|
||||
"log", "--no-merges", "--no-renames", "--name-only", f"-n{int(opts.get('scan_commits', 1500))}",
|
||||
"--format=%x1e%H%x1f%ct%x1f%B%x1d", str(case["fixed"]),
|
||||
) # fmt: skip
|
||||
commits: list[tuple[str, int, str, set[str]]] = []
|
||||
for record in log.split("\x1e"):
|
||||
head, _, names = record.partition("\x1d")
|
||||
parts = head.split("\x1f", 2)
|
||||
if len(parts) == 3:
|
||||
commits.append((parts[0].strip(), int(parts[1]), parts[2], {n for n in names.split("\n") if n.strip()}))
|
||||
stats["repositories"] += 1
|
||||
frameworks = clone.frameworks(str(case["vulnerable"]))
|
||||
flagged_files: list[tuple[int, set[str]]] = [(t, fix_files) for t in fixes]
|
||||
ordinary: list[tuple[str, int, set[str]]] = []
|
||||
for sha, when, message, touched in commits:
|
||||
stats["commits_scanned"] += 1
|
||||
reason = "protocol pin" if sha in pins else security_reason(message)
|
||||
if reason in ("reference", "english", "other_language", "dependency"):
|
||||
flagged_files.append((when, touched))
|
||||
if reason is not None:
|
||||
stats["rejected"][reason] = stats["rejected"].get(reason, 0) + 1
|
||||
continue
|
||||
if any(abs(when - fix) < int(opts.get("min_days_from_fix", 90)) * DAY for fix in fixes):
|
||||
stats["rejected"]["near_fix"] = stats["rejected"].get("near_fix", 0) + 1
|
||||
continue
|
||||
ordinary.append((sha, when, touched))
|
||||
accepted = 0
|
||||
for sha, when, touched in ordinary:
|
||||
if accepted >= int(opts.get("max_commits_per_repository", 40)):
|
||||
break
|
||||
files = {p for p in touched if detect_language(Path(p)) in DECLARATION and _product(p)}
|
||||
later = {name for t, f in flagged_files if t > when for name in f}
|
||||
if not files or len(touched) > int(opts.get("max_files_per_commit", 20)) or files & later:
|
||||
why = (
|
||||
"no product source"
|
||||
if not files
|
||||
else "bulk change"
|
||||
if len(touched) > int(opts.get("max_files_per_commit", 20))
|
||||
else "later fix touches its files"
|
||||
)
|
||||
stats["rejected"][why] = stats["rejected"].get(why, 0) + 1
|
||||
continue
|
||||
_, tip_side = parse_diff(clone.git("show", "--format=", "-U0", "--no-renames", sha))
|
||||
functions = changed_functions(clone, sha, {p: tip_side[p] for p in files if p in tip_side})
|
||||
if not functions:
|
||||
stats["rejected"]["no changed function"] = stats["rejected"].get("no changed function", 0) + 1
|
||||
continue
|
||||
accepted += 1
|
||||
for path, function in sorted(functions):
|
||||
self._row(
|
||||
candidate=f"{path}::{function}", family=ANY_FAMILY, label=0, source=source.id, source_ref=f"{group}@{sha[:12]}",
|
||||
unit="delta", weight=float(opts.get("weight", 0.5)), split=source.id, group=group, repo=str(case["repo"]),
|
||||
language=_language(path), evidence="ordinary commit, no security keyword, no later fix touching its files",
|
||||
direction="benign_history", pin_role="benign", pin=sha, frameworks=frameworks, conditional=CONDITIONAL,
|
||||
) # fmt: skip
|
||||
stats["accepted"] += accepted
|
||||
stats["rejected"] = dict(sorted(stats["rejected"].items()))
|
||||
self.build.history = stats
|
||||
|
||||
|
||||
class LabelIndex:
|
||||
"""Verified labels by ``(candidate, family, unit, pin_role)``: a candidate with no row is **unlabelled** (a
|
||||
vulnerable-pin candidate matching no site is never a negative)."""
|
||||
|
||||
def __init__(self, labels: Iterable[Label]) -> None:
|
||||
self._rows: dict[tuple[str, str, str, str], Label] = {}
|
||||
for row in verified(labels):
|
||||
self._rows.setdefault((row.candidate, row.family, row.unit, row.pin_role), row)
|
||||
|
||||
def lookup(self, candidate: str, family: str, *, unit: str = "pin", pin_role: str = "vulnerable") -> Label | None:
|
||||
return self._rows.get((candidate, family, unit, pin_role))
|
||||
|
||||
|
||||
# --- the snapshot: counts only ---------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _tally(rows: Iterable[Label], key: Any) -> dict[str, dict[str, Any]]:
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
for row in rows:
|
||||
for value in key(row):
|
||||
entry = out.setdefault(str(value), {"positive_rows": 0, "negative_rows": 0, "positive_groups": set(), "negative_groups": set()})
|
||||
side = "positive" if row.label == 1 else "negative"
|
||||
entry[f"{side}_rows"] += 1
|
||||
entry[f"{side}_groups"].add(row.group)
|
||||
return {k: {n: (len(v) if isinstance(v, set) else v) for n, v in entry.items()} for k, entry in sorted(out.items())}
|
||||
|
||||
|
||||
def verified(rows: Iterable[Label]) -> list[Label]:
|
||||
"""Rows a model may train on as verified labels: unconditional, not ``assumed_benign``."""
|
||||
return [r for r in rows if r.conditional is None and r.source != "assumed_benign"]
|
||||
|
||||
|
||||
def snapshot(build: Build, sources: Sources) -> dict[str, Any]:
|
||||
"""The committed record: counts per source, family, language, framework, unit and label, and repository-group
|
||||
counts. No candidate, path, function, commit or repository name."""
|
||||
rows = build.labels
|
||||
pins = [r for r in verified(rows) if r.unit == "pin"]
|
||||
families = _tally(pins, lambda r: [r.family])
|
||||
for entry in families.values():
|
||||
pos, neg = entry["positive_groups"], entry["negative_groups"]
|
||||
entry["model"] = (
|
||||
"per_family" if min(pos, neg) >= PER_FAMILY_FLOOR else "pooled" if min(pos, neg) >= POOLED_FLOOR else "insufficient"
|
||||
)
|
||||
conditional = [r for r in rows if r.conditional is not None]
|
||||
return {
|
||||
"schema": 1,
|
||||
"created": rows[0].created if rows else None,
|
||||
"sources": sorted(s.id for s in sources.sources),
|
||||
"excluded_unread": sorted(sources.excluded),
|
||||
"rows": len(rows),
|
||||
"groups": len({r.group for r in rows}),
|
||||
"group_merges": {},
|
||||
"verified_pin_labels": {
|
||||
"by_source": _tally(pins, lambda r: [r.source]),
|
||||
"by_family": families,
|
||||
"by_language": _tally(pins, lambda r: [r.language]),
|
||||
"by_framework": _tally(
|
||||
pins, lambda r: list(r.frameworks) if r.frameworks else ["none" if r.frameworks is not None else "unknown"]
|
||||
),
|
||||
"by_split": _tally(pins, lambda r: [r.split]),
|
||||
"design_informed": _tally([r for r in pins if r.design_informed], lambda r: [r.source]),
|
||||
"privileged_positives": sum(1 for r in pins if r.privileged and r.label == 1),
|
||||
},
|
||||
"verified_delta_labels": _tally([r for r in verified(rows) if r.unit == "delta"], lambda r: [f"{r.family}:{r.direction}"]),
|
||||
"conditional_negatives": {
|
||||
source: {
|
||||
"rows": sum(1 for r in conditional if r.source == source),
|
||||
"groups": len({r.group for r in conditional if r.source == source}),
|
||||
"pushes": len(
|
||||
{r.source_ref if source == "assumed_benign" else f"{r.source_ref}:{r.pin}" for r in conditional if r.source == source}
|
||||
),
|
||||
"by_language": dict(sorted(_count(r.language for r in conditional if r.source == source).items())),
|
||||
}
|
||||
for source in sorted({r.source for r in conditional})
|
||||
},
|
||||
"assumed_benign_history": build.history,
|
||||
"skipped": dict(sorted(build.skipped.items())),
|
||||
"floors": {
|
||||
"per_family": PER_FAMILY_FLOOR,
|
||||
"pooled": POOLED_FLOOR,
|
||||
"unit": "repository groups with positive and with negative pin labels",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _count(values: Iterable[str]) -> dict[str, int]:
|
||||
out: dict[str, int] = {}
|
||||
for value in values:
|
||||
out[value] = out.get(value, 0) + 1
|
||||
return out
|
||||
|
||||
|
||||
def build_labels(
|
||||
root: Path,
|
||||
*,
|
||||
sources_path: Path = DEFAULT_SOURCES,
|
||||
cache: Path = DEFAULT_CACHE,
|
||||
include_title: bool = False,
|
||||
assumed_benign: bool = True,
|
||||
created: str | None = None,
|
||||
) -> tuple[Build, dict[str, Any]]:
|
||||
"""Every label the listed sources give, and the counts-only snapshot."""
|
||||
sources = load_sources(sources_path)
|
||||
builder = LabelBuilder(sources, root, cache=cache, include_title=include_title, assumed_benign=assumed_benign, created=created)
|
||||
build = builder.run()
|
||||
record = snapshot(build, sources)
|
||||
record["group_merges"] = dict(builder.groups.merged)
|
||||
return build, record
|
||||
|
||||
|
||||
def write_labels(build: Build, path: Path) -> str:
|
||||
"""``labels.jsonl`` (full rows, local only: they name paths and functions); returns its sha256."""
|
||||
import hashlib
|
||||
|
||||
text = "".join(
|
||||
json.dumps(asdict(label), sort_keys=True) + "\n"
|
||||
for label in sorted(build.labels, key=lambda r: json.dumps(asdict(r), sort_keys=True))
|
||||
)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(text, encoding="utf-8")
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ANY_FAMILY", "CONDITIONAL", "DEFAULT_SOURCES", "Build", "Clone", "Groups", "Label", "LabelBuilder", "LabelIndex", "LabelSourceError",
|
||||
"SourceGuard", "Sources", "advisories", "build_labels", "frameworks_from", "load_sources", "parse_diff", "repo_name",
|
||||
"security_reason", "snapshot", "verified", "write_labels",
|
||||
] # fmt: skip
|
||||
@@ -0,0 +1,110 @@
|
||||
# Label sources, fail-closed (learned-decision-engine Req 2.3; design section 3).
|
||||
#
|
||||
# `ousast learn labels` reads ONLY the files named here. A population that is not listed cannot produce a label, and
|
||||
# the builder never opens a file it does not list: every read goes through one guard that refuses any path outside
|
||||
# this list and every `[[excluded]]` file name, before the file is touched.
|
||||
#
|
||||
# Population v3 (PHP) is reserved for the one-time check (Req 5.3, design section 7). It is named below only as
|
||||
# excluded, by file name; nothing in it is read. It becomes a label source only in a later commit that adds it here
|
||||
# after `result-v3-engine.json` is recorded.
|
||||
#
|
||||
# Paths are relative to the repository root. `evaluation` names the files that prove the source's evaluation was
|
||||
# recorded (checked to exist, never parsed for labels); `recorded` is that record's commit date.
|
||||
|
||||
version = 1
|
||||
|
||||
[[source]]
|
||||
id = "population-v1"
|
||||
kind = "population"
|
||||
file = "benchmarks/independent/population-v1.toml"
|
||||
evaluation = ["benchmarks/independent/results-v1.json"]
|
||||
recorded = "2026-09-25"
|
||||
status = "spent" # retuned on after its evaluation (population-v2.toml header)
|
||||
matching = "fix_range" # v1 declares no sites: the functions the fix changed on the vulnerable side
|
||||
cache = "independent" # one clone per case id under ~/.cache/openultrasast/independent
|
||||
|
||||
[[source]]
|
||||
id = "population-v2"
|
||||
kind = "population"
|
||||
file = "benchmarks/independent/population-v2.toml"
|
||||
evaluation = [
|
||||
"benchmarks/independent/results-v2.json",
|
||||
"benchmarks/independent/results-v2-hunter.json",
|
||||
"benchmarks/independent/plane-increment-2.json",
|
||||
]
|
||||
recorded = "2026-09-28"
|
||||
status = "spent"
|
||||
matching = "declared_sites"
|
||||
design_informed = true # verify, the tie-break and the callers line were chosen after seeing v2 (design section 4)
|
||||
cache = "independent"
|
||||
|
||||
[[source]]
|
||||
id = "dev-php"
|
||||
kind = "recipes"
|
||||
files = ["benchmarks/repos/pmpro.toml", "benchmarks/repos/wpstatistics.toml", "benchmarks/repos/mwwpform.toml"]
|
||||
evaluation = ["benchmarks/push/inputs.json"]
|
||||
recorded = "2026-09-12"
|
||||
status = "development"
|
||||
checkouts = "repos" # ~/.cache/openultrasast/repos/<name>, read for dependency manifests only
|
||||
|
||||
[[source]]
|
||||
id = "pairs"
|
||||
kind = "pairs"
|
||||
file = "benchmarks/pairs/catalog.toml" # with the slice catalogs load_pair_catalog adds
|
||||
evaluation = ["benchmarks/measurements/2026-09-30-harnessx-removal-baseline/pairs.json"]
|
||||
recorded = "2026-09-30"
|
||||
status = "development"
|
||||
exclude_tiers = ["title"] # labels read from advisory titles; a sensitivity arm only (`--include-title`)
|
||||
|
||||
[[source]]
|
||||
id = "adjudications"
|
||||
kind = "adjudications"
|
||||
files = [
|
||||
"benchmarks/independent/results-v1.json#precision_review",
|
||||
"benchmarks/independent/results-v2.json#precision_sample",
|
||||
"benchmarks/independent/results-v2-hunter.json#precision_sample",
|
||||
]
|
||||
evaluation = ["benchmarks/independent/results-v1.json", "benchmarks/independent/results-v2.json"]
|
||||
recorded = "2026-09-28"
|
||||
status = "recorded verdicts"
|
||||
# verifier-check-2026-09-29.json records aggregate counts only (no per-finding verdict), so it labels nothing.
|
||||
|
||||
# Maintainer decision at design approval (2026-09-30): ordinary non-security commits of the training repositories are
|
||||
# mined as ASSUMED-BENIGN pushes -- their own source, reported separately, never merged with verified negatives.
|
||||
[[source]]
|
||||
id = "assumed_benign"
|
||||
kind = "history"
|
||||
repositories_from = ["population-v1", "population-v2"]
|
||||
recorded = "2026-09-30"
|
||||
status = "assumption"
|
||||
weight = 0.5
|
||||
min_days_from_fix = 90
|
||||
scan_commits = 1500 # newest ancestors of the fixed pin looked at, per repository
|
||||
max_commits_per_repository = 40
|
||||
max_files_per_commit = 20 # a larger commit is a bulk change, not a push
|
||||
|
||||
[[excluded]]
|
||||
file = "population-v3-php.toml"
|
||||
reason = "population v3 is reserved for the one-time check (protocol-v3.md); never opened"
|
||||
|
||||
[[excluded]]
|
||||
file = "freeze-v3-php.json"
|
||||
reason = "v3 freeze record; never opened"
|
||||
|
||||
[[excluded]]
|
||||
file = "candidates-v3-php.json"
|
||||
reason = "v3 candidate list; never opened"
|
||||
|
||||
[[excluded]]
|
||||
file = "research-2026-09-30-php.md"
|
||||
reason = "v3 selection research; never opened"
|
||||
|
||||
[[excluded]]
|
||||
file = "protocol-v3.md"
|
||||
reason = "v3 protocol; never opened"
|
||||
|
||||
# Repository aliases: forks and renames that are one repository group. Groups sharing an advisory id (CVE/GHSA) or
|
||||
# a fix commit are merged automatically; list here only what those cannot see.
|
||||
# [[alias]]
|
||||
# repo = "someone/fork"
|
||||
# group = "upstream/project"
|
||||
@@ -0,0 +1,297 @@
|
||||
"""Labels from ground truth with provenance (learned-decision-engine task 4; Req 2.1-2.3, 7.1) on fixture
|
||||
populations, pairs, adjudications and a fixture git history. The fail-closed source list is checked with a
|
||||
v3-named population that must never be opened."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast.benchmark import ExpectedFinding
|
||||
from openultrasast.learn import labels as L
|
||||
from openultrasast.pairs import PairCase
|
||||
|
||||
DAY = 86_400
|
||||
T0 = 1_600_000_000 # 2020-09-13
|
||||
|
||||
|
||||
def _git(repo: Path, *args: str, when: int | None = None) -> str:
|
||||
env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"}
|
||||
if when is not None:
|
||||
env |= {"GIT_AUTHOR_DATE": f"{when} +0000", "GIT_COMMITTER_DATE": f"{when} +0000"}
|
||||
return subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True, text=True, env=env).stdout.strip()
|
||||
|
||||
|
||||
def _commit(repo: Path, files: dict[str, str], message: str, when: int) -> str:
|
||||
for name, text in files.items():
|
||||
(repo / name).parent.mkdir(parents=True, exist_ok=True)
|
||||
(repo / name).write_text(text)
|
||||
_git(repo, "add", "-A")
|
||||
_git(repo, "commit", "-q", "-m", message, when=when)
|
||||
return _git(repo, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
APP = (
|
||||
"<?php\nfunction show($id) {\n $q = 'SELECT * WHERE id=' . $id;\n return mysqli_query($db, $q);\n}\n"
|
||||
"function helper($x) {\n return $x;\n}\n"
|
||||
)
|
||||
FIXED = APP.replace("'SELECT * WHERE id=' . $id", "'SELECT * WHERE id=' . intval($id)")
|
||||
OTHER = "<?php\nfunction render($a) {\n return $a;\n}\n"
|
||||
|
||||
|
||||
def _case_repo(cache: Path, case_id: str) -> dict[str, str]:
|
||||
"""A clone with history: old ordinary commits, a Chinese security commit, the vulnerable and fixed pins, a benign push."""
|
||||
repo = cache / "independent" / case_id
|
||||
repo.mkdir(parents=True)
|
||||
_git(repo, "init", "-q")
|
||||
pins: dict[str, str] = {}
|
||||
_commit(repo, {"src/app.php": APP, "src/other.php": OTHER, "composer.json": '{"require": {"php": ">=7"}}'}, "initial import", T0)
|
||||
pins["ordinary"] = _commit(repo, {"src/other.php": OTHER.replace("$a;", "$a . '';")}, "tidy the render helper", T0 + 10 * DAY)
|
||||
pins["chinese"] = _commit(repo, {"src/admin.php": OTHER.replace("render", "grant")}, "修复越权访问问题", T0 + 20 * DAY)
|
||||
pins["touched_later"] = _commit(repo, {"src/app.php": APP + "\n"}, "whitespace in app", T0 + 30 * DAY)
|
||||
pins["near_fix"] = _commit(repo, {"src/other.php": OTHER}, "reword the renderer", T0 + 190 * DAY)
|
||||
pins["vulnerable"] = _commit(repo, {"src/app.php": APP}, "restore app", T0 + 200 * DAY)
|
||||
pins["fixed"] = _commit(repo, {"src/app.php": FIXED}, "cast the id", T0 + 260 * DAY)
|
||||
pins["benign_base"] = pins["fixed"]
|
||||
pins["benign_tip"] = _commit(repo, {"src/app.php": FIXED.replace("return $x;", "return $x; // same")}, "comment", T0 + 270 * DAY)
|
||||
return pins
|
||||
|
||||
|
||||
def _layout(tmp_path: Path) -> tuple[Path, Path, dict[str, str]]:
|
||||
"""A repository root with a v2-shaped population, a v3-named reserved population beside it, results files, and
|
||||
the sources.toml that lists the first and excludes the second."""
|
||||
root, cache = tmp_path / "repo", tmp_path / "cache"
|
||||
pins = _case_repo(cache, "acme-sqli")
|
||||
ind = root / "benchmarks" / "independent"
|
||||
ind.mkdir(parents=True)
|
||||
population = (
|
||||
'[[case]]\nid = "acme-sqli"\nrepo = "https://github.com/Acme/Shop.git"\nfamily = "injection"\nlanguage = "php"\n'
|
||||
f'advisory = ["CVE-2024-0001"]\nvulnerable = "{pins["vulnerable"]}"\nfixed = "{pins["fixed"]}"\n'
|
||||
'sites = ["src/app.php::show", "src/app.php::helper"]\n'
|
||||
f'benign = {{ base = "{pins["benign_base"]}", tip = "{pins["benign_tip"]}" }}\n'
|
||||
)
|
||||
(ind / "population-v2.toml").write_text(population)
|
||||
(ind / "population-v3-php.toml").write_text('[[case]]\nid = "secret"\n')
|
||||
(ind / "results-v2.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"precision_sample": [
|
||||
{
|
||||
"case": "acme-sqli",
|
||||
"site": "src/other.php:3:render",
|
||||
"family": "injection",
|
||||
"verdict": "False, guarded",
|
||||
"reason": "cast",
|
||||
},
|
||||
{
|
||||
"case": "acme-sqli",
|
||||
"site": "src/app.php:4:",
|
||||
"family": "injection",
|
||||
"verdict": "True, privileged",
|
||||
"reason": "admin only",
|
||||
},
|
||||
]
|
||||
}
|
||||
) # fmt: skip
|
||||
)
|
||||
sources = (
|
||||
'version = 1\n\n[[source]]\nid = "population-v2"\nkind = "population"\nfile = "benchmarks/independent/population-v2.toml"\n'
|
||||
'evaluation = ["benchmarks/independent/results-v2.json"]\nrecorded = "2026-09-28"\nstatus = "spent"\nmatching = "declared_sites"\n'
|
||||
'design_informed = true\ncache = "independent"\n\n'
|
||||
'[[source]]\nid = "adjudications"\nkind = "adjudications"\nfiles = ["benchmarks/independent/results-v2.json#precision_sample"]\n'
|
||||
'evaluation = []\nrecorded = "2026-09-28"\nstatus = "recorded"\n\n'
|
||||
'[[source]]\nid = "assumed_benign"\nkind = "history"\nrepositories_from = ["population-v2"]\nrecorded = "2026-09-30"\n'
|
||||
'status = "assumption"\nweight = 0.5\nmin_days_from_fix = 90\nscan_commits = 100\nmax_commits_per_repository = 10\n'
|
||||
'max_files_per_commit = 20\n\n[[excluded]]\nfile = "population-v3-php.toml"\nreason = "reserved"\n'
|
||||
)
|
||||
(root / "sources.toml").write_text(sources)
|
||||
return root, cache, pins
|
||||
|
||||
|
||||
def _build(tmp_path: Path, monkeypatch: pytest.MonkeyPatch | None = None) -> tuple[L.Build, dict[str, Any], list[str], dict[str, str]]:
|
||||
root, cache, pins = _layout(tmp_path)
|
||||
opened: list[str] = []
|
||||
if monkeypatch is not None:
|
||||
real_read = Path.read_text
|
||||
real_open = open
|
||||
|
||||
def read_text(self: Path, *a: Any, **k: Any) -> str:
|
||||
opened.append(self.name)
|
||||
return real_read(self, *a, **k)
|
||||
|
||||
def tracking_open(file: Any, *a: Any, **k: Any) -> Any:
|
||||
opened.append(Path(str(file)).name)
|
||||
return real_open(file, *a, **k)
|
||||
|
||||
monkeypatch.setattr(Path, "read_text", read_text)
|
||||
monkeypatch.setattr("builtins.open", tracking_open)
|
||||
build, record = L.build_labels(root, sources_path=root / "sources.toml", cache=cache, created="2026-09-30")
|
||||
return build, record, opened, pins
|
||||
|
||||
|
||||
def test_population_positives_negatives_and_the_untouched_function(tmp_path: Path) -> None:
|
||||
build, _, _, pins = _build(tmp_path)
|
||||
index = L.LabelIndex(build.labels)
|
||||
show = index.lookup("src/app.php::show", "injection")
|
||||
assert show is not None and show.label == 1 and show.design_informed and show.group == "acme/shop" and show.pin == pins["vulnerable"]
|
||||
negative = index.lookup("src/app.php::show", "injection", pin_role="fixed")
|
||||
assert negative is not None and negative.label == 0 and negative.pin == pins["fixed"]
|
||||
assert index.lookup("src/app.php::helper", "injection").label == 1 # type: ignore[union-attr]
|
||||
assert index.lookup("src/app.php::helper", "injection", pin_role="fixed") is None, "the fix did not touch helper: not a negative"
|
||||
assert index.lookup("src/app.php::nowhere", "injection") is None, "an unmatched candidate is unlabelled, never negative"
|
||||
deltas = {(r.candidate, r.direction, r.label) for r in build.labels if r.unit == "delta" and r.source == "population-v2"}
|
||||
assert ("src/app.php::show", "introduce", 1) in deltas and ("src/app.php::show", "repair", 0) in deltas
|
||||
benign = [r for r in build.labels if r.source == "benign_control"]
|
||||
assert [r.candidate for r in benign] == ["src/app.php::helper"] and benign[0].conditional == L.CONDITIONAL
|
||||
assert all(r.frameworks == () for r in build.labels if r.source == "population-v2"), "composer.json names no framework"
|
||||
|
||||
|
||||
def test_adjudications_map_verdicts_and_resolve_a_missing_function(tmp_path: Path) -> None:
|
||||
build, _, _, _ = _build(tmp_path)
|
||||
rows = {r.candidate: r for r in build.labels if r.source == "adjudications"}
|
||||
assert rows["src/other.php::render"].label == 0 and "guarded" in rows["src/other.php::render"].evidence
|
||||
assert rows["src/app.php::show"].label == 1 and rows["src/app.php::show"].privileged, "the line resolved to its function"
|
||||
|
||||
|
||||
def test_v3_is_never_opened_and_an_unlisted_source_cannot_label(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
build, _, opened, _ = _build(tmp_path, monkeypatch)
|
||||
assert build.labels and "population-v2.toml" in opened
|
||||
assert "population-v3-php.toml" not in opened, "the reserved population must never be read"
|
||||
sources = L.load_sources(tmp_path / "repo" / "sources.toml")
|
||||
with pytest.raises(L.LabelSourceError, match="not a label source"):
|
||||
sources.get("population-v3")
|
||||
guard = L.SourceGuard(sources, tmp_path / "repo")
|
||||
with pytest.raises(L.LabelSourceError, match="excluded"):
|
||||
guard.text("benchmarks/independent/population-v3-php.toml")
|
||||
with pytest.raises(L.LabelSourceError, match="not a listed label source"):
|
||||
guard.text("benchmarks/independent/results-v1.json")
|
||||
assert "population-v3-php.toml" not in opened
|
||||
|
||||
|
||||
def test_a_source_naming_an_excluded_file_is_refused(tmp_path: Path) -> None:
|
||||
(tmp_path / "s.toml").write_text(
|
||||
'[[source]]\nid = "v3"\nkind = "population"\nfile = "b/population-v3-php.toml"\nrecorded = "x"\nstatus = "x"\n'
|
||||
'[[excluded]]\nfile = "population-v3-php.toml"\n'
|
||||
)
|
||||
with pytest.raises(L.LabelSourceError, match="excluded"):
|
||||
L.load_sources(tmp_path / "s.toml")
|
||||
|
||||
|
||||
def test_a_source_without_its_evaluation_record_cannot_label(tmp_path: Path) -> None:
|
||||
root, cache, _ = _layout(tmp_path)
|
||||
(root / "benchmarks" / "independent" / "results-v2.json").unlink()
|
||||
with pytest.raises(L.LabelSourceError, match="evaluation record"):
|
||||
L.build_labels(root, sources_path=root / "sources.toml", cache=cache)
|
||||
|
||||
|
||||
def test_assumed_benign_is_mined_separately_and_the_filter_is_multilingual(tmp_path: Path) -> None:
|
||||
build, record, _, pins = _build(tmp_path)
|
||||
mined = [r for r in build.labels if r.source == "assumed_benign"]
|
||||
assert {r.pin for r in mined} == {pins["ordinary"]}, "only the ordinary commit far from the fix, whose files no later fix touched"
|
||||
assert all(r.weight == 0.5 and r.conditional == L.CONDITIONAL and r.label == 0 for r in mined)
|
||||
history = record["assumed_benign_history"]
|
||||
assert history["rejected"]["other_language"] == 1, "the Chinese 'fix IDOR' commit is caught with no English word"
|
||||
assert history["rejected"]["near_fix"] >= 1 and history["rejected"]["later fix touches its files"] >= 1
|
||||
assert not [r for r in L.verified(build.labels) if r.source == "assumed_benign"], "never merged with verified negatives"
|
||||
assert "assumed_benign" in record["conditional_negatives"] and "assumed_benign" not in record["verified_pin_labels"]["by_source"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"message,reason",
|
||||
[
|
||||
("修复越权访问问题", "other_language"),
|
||||
("修复 IDOR", "english"),
|
||||
("脆弱性の修正", "other_language"),
|
||||
("보안 취약점 수정", "other_language"),
|
||||
("исправлена уязвимость", "other_language"),
|
||||
("Sicherheitslücke geschlossen", "other_language"),
|
||||
("corrige une faille", "other_language"),
|
||||
("Corrige una falla de seguridad", "other_language"),
|
||||
("Fix CVE-2024-1234", "reference"),
|
||||
("see GHSA-abcd-efgh-ijkl", "reference"),
|
||||
("Escape the title (CWE-79)", "reference"),
|
||||
("Prevent XSS in comments", "english"),
|
||||
("sanitize input", "english"),
|
||||
("Bump lodash from 4.17.20 to 4.17.21", "dependency"),
|
||||
("fix: validate log file name more thoroughly", "english"),
|
||||
("add password masking to the keyboard", "english"),
|
||||
("增加文件名校验", "other_language"),
|
||||
("Update README", None),
|
||||
("修复拼写错误", None),
|
||||
("重构代码结构", None),
|
||||
("Añadir traducción al español", None),
|
||||
("Refactor the settings page", None),
|
||||
],
|
||||
)
|
||||
def test_security_filter(message: str, reason: str | None) -> None:
|
||||
assert L.security_reason(message) == reason
|
||||
|
||||
|
||||
def test_repository_groups_merge_urls_forks_and_advisories() -> None:
|
||||
assert L.repo_name("https://github.com/Acme/Shop.git") == L.repo_name("[email protected]:acme/shop") == "acme/shop"
|
||||
assert L.repo_name("github.com/acme/shop/") == "acme/shop"
|
||||
groups = L.Groups({"someone/renamed": "acme/shop"})
|
||||
groups.add("https://github.com/acme/shop", advisory_ids={"CVE-2024-0001"}, commits=("a" * 40,))
|
||||
groups.add("https://github.com/fork/shop-mirror", advisory_ids={"CVE-2024-0001"})
|
||||
groups.add("https://gitlab.com/other/copy", commits=("a" * 40,))
|
||||
groups.add("someone/renamed")
|
||||
groups.add("unrelated/shop")
|
||||
assert groups.group("fork/shop-mirror") == groups.group("other/copy") == groups.group("someone/renamed") == groups.group("acme/shop")
|
||||
assert groups.group("unrelated/shop") != groups.group("acme/shop"), "a shared name alone does not merge"
|
||||
assert groups.merged == {"advisory": 1, "commit": 1, "alias": 1} and groups.count() == 2
|
||||
assert L.advisories("fix for cve-2024-0001 and GHSA-mh7h-gpmx-fggj") == {"CVE-2024-0001", "GHSA-MH7H-GPMX-FGGJ"}
|
||||
|
||||
|
||||
def _pair(name: str, *, tier: str = "advisory", unscorable: str | None = None, function: str | None = "f", repo: str = "") -> PairCase:
|
||||
expected = (ExpectedFinding(cwe="CWE-89", vulnerability_class="sqli", path="a.py", evidence="e", function=function),)
|
||||
return PairCase(
|
||||
name=name, slice="vfc", language="python", origin="o", vuln_file=Path("v"), fixed_file=Path("f"), relpath="a.py",
|
||||
expected=expected, min_recall=1.0, fix_policy="p", repo=repo, cve="CVE-2024-0001", review_tier=tier, unscorable=unscorable,
|
||||
) # fmt: skip
|
||||
|
||||
|
||||
def test_pairs_exclude_unscorable_and_title_and_group_by_repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
(tmp_path / "pairs").mkdir()
|
||||
(tmp_path / "pairs" / "catalog.toml").write_text("")
|
||||
(tmp_path / "s.toml").write_text(
|
||||
'[[source]]\nid = "pairs"\nkind = "pairs"\nfile = "pairs/catalog.toml"\nrecorded = "x"\nstatus = "x"\nexclude_tiers = ["title"]\n'
|
||||
)
|
||||
cases = (
|
||||
_pair("ok", repo="https://github.com/o/r"), _pair("twin", unscorable="identical_twin"), _pair("titled", tier="title"),
|
||||
_pair("nofn", function=None), _pair("fork", repo="https://github.com/x/r-fork"),
|
||||
) # fmt: skip
|
||||
monkeypatch.setattr("openultrasast.pairs.load_pair_catalog", lambda path: cases)
|
||||
build, record = L.build_labels(tmp_path, sources_path=tmp_path / "s.toml", cache=tmp_path, assumed_benign=False)
|
||||
refs = {r.source_ref for r in build.labels}
|
||||
assert refs == {"ok", "fork"}
|
||||
assert {r.group for r in build.labels} == {"o/r"}, "the fork shares the CVE, so it is one repository group"
|
||||
assert {(r.unit, r.label, r.pin_role) for r in build.labels if r.source_ref == "ok"} == {
|
||||
("pin", 1, "vulnerable"),
|
||||
("delta", 1, "vulnerable"),
|
||||
("pin", 0, "fixed"),
|
||||
("delta", 0, "fixed"),
|
||||
}
|
||||
assert record["skipped"] == {"pairs: label names no function": 1, "pairs: title tier (sensitivity arm)": 1, "pairs: unscorable pair": 1}
|
||||
family = record["verified_pin_labels"]["by_family"]["injection"]
|
||||
assert family["positive_groups"] == family["negative_groups"] == 1 and family["model"] == "insufficient"
|
||||
titled, _ = L.build_labels(tmp_path, sources_path=tmp_path / "s.toml", cache=tmp_path, assumed_benign=False, include_title=True)
|
||||
assert "titled" in {r.source_ref for r in titled.labels}
|
||||
|
||||
|
||||
def test_the_snapshot_carries_counts_only(tmp_path: Path) -> None:
|
||||
_, record, _, pins = _build(tmp_path)
|
||||
text = json.dumps(record)
|
||||
for identity in ("acme", "shop", "show", "helper", "src/app.php", *pins.values()):
|
||||
assert identity not in text, f"{identity} leaked into the committed snapshot"
|
||||
assert record["verified_pin_labels"]["by_family"]["injection"]["positive_groups"] == 1
|
||||
|
||||
|
||||
def test_frameworks_from_manifests_and_plugin_header() -> None:
|
||||
assert L.frameworks_from({"requirements.txt": "Django==4.2\nrequests"}) == ("django", "requests")
|
||||
assert L.frameworks_from({}, ["<?php\n/*\n * Plugin Name: Shop\n */"]) == ("wordpress",)
|
||||
assert L.frameworks_from({"package.json": '{"dependencies": {"express-session": "1"}}'}) == ()
|
||||
Reference in New Issue
Block a user