From b5734617db6c0f73b89ace79abef1a2c4ec8be87 Mon Sep 17 00:00:00 2001 From: "secopsai-blog-ops[bot]" <115749095+Techris93@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:23:29 +0300 Subject: [PATCH 1/2] fix: make candidate promotion atomic and corroborated --- secopsai/research_discovery.py | 51 +++++++++++++++++++------ tests/test_research_promotion_policy.py | 20 ++++++++++ 2 files changed, 60 insertions(+), 11 deletions(-) diff --git a/secopsai/research_discovery.py b/secopsai/research_discovery.py index 8ddfb3f..01a1d34 100644 --- a/secopsai/research_discovery.py +++ b/secopsai/research_discovery.py @@ -563,27 +563,56 @@ def run_promotion_policy(*, ecosystem: str = "all", apply: bool = False, actor: 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)] + evidence_values = set() + for key in ("metadata_url", "artifact_url"): + value = str(evidence.get(key) or "").strip().lower() + if value: + evidence_values.add(value) + publisher_value = normalize_identifier("", evidence.get("publisher") or "") + if publisher_value: + evidence_values.add(f"publisher:{publisher_value}") 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"]): + if len(evidence_values) < 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")} + decision = {"candidate_id": candidate["candidate_id"], "eligible": eligible, "score": candidate["score"], "evidence_count": len(evidence_values), "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) + case_id = f"RSC-{hashlib.sha256(candidate['candidate_id'].encode()).hexdigest()[:12].upper()}" + subject_id = f"SUB-{hashlib.sha256(f'{case_id}|package|{candidate["ecosystem"]}|{candidate["package"]}|{candidate.get("version") or ""}'.encode()).hexdigest()[:16].upper()}" + evidence_id = f"EVD-{hashlib.sha256(f'{case_id}|registry_metadata|{evidence.get("metadata_url") or candidate["candidate_id"]}'.encode()).hexdigest()[:16].upper()}" + now = _now() 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.execute("BEGIN IMMEDIATE") + current = connection.execute("SELECT status, case_id FROM research_candidates WHERE candidate_id = ?", (candidate["candidate_id"],)).fetchone() + if not current or current["status"] != "new" or current["case_id"]: + connection.rollback() + decision["eligible"] = False + decision["reasons"] = ["already_promoted_or_claimed"] + decisions.append(decision) + continue + connection.execute("""INSERT OR IGNORE INTO research_cases + (case_id, title, summary, case_type, severity, confidence, status, owner, disclosure_status, + embargo_until, created_at, updated_at, closed_at, published_at, payload_json) + VALUES (?, ?, ?, 'typosquatting', 'medium', ?, 'draft', ?, 'not_started', NULL, ?, ?, NULL, NULL, ?)""", + (case_id, f"Investigate {candidate['ecosystem']} package {candidate['package']}", f"Automatically promoted as a draft investigation after deterministic policy checks. Similarity score: {candidate['score']}. This is a lead, not a maliciousness verdict.", min(int(float(candidate["score"])), 80), actor, now, now, _json({"candidate_id": candidate["candidate_id"], "promotion_policy": policy}))) + connection.execute("""INSERT OR IGNORE INTO research_subjects + (subject_id, case_id, subject_type, ecosystem, name, version, publisher, status, metadata_json, created_at) + VALUES (?, ?, 'package', ?, ?, ?, ?, 'active', ?, ?)""", + (subject_id, case_id, candidate["ecosystem"], candidate["package"], candidate.get("version") or "", evidence.get("publisher") or "", _json({"candidate_id": candidate["candidate_id"], "reference_identifier": candidate.get("reference_identifier")}), now)) + if evidence.get("metadata_url"): + connection.execute("""INSERT OR IGNORE INTO research_evidence + (evidence_id, case_id, evidence_type, title, locator, sha256, provenance, notes, status, collected_at, created_at, metadata_json) + VALUES (?, ?, 'registry_metadata', 'Registry metadata captured during candidate discovery', ?, '', 'official registry monitor', ?, 'active', ?, ?, ?)""", + (evidence_id, case_id, evidence["metadata_url"], candidate.get("reason") or "", now, now, _json({"candidate_id": candidate["candidate_id"]}))) + connection.execute("INSERT INTO research_case_events (case_id, event_type, actor, message, data_json, created_at) VALUES (?, 'case_created', ?, 'Research case created by candidate promotion policy.', ?, ?)", (case_id, actor, _json({"candidate_id": candidate["candidate_id"], "promotion_policy": policy}), now)) + connection.execute("UPDATE research_candidates SET status = 'promoted', case_id = ? WHERE candidate_id = ? AND status = 'new' AND case_id IS NULL", (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_id, actor, now)) connection.commit() - decision["case_id"] = case["case_id"] + decision["case_id"] = 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} diff --git a/tests/test_research_promotion_policy.py b/tests/test_research_promotion_policy.py index fb16205..1799372 100644 --- a/tests/test_research_promotion_policy.py +++ b/tests/test_research_promotion_policy.py @@ -54,3 +54,23 @@ def test_enabled_policy_creates_auditable_draft_case_without_a_verdict() -> None assert "promotion_policy" in case["payload_json"] assert event["decision"] == "draft_case_created" assert event["actor"] == "test-operator" + second = run_promotion_policy(ecosystem="nuget", apply=True, actor="test-operator", db_path=db_path) + assert second["promoted"] == 0 + with soc_store.connect(db_path) as connection: + assert connection.execute("SELECT COUNT(*) FROM research_cases").fetchone()[0] == 1 + assert connection.execute("SELECT COUNT(*) FROM research_promotion_events").fetchone()[0] == 1 + + +def test_duplicate_urls_count_as_one_evidence_reference() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + db_path = str(Path(temp_dir) / "soc.db") + candidate_id = _candidate(db_path) + duplicate = "https://api.nuget.org/v3/registration5-gz-semver2/braintree.net.test/index.json" + with soc_store.connect(db_path) as connection: + connection.execute("UPDATE research_candidates SET evidence_json = ? WHERE candidate_id = ?", (json.dumps({"metadata_url": duplicate, "artifact_url": duplicate, "publisher": ""}), candidate_id)) + connection.commit() + set_promotion_policy(ecosystem="nuget", enabled=True, score_threshold=90, minimum_evidence=2, db_path=db_path) + preview = run_promotion_policy(ecosystem="nuget", db_path=db_path) + assert preview["eligible"] == 0 + assert preview["decisions"][0]["evidence_count"] == 1 + assert "insufficient_evidence" in preview["decisions"][0]["reasons"] From a3b75f5ba645676c32adeccb4c4861eeb9a9e005 Mon Sep 17 00:00:00 2001 From: "secopsai-blog-ops[bot]" <115749095+Techris93@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:25:30 +0300 Subject: [PATCH 2/2] fix: retain Python 3.10 promotion compatibility --- secopsai/research_discovery.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/secopsai/research_discovery.py b/secopsai/research_discovery.py index 01a1d34..d9eceee 100644 --- a/secopsai/research_discovery.py +++ b/secopsai/research_discovery.py @@ -582,8 +582,10 @@ def run_promotion_policy(*, ecosystem: str = "all", apply: bool = False, actor: decision = {"candidate_id": candidate["candidate_id"], "eligible": eligible, "score": candidate["score"], "evidence_count": len(evidence_values), "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"): case_id = f"RSC-{hashlib.sha256(candidate['candidate_id'].encode()).hexdigest()[:12].upper()}" - subject_id = f"SUB-{hashlib.sha256(f'{case_id}|package|{candidate["ecosystem"]}|{candidate["package"]}|{candidate.get("version") or ""}'.encode()).hexdigest()[:16].upper()}" - evidence_id = f"EVD-{hashlib.sha256(f'{case_id}|registry_metadata|{evidence.get("metadata_url") or candidate["candidate_id"]}'.encode()).hexdigest()[:16].upper()}" + subject_stable = "|".join((case_id, "package", str(candidate["ecosystem"]), str(candidate["package"]), str(candidate.get("version") or ""))) + subject_id = f"SUB-{hashlib.sha256(subject_stable.encode()).hexdigest()[:16].upper()}" + evidence_stable = "|".join((case_id, "registry_metadata", str(evidence.get("metadata_url") or candidate["candidate_id"]))) + evidence_id = f"EVD-{hashlib.sha256(evidence_stable.encode()).hexdigest()[:16].upper()}" now = _now() with closing(soc_store.connect(db_path)) as connection: connection.execute("BEGIN IMMEDIATE")