From 7feb1e3b6e2a1034d09ec248a67e70951049c9c1 Mon Sep 17 00:00:00 2001 From: "secopsai-blog-ops[bot]" <115749095+Techris93@users.noreply.github.com> Date: Mon, 27 Jul 2026 05:40:44 +0300 Subject: [PATCH] fix: reuse verified artifacts after registry removal --- docs/implementation-checkpoints.md | 2 + docs/research-automation.md | 2 + secopsai/research_intake.py | 29 +++++++++ secopsai/research_pipeline.py | 96 ++++++++++++++++++++++++++++-- tests/test_research_pipeline.py | 38 ++++++++++++ 5 files changed, 163 insertions(+), 4 deletions(-) diff --git a/docs/implementation-checkpoints.md b/docs/implementation-checkpoints.md index 9953bb0..87bed8a 100644 --- a/docs/implementation-checkpoints.md +++ b/docs/implementation-checkpoints.md @@ -18,6 +18,8 @@ The remaining human gates are external sandbox submission, external disclosure d Follow-up: bridge service installation now persists the selected OpenCodex provider/model identifier in the fixed service command. This prevents an autonomous background worker from silently reverting to a provider default after the browser closes or the workstation restarts. The service file still contains no provider credentials. +Live acceptance exposed a retracted-package recovery gap: an exact NuGet version previously collected and hashed had disappeared from the registry. Investigation pipelines now prefer a fresh official-registry collection but may reuse an exact prior quarantine artifact after independently verifying ecosystem, package, version, byte size, SHA-256, regular-file state, and configured size limit. Missing or altered cache entries still fail closed. The recorded step makes registry unavailability and quarantine reuse explicit, and no package code is executed. + ## 107 Hermes Agent v1.0.0 Integration Status: complete in production. diff --git a/docs/research-automation.md b/docs/research-automation.md index 9479092..cc51168 100644 --- a/docs/research-automation.md +++ b/docs/research-automation.md @@ -18,6 +18,8 @@ The normal workflow no longer requires an evidence-bundle export, a file upload, If the bridge or collection step fails, click **Retry from checkpoint**. A new pipeline revision is created, stale proposals are superseded, and the previous revision remains auditable. If comparison was incomplete, enter a verified reference and click **Add reference and rerun analysis**. +If an exact package version is later removed from its registry, the pipeline may reuse a previously collected local quarantine artifact only when its ecosystem, package, version, byte size, and SHA-256 all match. The reuse reason and failed registry retrieval are recorded in the step result. SecOpsAI never substitutes another version or trusts an unverified local file. + The pipeline never executes package code, submits an artifact to an external sandbox, sends external communication, approves publication, or publishes an article. In agent-review mode it may record a bounded case verdict, but only against accepted, pipeline-specific evidence and with all guardrail decisions retained in the audit trail. ## Agent-review mode diff --git a/secopsai/research_intake.py b/secopsai/research_intake.py index 85bb631..171e13b 100644 --- a/secopsai/research_intake.py +++ b/secopsai/research_intake.py @@ -602,6 +602,35 @@ def collect_package_intake( } +def validate_quarantined_intake(result: Dict[str, Any]) -> Dict[str, Any]: + """Verify that a normalized intake still matches its owner-only quarantined artifact.""" + metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} + analysis = result.get("analysis") if isinstance(result.get("analysis"), dict) else {} + quarantine = result.get("quarantine") if isinstance(result.get("quarantine"), dict) else {} + digest = _text(metadata.get("artifact_sha256") or quarantine.get("artifact_id"), 64).lower() + if not re.fullmatch(r"[a-f0-9]{64}", digest): + raise IntakeError("cached intake does not contain a valid artifact hash") + filename = _text(analysis.get("filename") or "artifact", 512) + path = _quarantine_path(digest, filename) + if not path.is_file() or path.is_symlink(): + raise IntakeError("cached quarantined artifact is unavailable") + if path.stat().st_size > MAX_ARTIFACT_BYTES: + raise IntakeError("cached quarantined artifact exceeds the configured size limit") + actual = hashlib.sha256(path.read_bytes()).hexdigest() + if actual != digest: + raise IntakeError("cached quarantined artifact hash changed") + expected_bytes = int(metadata.get("artifact_bytes") or quarantine.get("bytes") or 0) + if expected_bytes and expected_bytes != path.stat().st_size: + raise IntakeError("cached quarantined artifact size changed") + return { + "artifact_sha256": digest, + "artifact_bytes": path.stat().st_size, + "filename": filename, + "verified": True, + "execution_performed": False, + } + + def run_package_intake( *, case_id: str, diff --git a/secopsai/research_pipeline.py b/secopsai/research_pipeline.py index 8d0b1b0..8e1b0db 100644 --- a/secopsai/research_pipeline.py +++ b/secopsai/research_pipeline.py @@ -13,7 +13,12 @@ from secopsai.intelligence_jobs import cancel_job, enqueue_job, get_job from secopsai.research_analysis import compare_intakes from secopsai.research_cases import add_evidence, get_case -from secopsai.research_intake import attach_intake_result, collect_package_intake +from secopsai.research_intake import ( + IntakeError, + attach_intake_result, + collect_package_intake, + validate_quarantined_intake, +) from secopsai.research_workflow import ( build_evidence_matrix, publication_safety_check, @@ -401,7 +406,12 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch suspect_result = _step_result(pipeline, "collect_subject") if not suspect_result: _set_step(pipeline_id, "collect_subject", "running", db_path=db_path) - suspect_result = collect_package_intake(fetcher=fetcher, **config["suspect"]) + suspect_result = _collect_or_reuse_intake( + case_id, + config["suspect"], + fetcher=fetcher, + db_path=db_path, + ) _set_step(pipeline_id, "collect_subject", "succeeded", result=suspect_result, db_path=db_path) _put_review_item( pipeline_id, @@ -418,7 +428,12 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch reference_result: Dict[str, Any] = {} if config.get("reference"): _set_step(pipeline_id, "collect_reference", "running", db_path=db_path) - reference_result = collect_package_intake(fetcher=fetcher, **config["reference"]) + reference_result = _collect_or_reuse_intake( + case_id, + config["reference"], + fetcher=fetcher, + db_path=db_path, + ) _set_step(pipeline_id, "collect_reference", "succeeded", result=reference_result, db_path=db_path) _put_review_item( pipeline_id, @@ -489,6 +504,10 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch "static_collection_complete": True, "comparison_complete": bool(reference_result), "comparison_input_required": not bool(reference_result), + "quarantine_reuse_count": sum( + bool(item.get("reuse")) for item in (suspect_result, reference_result) if item + ), + "registry_collection_degraded": bool(suspect_result.get("reuse") or reference_result.get("reuse")), "ai_jobs_queued": len(AI_STEPS), "human_review_required": True, "raw_artifact_sent_to_ai": False, @@ -529,9 +548,14 @@ def _preliminary_matrix(case_id: str, suspect: Dict[str, Any], reference: Dict[s suspect_meta = suspect.get("metadata") or {} suspect_analysis = suspect.get("analysis") or {} indicators = suspect_analysis.get("indicators") or [] + collection_statement = ( + f"SecOpsAI verified a previously quarantined {_target_label(suspect_meta)} artifact, originally collected from its allowlisted registry source, and " + if suspect.get("reuse") + else f"SecOpsAI collected {_target_label(suspect_meta)} from its allowlisted registry source and " + ) claims = [ { - "statement": f"SecOpsAI collected {_target_label(suspect_meta)} from its allowlisted registry source and recorded SHA-256 {suspect_meta.get('artifact_sha256', 'unavailable')}.", + "statement": f"{collection_statement}recorded SHA-256 {suspect_meta.get('artifact_sha256', 'unavailable')}.", "confidence": 100, "status": "supported", "evidence_refs": ["collect_subject"], @@ -699,6 +723,70 @@ def reconcile_intelligence_job(job: Dict[str, Any], *, db_path: Optional[str] = return get_pipeline(pipeline_id, db_path=db_path) +def _collect_or_reuse_intake( + case_id: str, + target: Dict[str, Any], + *, + fetcher: Any, + db_path: Optional[str], +) -> Dict[str, Any]: + """Prefer fresh registry collection, then reuse exact hash-verified local evidence.""" + try: + return collect_package_intake(fetcher=fetcher, **target) + except IntakeError as registry_error: + cached = _find_verified_cached_intake(case_id, target, db_path=db_path) + if cached is None: + raise + result = json.loads(json.dumps(cached)) + result["reuse"] = { + "mode": "verified_quarantine", + "reason": "official registry collection was unavailable", + "registry_error": _clean(registry_error, 500), + "hash_verified": True, + "execution_performed": False, + } + return result + + +def _find_verified_cached_intake( + case_id: str, + target: Dict[str, Any], + *, + db_path: Optional[str], +) -> Optional[Dict[str, Any]]: + ecosystem = _clean(target.get("ecosystem"), 80).lower() + package = _clean(target.get("package"), 512).lower() + version = _clean(target.get("version"), 160) + if not ecosystem or not package or not version: + return None + with closing(soc_store.connect(db_path)) as connection: + rows = connection.execute( + """SELECT s.result_json + FROM research_pipeline_steps s + JOIN research_pipeline_runs p ON p.pipeline_id = s.pipeline_id + WHERE p.case_id = ? AND s.step_key IN ('collect_subject', 'collect_reference') + AND s.status = 'succeeded' + ORDER BY s.completed_at DESC + LIMIT 50""", + (case_id,), + ).fetchall() + for row in rows: + result = _decode(row["result_json"], {}) + metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} + if ( + _clean(metadata.get("ecosystem"), 80).lower() != ecosystem + or _clean(metadata.get("package"), 512).lower() != package + or _clean(metadata.get("version"), 160) != version + ): + continue + try: + validate_quarantined_intake(result) + except IntakeError: + continue + return result + return None + + def _materialize_ai_review_items(pipeline_id: str, *, db_path: Optional[str]) -> None: pipeline = get_pipeline(pipeline_id, db_path=db_path) revision = int(pipeline.get("revision") or 1) diff --git a/tests/test_research_pipeline.py b/tests/test_research_pipeline.py index fb1612b..583eebd 100644 --- a/tests/test_research_pipeline.py +++ b/tests/test_research_pipeline.py @@ -9,6 +9,7 @@ from secopsai.intelligence_jobs import claim_next_job, complete_job, fail_job from secopsai.research_cases import add_evidence, add_subject, create_case, get_case from secopsai.research_intake import SafeFetcher +from secopsai.research_intake import IntakeError from secopsai.research_pipeline import ( agent_complete_pipeline, get_pipeline, @@ -320,6 +321,43 @@ def test_agent_review_mode_completes_pipeline_when_last_job_finishes(tmp_path, m assert completed["review_summary"]["pending"] == 0 +def test_retracted_package_reuses_exact_hash_verified_quarantine(tmp_path, monkeypatch): + db = str(tmp_path / "research.db") + monkeypatch.setenv("SECOPSAI_RESEARCH_QUARANTINE", str(tmp_path / "quarantine")) + case = _case(db) + first = start_investigation_pipeline( + case["case_id"], + reference_ecosystem="npm", + reference_package="legitimate-pkg", + reference_version="1.0.0", + db_path=db, + fetcher=_fetcher(), + ) + _complete_bridge_queue(db) + auto_review_pipeline(first["pipeline_id"], actor="test-reviewer", db_path=db) + + def unavailable(_url, _max_bytes): + raise IntakeError("registry returned HTTP 404") + + second = start_investigation_pipeline( + case["case_id"], + reference_ecosystem="npm", + reference_package="legitimate-pkg", + reference_version="1.0.0", + db_path=db, + fetcher=SafeFetcher(fetch=unavailable), + ) + steps = {step["step_key"]: step for step in second["steps"]} + assert second["pipeline_id"] != first["pipeline_id"] + assert second["status"] == "awaiting_ai" + assert steps["collect_subject"]["result"]["reuse"]["mode"] == "verified_quarantine" + assert steps["collect_reference"]["result"]["reuse"]["hash_verified"] is True + assert steps["collect_subject"]["result"]["safety"]["execution_performed"] is False + assert second["summary"]["quarantine_reuse_count"] == 2 + assert second["summary"]["registry_collection_degraded"] is True + assert "verified a previously quarantined" in steps["evidence_matrix"]["result"]["claims"][0]["statement"] + + def test_pipeline_does_not_guess_a_legitimate_reference(tmp_path, monkeypatch): db = str(tmp_path / "research.db") monkeypatch.setenv("SECOPSAI_RESEARCH_QUARANTINE", str(tmp_path / "quarantine"))