From 6145bd4f387cde3168795639eaf0d55e1017bcb4 Mon Sep 17 00:00:00 2001 From: "secopsai-blog-ops[bot]" <115749095+Techris93@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:13:30 +0300 Subject: [PATCH] feat: add auditable research promotion policy --- docs/research-discovery.md | 4 ++ secopsai/cli.py | 26 +++++++++ secopsai/research_discovery.py | 73 +++++++++++++++++++++++++ soc_store.py | 26 +++++++++ tests/test_research_promotion_policy.py | 56 +++++++++++++++++++ 5 files changed, 185 insertions(+) create mode 100644 tests/test_research_promotion_policy.py diff --git a/docs/research-discovery.md b/docs/research-discovery.md index 8e3ad6a..9c9c634 100644 --- a/docs/research-discovery.md +++ b/docs/research-discovery.md @@ -97,12 +97,16 @@ secopsai research monitor create --ecosystem nuget --watchlist-id WL-... --inter secopsai research monitor list secopsai research monitor run-due --limit 25 secopsai research candidate list +secopsai research candidate promotion-policy --ecosystem all +secopsai research candidate run-promotion-policy --ecosystem all secopsai research campaign correlate secopsai research campaign list ``` The dashboard provides the same actions under Research Discovery. Use **Add watchlist**, **Create monitor**, **Run due monitors**, **Compare exact packages**, and **Correlate campaigns**. Protected writes require the research action token. +Candidate promotion is disabled by default. Configure it with `research candidate promotion-policy --set`, then run `run-promotion-policy` without `--apply` to preview every decision. The policy evaluates a minimum score, evidence-reference count, optional publisher evidence, and ecosystem scope. Adding `--apply` may create draft Research Cases only; it never records a malicious verdict. Each promotion persists the candidate, exact policy, reasons, actor, resulting case ID, and event timestamp for audit and rollback review. + On macOS, install the local due-monitor trigger as a background service: ```bash diff --git a/secopsai/cli.py b/secopsai/cli.py index deee1b6..638f667 100644 --- a/secopsai/cli.py +++ b/secopsai/cli.py @@ -137,6 +137,7 @@ create_monitor as create_research_monitor, create_watchlist as create_research_watchlist, get_candidate as get_research_candidate, + get_promotion_policy as get_research_promotion_policy, ingest_registry_metadata, list_candidates as list_research_candidates, list_alerts as list_research_alerts, @@ -146,6 +147,8 @@ run_monitor as run_research_monitor, run_due_monitors as run_due_research_monitors, recover_stale_monitor_runs, + run_promotion_policy as run_research_promotion_policy, + set_promotion_policy as set_research_promotion_policy, ) from secopsai.research_surveillance import ( collector_status as registry_collector_status, @@ -1314,6 +1317,22 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: candidate_show = research_candidate_sub.add_parser("show") candidate_show.add_argument("candidate_id") candidate_show.add_argument("--db-path", default=None) + candidate_policy = research_candidate_sub.add_parser("promotion-policy") + candidate_policy.add_argument("--ecosystem", default="all") + candidate_policy.add_argument("--set", action="store_true") + candidate_policy.add_argument("--enabled", choices=["true", "false"], default=None) + candidate_policy.add_argument("--score-threshold", type=float, default=90.0) + candidate_policy.add_argument("--minimum-evidence", type=int, default=2) + candidate_policy.add_argument("--require-publisher", action="store_true") + candidate_policy.add_argument("--mode", choices=["review_only", "draft_case"], default="draft_case") + candidate_policy.add_argument("--actor", default="operator") + candidate_policy.add_argument("--db-path", default=None) + candidate_promote = research_candidate_sub.add_parser("run-promotion-policy") + candidate_promote.add_argument("--ecosystem", default="all") + candidate_promote.add_argument("--apply", action="store_true") + candidate_promote.add_argument("--actor", default="operator") + candidate_promote.add_argument("--limit", type=int, default=100) + candidate_promote.add_argument("--db-path", default=None) research_collect = research_sub.add_parser("collect", help="Run global registry feed collectors") research_collect_sub = research_collect.add_subparsers(dest="research_collect_cmd", required=True) @@ -2404,6 +2423,13 @@ def _run_research_automation_command(args: argparse.Namespace) -> int: elif args.research_cmd == "candidate": if args.research_candidate_cmd == "list": payload = {"candidates": list_research_candidates(status=args.status, ecosystem=args.ecosystem, limit=args.limit, db_path=args.db_path)} + elif args.research_candidate_cmd == "promotion-policy": + if args.set: + payload = set_research_promotion_policy(ecosystem=args.ecosystem, enabled=args.enabled == "true", score_threshold=args.score_threshold, minimum_evidence=args.minimum_evidence, require_publisher=args.require_publisher, mode=args.mode, actor=args.actor, db_path=args.db_path) + else: + payload = get_research_promotion_policy(ecosystem=args.ecosystem, db_path=args.db_path) + elif args.research_candidate_cmd == "run-promotion-policy": + payload = run_research_promotion_policy(ecosystem=args.ecosystem, apply=args.apply, actor=args.actor, limit=args.limit, db_path=args.db_path) else: payload = get_research_candidate(args.candidate_id, db_path=args.db_path) elif args.research_cmd == "collect": diff --git a/secopsai/research_discovery.py b/secopsai/research_discovery.py index c5dcaa4..8ddfb3f 100644 --- a/secopsai/research_discovery.py +++ b/secopsai/research_discovery.py @@ -26,6 +26,7 @@ WATCH_TYPES = {"package", "namespace", "publisher", "brand", "repository", "organization"} PRIORITIES = {"low", "normal", "high", "critical"} STATUSES = {"active", "paused", "archived", "new", "review", "dismissed", "promoted"} +PROMOTION_MODES = {"review_only", "draft_case"} def _id(prefix: str) -> str: @@ -516,6 +517,78 @@ def get_candidate(candidate_id: str, *, db_path: Optional[str] = None) -> Dict[s return item +def get_promotion_policy(*, ecosystem: str = "all", db_path: Optional[str] = None) -> Dict[str, Any]: + """Return the durable, auditable candidate-promotion policy.""" + soc_store.init_db(db_path) + ecosystem = normalize_identifier("", ecosystem or "all") + with closing(soc_store.connect(db_path)) as connection: + row = connection.execute("SELECT * FROM research_promotion_policies WHERE ecosystem = ?", (ecosystem,)).fetchone() + if row: + return dict(row) + return {"ecosystem": ecosystem, "enabled": 0, "score_threshold": 90.0, "minimum_evidence": 2, "require_publisher": 0, "mode": "draft_case", "updated_by": "", "created_at": None, "updated_at": None} + + +def set_promotion_policy(*, ecosystem: str = "all", enabled: bool = False, score_threshold: float = 90.0, minimum_evidence: int = 2, require_publisher: bool = False, mode: str = "draft_case", actor: str = "operator", db_path: Optional[str] = None) -> Dict[str, Any]: + soc_store.init_db(db_path) + ecosystem = normalize_identifier("", ecosystem or "all") + if ecosystem != "all" and ecosystem not in CAPABILITIES: + raise ValueError("unsupported promotion-policy ecosystem") + threshold = float(score_threshold) + if not 70 <= threshold <= 100: + raise ValueError("score threshold must be between 70 and 100") + evidence_count = max(1, min(int(minimum_evidence), 8)) + if mode not in PROMOTION_MODES: + raise ValueError("unsupported promotion mode") + now = _now() + with closing(soc_store.connect(db_path)) as connection: + connection.execute("""INSERT INTO research_promotion_policies + (ecosystem, enabled, score_threshold, minimum_evidence, require_publisher, mode, updated_by, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(ecosystem) DO UPDATE SET enabled=excluded.enabled, score_threshold=excluded.score_threshold, + minimum_evidence=excluded.minimum_evidence, require_publisher=excluded.require_publisher, + mode=excluded.mode, updated_by=excluded.updated_by, updated_at=excluded.updated_at""", + (ecosystem, int(bool(enabled)), threshold, evidence_count, int(bool(require_publisher)), mode, str(actor or "operator")[:160], now, now)) + connection.commit() + return get_promotion_policy(ecosystem=ecosystem, db_path=db_path) + + +def run_promotion_policy(*, ecosystem: str = "all", apply: bool = False, actor: str = "operator", limit: int = 100, db_path: Optional[str] = None) -> Dict[str, Any]: + """Evaluate candidates deterministically and optionally create draft cases. + + Promotion never records a malicious verdict. It creates an auditable draft + investigation only when the configured evidence gates pass. + """ + policy = get_promotion_policy(ecosystem=ecosystem, db_path=db_path) + candidates = list_candidates(status="new", ecosystem=None if policy["ecosystem"] == "all" else policy["ecosystem"], limit=limit, db_path=db_path) + decisions: List[Dict[str, Any]] = [] + for candidate in candidates: + evidence = candidate.get("evidence") if isinstance(candidate.get("evidence"), dict) else {} + evidence_refs = [key for key in ("metadata_url", "artifact_url", "publisher") if evidence.get(key)] + reasons = [] + if float(candidate.get("score") or 0) < float(policy["score_threshold"]): + reasons.append("score_below_threshold") + if len(evidence_refs) < int(policy["minimum_evidence"]): + reasons.append("insufficient_evidence") + if policy["require_publisher"] and not evidence.get("publisher"): + reasons.append("publisher_missing") + eligible = not reasons and bool(policy["enabled"]) + decision = {"candidate_id": candidate["candidate_id"], "eligible": eligible, "score": candidate["score"], "evidence_count": len(evidence_refs), "reasons": reasons or (["policy_disabled"] if not policy["enabled"] else ["eligible"]), "case_id": candidate.get("case_id")} + if apply and eligible and policy["mode"] == "draft_case" and not candidate.get("case_id"): + from secopsai.research_cases import add_evidence, add_subject, create_case + case = create_case(title=f"Investigate {candidate['ecosystem']} package {candidate['package']}", summary=f"Automatically promoted as a draft investigation after deterministic policy checks. Similarity score: {candidate['score']}. This is a lead, not a maliciousness verdict.", case_type="typosquatting", severity="medium", confidence=min(int(float(candidate["score"])), 80), owner=actor, metadata={"candidate_id": candidate["candidate_id"], "promotion_policy": policy}, db_path=db_path) + add_subject(case["case_id"], subject_type="package", name=candidate["package"], ecosystem=candidate["ecosystem"], version=candidate.get("version") or "", publisher=evidence.get("publisher") or "", registry_state="available", metadata={"candidate_id": candidate["candidate_id"], "reference_identifier": candidate.get("reference_identifier")}, db_path=db_path, actor=actor) + if evidence.get("metadata_url"): + add_evidence(case["case_id"], evidence_type="registry_metadata", title="Registry metadata captured during candidate discovery", locator=evidence["metadata_url"], provenance="official registry monitor", notes=candidate.get("reason") or "", metadata={"candidate_id": candidate["candidate_id"]}, db_path=db_path, actor=actor) + with closing(soc_store.connect(db_path)) as connection: + connection.execute("UPDATE research_candidates SET status = 'promoted', case_id = ? WHERE candidate_id = ?", (case["case_id"], candidate["candidate_id"])) + connection.execute("INSERT OR IGNORE INTO research_promotion_events (event_id, candidate_id, policy_ecosystem, decision, reasons_json, case_id, actor, created_at) VALUES (?, ?, ?, 'draft_case_created', ?, ?, ?, ?)", (_id("RPE"), candidate["candidate_id"], policy["ecosystem"], _json(decision["reasons"]), case["case_id"], actor, _now())) + connection.commit() + decision["case_id"] = case["case_id"] + decision["applied"] = True + decisions.append(decision) + return {"schema_version": "secopsai.research.promotion-policy.v1", "policy": policy, "apply": bool(apply), "evaluated": len(decisions), "eligible": sum(bool(item["eligible"]) for item in decisions), "promoted": sum(bool(item.get("applied")) for item in decisions), "decisions": decisions} + + def create_candidate_alert(candidate: Dict[str, Any], *, db_path: Optional[str] = None) -> Dict[str, Any]: """Create a deduplicated alert for a high-confidence candidate.""" soc_store.init_db(db_path) diff --git a/soc_store.py b/soc_store.py index da2a472..e6b9ba0 100644 --- a/soc_store.py +++ b/soc_store.py @@ -721,6 +721,32 @@ def init_db(db_path: str | None = None) -> None: FOREIGN KEY (case_id) REFERENCES research_cases (case_id) ON DELETE SET NULL ); + CREATE TABLE IF NOT EXISTS research_promotion_policies ( + ecosystem TEXT PRIMARY KEY, + enabled INTEGER NOT NULL DEFAULT 0, + score_threshold REAL NOT NULL DEFAULT 90, + minimum_evidence INTEGER NOT NULL DEFAULT 2, + require_publisher INTEGER NOT NULL DEFAULT 0, + mode TEXT NOT NULL DEFAULT 'draft_case', + updated_by TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS research_promotion_events ( + event_id TEXT PRIMARY KEY, + candidate_id TEXT NOT NULL, + policy_ecosystem TEXT NOT NULL, + decision TEXT NOT NULL, + reasons_json TEXT NOT NULL, + case_id TEXT, + actor TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(candidate_id, policy_ecosystem, decision), + FOREIGN KEY (candidate_id) REFERENCES research_candidates (candidate_id) ON DELETE CASCADE, + FOREIGN KEY (case_id) REFERENCES research_cases (case_id) ON DELETE SET NULL + ); + CREATE TABLE IF NOT EXISTS research_campaigns ( campaign_id TEXT PRIMARY KEY, title TEXT NOT NULL, diff --git a/tests/test_research_promotion_policy.py b/tests/test_research_promotion_policy.py new file mode 100644 index 0000000..fb16205 --- /dev/null +++ b/tests/test_research_promotion_policy.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import soc_store +from secopsai.research_discovery import get_promotion_policy, run_promotion_policy, set_promotion_policy + + +def _candidate(db_path: str, *, score: float = 96.0) -> str: + soc_store.init_db(db_path) + candidate_id = "CAN-POLICYTEST0001" + with soc_store.connect(db_path) as connection: + connection.execute( + """INSERT INTO research_candidates + (candidate_id, event_id, watchlist_id, ecosystem, package, version, reference_identifier, + score, score_components_json, reason, status, case_id, evidence_json, first_seen, + last_seen, algorithm_version) + VALUES (?, NULL, NULL, 'nuget', 'Braintree.Net.Test', '1.0.0', 'Braintree.Net', ?, '{}', + 'test similarity candidate', 'new', NULL, ?, '2026-07-29T00:00:00Z', + '2026-07-29T00:00:00Z', 'similarity-1')""", + (candidate_id, score, json.dumps({"metadata_url": "https://api.nuget.org/v3/registration5-gz-semver2/braintree.net.test/index.json", "artifact_url": "https://api.nuget.org/v3-flatcontainer/braintree.net.test/1.0.0/braintree.net.test.1.0.0.nupkg", "publisher": "unexpected"})), + ) + connection.commit() + return candidate_id + + +def test_promotion_policy_is_disabled_and_preview_only_by_default() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + db_path = str(Path(temp_dir) / "soc.db") + _candidate(db_path) + policy = get_promotion_policy(ecosystem="nuget", db_path=db_path) + assert policy["enabled"] == 0 + preview = run_promotion_policy(ecosystem="nuget", db_path=db_path) + assert preview["promoted"] == 0 + assert preview["decisions"][0]["eligible"] is False + + +def test_enabled_policy_creates_auditable_draft_case_without_a_verdict() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + db_path = str(Path(temp_dir) / "soc.db") + candidate_id = _candidate(db_path) + set_promotion_policy(ecosystem="nuget", enabled=True, score_threshold=90, minimum_evidence=2, require_publisher=True, actor="test-operator", db_path=db_path) + result = run_promotion_policy(ecosystem="nuget", apply=True, actor="test-operator", db_path=db_path) + assert result["promoted"] == 1 + assert result["decisions"][0]["case_id"].startswith("RSC-") + with soc_store.connect(db_path) as connection: + candidate = connection.execute("SELECT status, case_id FROM research_candidates WHERE candidate_id = ?", (candidate_id,)).fetchone() + case = connection.execute("SELECT status, payload_json FROM research_cases WHERE case_id = ?", (candidate["case_id"],)).fetchone() + event = connection.execute("SELECT decision, actor FROM research_promotion_events WHERE candidate_id = ?", (candidate_id,)).fetchone() + assert candidate["status"] == "promoted" + assert case["status"] == "draft" + assert "promotion_policy" in case["payload_json"] + assert event["decision"] == "draft_case_created" + assert event["actor"] == "test-operator"