Remediate review-of-review findings (thoth-l0n, thoth-hai, thoth-hu3, thoth-4or, thoth-0x0, thoth-6rt, thoth-v2j)

SECURITY_REVIEW_CCF_0.1.2-rc1.md

@@ -0,0 +1,152 @@
+# Security review — `ccf-0.1.2-rc1`
+
+Review date: 2026-02-13
+Scope: branch `ccf-0.1.2-rc1`, production code only (~21k LOC: `ccf/` package,
+core wiring in `core/`, `scripts/`). Diff base: `main` at merge-base `0ed653b`.
+
+Areas read end to end: crypto/integrity core (`hashing.py`, `jcs.py`,
+`keys.py`, `ids.py`), untrusted-input surfaces (`sync/packio.py`,
+`sync/restore.py`, `sync/transport.py`, `sync/chunks.py`), admission
+(`admission.py`), journal (`journal.py`), governance (`engine.py`,
+`capabilities.py`, `evaluator.py`, `authority.py`), erasure
+(`suppression.py`, `purge.py`, `media.py`, `operations.py`), credentials
+(`credentials.py`), obsidian importer (`vault.py`, `notes.py`), and the
+core wiring diffs (`ccf_dualwrite.py`, `capture_lifecycle.py`,
+`connector_capture.py`, `postgres_migrations.py`).
+
+Overall assessment: the cryptographic core and governance/erasure engines
+are unusually disciplined — fail-closed everywhere, default-deny policy
+engine, signed member Merkle roots, strict RFC 8785 canonicalization with
+duplicate-key and surrogate rejection, 0600 key handling, anchored delta
+apply. The findings below are the gaps that survive that discipline,
+concentrated in the seams: bootstrap anchoring, a file re-read race,
+resource limits, and credential lifecycle enforcement.
+
+## High
+
+### H1. Restore has no identity anchor by default — a fully forged archive passes verification
+
+`ccf/sync/restore.py:405-425`: `restore_mindpack(...,
+trusted_genesis_hash=None, trusted_head_hash=None)` — both optional, and no
+production caller passes them (`git grep restore_mindpack` → only
+`scripts/ccf_stage9.py`, which *does* pass them). The pack's chain is
+signed, but by a key carried *inside the pack*. An attacker who generates
+their own Ed25519 key and signs a malicious, self-consistent archive
+passes every check (stream digests, commitment recomputation, chain,
+in-database `verify_chain`) unless the operator supplies an out-of-band
+trusted hash. The whole CCF integrity model collapses to nothing at the
+bootstrap point.
+
+Fix: require `trusted_genesis_hash` (or an explicit
+`bootstrap_new_archive=True` flag that prints the new genesis hash for
+out-of-band pinning). Fail closed otherwise.
+
+## Medium
+
+### M1. TOCTOU on the pack file during restore — operational streams re-read unverified
+
+`ccf/sync/restore.py:434-447`: the pack is fully verified through one
+`PackReader`, closed, then `archive.json`, `lineage-heads.ndjson`,
+`origin-index.ndjson`, `producer-heads.ndjson`, and `producer-batches/*`
+are re-read through a *second* `PackReader` on the same path **without
+digest re-verification**. A local attacker who can swap the pack file/dir
+between the two opens injects unverified operational state (epoch, signer
+key id, origins, producer heads). The epoch/genesis swap mostly DoSes
+(final `verify_chain` catches it), but origin/producer-head rows land as
+given.
+
+Fix: keep one reader open for the whole restore, or re-check the manifest
+digests on every re-read.
+
+### M2. Zip bomb / unbounded decompression
+
+`ccf/sync/packio.py:146-161` (`PackReader.read`) and
+`verify_stream_digests` read each entry fully into memory **before**
+checking the manifest's `byte_length` or digest, with no caps on entry
+count, per-entry compressed size, or total uncompressed size. Packs arrive
+over HTTP (`ccf/sync/transport.py`) — a malicious `.mindpack` OOMs the host
+before any integrity check runs. Same pattern in `sync/chunks.py`
+(`build_sidecar`, `verify_file`) and `make_pack_app` (full `read_bytes` per
+request, no auth — conformance tool, keep it off the network).
+
+Fix: check `ZipInfo.file_size` against the manifest `byte_length` and a
+total-uncompressed cap before reading; stream-hash with the cap enforced.
+
+### M3. Credential expiry and scopes are never enforced at admission
+
+`ccf/credentials.py:resolve_credential_public_key` checks only that the
+credential record exists, is unambiguous, and its lineage head isn't
+`revoke`. The payload's `valid_from`, `expires_at`, `offline_grace_until`,
+and `scopes` are declared (`ccf/dualwrite/service.py` issues
+`scopes=["capture","sync","derive"]`) but never consulted by
+`_verify_batch_envelope` (`ccf/admission.py:480-505`). An expired,
+out-of-scope, or not-yet-valid device credential keeps signing admissible
+batches until explicitly revoked.
+
+Fix: enforce `valid_from ≤ now < expires_at` and check the operation
+against `scopes` in `resolve_credential_public_key` (or return the full
+payload and enforce at the envelope).
+
+## Low
+
+### L1. Suppression key written with default umask
+
+`ccf/erasure/suppression.py:generate_suppression_key` uses
+`Path.write_text` → typically 0644, world-readable HMAC key protecting
+suppression-after-erasure. `ccf/keys.py` does this correctly (0600 +
+`O_EXCL`); mirror it.
+
+### L2. Exported packs are world-readable
+
+`PackWriter.write_bytes` → `Path.write_bytes` (0644) into a 0755 dir
+(`ccf/sync/packio.py`). Mindpacks contain plaintext compartments and blob
+bytes. Recommend 0700/0600 (or an explicit `chmod` flag) on export.
+
+### L3. Manifest availability lists are unauthenticated
+
+`withheld`, `erased`, `external_dependencies` (and `mode`, `counts`) in
+`manifest.json` are not bound by any digest in the signed chain. Traced:
+an attacker can inflate these lists, but cannot forge content — headers
+are still bound by member `object_hash` → Merkle root → signed payload, so
+the effect is limited to availability semantics, withheld rows, and DoS.
+Still: integrity-relevant metadata traveling unsigned. Bind the full
+manifest into a signed commitment in the next spec revision.
+
+### L4. Malformed member fields propagate raw KeyError/ValueError
+
+`verify_commit_chain`/`merkle_root` raise raw `KeyError`/`ValueError` on
+malformed member fields (missing `commit_position`, non-numeric
+`member_count`) instead of `PackVerificationError`
+(`ccf/sync/verify.py:130-160`). Fails closed (transaction rollback), but
+leaks raw stack traces and breaks the error contract. Wrap in `PackError`.
+
+## Notes (not vulnerabilities)
+
+- `StreamEntry.from_dict` coerces `required` via `bool()` — `"false"` →
+  `True` — but the manifest schema pins `"type": "boolean"` before this
+  runs (`spec/.../mindpack-manifest.schema.json:90-92`), so it's dead code.
+  Fail-closed direction anyway.
+- The branch ships a full `spec/ccf/0.1.2-rc1/` package (schemas, URNs
+  `0.1.2-rc1:*`, format `ccf.mindpack/0.1.2-rc1`), but **zero runtime code
+  references it** — everything is pinned to the 0.1.1 package
+  (`SCHEMA_MINDPACK_MANIFEST = "urn:ccf:schema:0.1.1:..."` in
+  `ccf/sync/export.py:27`). Nothing breaks (runtime stays self-consistent
+  on 0.1.1), but the RC package is unwired, and anyone pointing
+  `package_root` at 0.1.2-rc1 will get restore rejections from the
+  `format != "ccf.mindpack/0.1.1"` check in `restore.py`. Decide: wire the
+  RC package or document it as spec-only.
+- HTTP transport is only as trustworthy as the sidecar channel:
+  `fetch_sidecar_http` takes `pack_digest` at face value, so MITM wins
+  unless the restore-side trusted-genesis anchor (H1) exists. H1's fix also
+  fixes this.
+- `_verify_batch_envelope` correctly refuses to persist rejected envelopes
+  (anti-chain-poisoning), and the suppression response-shaping avoids
+  oracle leaks — both good.
+
+## Bottom line
+
+The cryptographic core and governance/erasure engines are solid; the
+failures are in the seams: bootstrap anchoring (H1), a file re-read race
+(M1), resource limits (M2), and credential lifecycle enforcement (M3). Fix
+H1 first; it is one guard and it makes the rest of the system actually
+mean something.

collectors/pi_skill_connector.py

@@ -6,6 +6,7 @@
 import hashlib
 import inspect
 import json
+import logging
 import shutil
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -34,6 +35,8 @@
     reject_direct_wiki_write_claims,
 )
 
+logger = logging.getLogger(__name__)
+
 
 DEFAULT_SOURCE_NAME = "pi_skill"
 DEFAULT_MAX_INPUT_BYTES = 200_000
@@ -342,6 +345,13 @@ def _load_skills(self) -> dict[str, PiSkillDefinition]:
             )
             artifact_types_defaulted = not raw_artifact_types
             artifact_types = raw_artifact_types or tuple(sorted(SUPPORTED_ARTIFACT_TYPES))
+            if artifact_types_defaulted:
+                logger.warning(
+                    "Pi skill %r does not declare artifact_types; "
+                    "defaulting to all supported types: %s",
+                    skill_id,
+                    ", ".join(artifact_types),
+                )
             unsupported = set(artifact_types) - SUPPORTED_ARTIFACT_TYPES
             if unsupported:
                 raise ValueError(

core/agent_context.py

@@ -6,6 +6,7 @@
 from collections.abc import Mapping
 from typing import Any
 
+from .collection_utils import first_present_value
 from .hybrid_search import HybridSearchHit
 from .metadata_db import IngestionQueueEntry
 from .prompt_security import (
@@ -57,7 +58,7 @@ def artifact_security_state(entry: IngestionQueueEntry) -> dict[str, Any]:
 def artifact_trust_state(entry: IngestionQueueEntry) -> dict[str, Any]:
     payload = _json_object(entry.payload_json)
     security = artifact_security_state(entry)
-    explicit_score = _first_present_value(
+    explicit_score = first_present_value(
         payload,
         "source_trust_score",
         "trust_score",
@@ -230,7 +231,7 @@ def capture_event_trust_state(event: Mapping[str, Any]) -> dict[str, Any]:
     provenance = event.get("provenance")
     provenance_payload = dict(provenance) if isinstance(provenance, Mapping) else {}
     security = capture_event_security_state(event)
-    explicit_score = _first_present_value(
+    explicit_score = first_present_value(
         provenance_payload,
         "source_trust_score",
         "trust_score",
@@ -381,13 +382,6 @@ def _first_string(value: Any) -> str | None:
     return values[0] if values else None
 
 
-def _first_present_value(payload: Mapping[str, Any], *keys: str) -> Any:
-    for key in keys:
-        if key in payload and payload[key] is not None:
-            return payload[key]
-    return None
-
-
 def _compact_mapping(value: Mapping[str, Any]) -> dict[str, Any]:
     return {
         key: item

core/agent_surface.py

@@ -428,16 +428,13 @@ def retry_artifact_review(
         metadata: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Retry a review-queue artifact by moving it back to pending."""
-        try:
-            entry = ArtifactReviewQueueService(self.db).retry(
-                artifact_id,
-                actor=actor,
-                reason=reason,
-                metadata=metadata,
-            )
-        except ArtifactReviewQueueError as exc:
-            raise AgentSurfaceError(str(exc)) from exc
-        return {"queue": self._serialize_ingestion_entry(entry)}
+        return self._apply_artifact_review(
+            "retry",
+            artifact_id,
+            actor=actor,
+            reason=reason,
+            metadata=metadata,
+        )
 
     def reject_artifact_review(
         self,
@@ -448,16 +445,13 @@ def reject_artifact_review(
         metadata: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Reject a bad artifact and keep its provenance/error audit."""
-        try:
-            entry = ArtifactReviewQueueService(self.db).reject(
-                artifact_id,
-                actor=actor,
-                reason=reason,
-                metadata=metadata,
-            )
-        except ArtifactReviewQueueError as exc:
-            raise AgentSurfaceError(str(exc)) from exc
-        return {"queue": self._serialize_ingestion_entry(entry)}
+        return self._apply_artifact_review(
+            "reject",
+            artifact_id,
+            actor=actor,
+            reason=reason,
+            metadata=metadata,
+        )
 
     def mark_artifact_reviewed(
         self,
@@ -468,8 +462,25 @@ def mark_artifact_reviewed(
         metadata: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Mark a bad artifact reviewed without retrying or accepting it."""
+        return self._apply_artifact_review(
+            "mark_reviewed",
+            artifact_id,
+            actor=actor,
+            reason=reason,
+            metadata=metadata,
+        )
+
+    def _apply_artifact_review(
+        self,
+        action: str,
+        artifact_id: str,
+        *,
+        actor: str,
+        reason: str,
+        metadata: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
         try:
-            entry = ArtifactReviewQueueService(self.db).mark_reviewed(
+            entry = getattr(ArtifactReviewQueueService(self.db), action)(
                 artifact_id,
                 actor=actor,
                 reason=reason,

core/archivist_state.py

@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import hashlib
 import json
 from typing import Any, Mapping
 
 from .archivist_selection import ArchivistCandidate
 from .archivist_topics import ArchivistTopicDefinition
 from .metadata_db import MetadataDB, get_metadata_db
+from .time_utils import utc_now, utc_now_iso
 
 ARCHIVIST_STATE_KEY_PREFIX = "archivist.topic."
 
@@ -109,7 +110,7 @@ def request_archivist_topic_force(
         last_candidate_count=existing.last_candidate_count,
         last_model_provider=existing.last_model_provider,
         last_model=existing.last_model,
-        force_requested_at=requested_at or _now_iso(),
+        force_requested_at=requested_at or utc_now_iso(),
         force_reason=reason,
     )
     _store_archivist_topic_state(updated, db=metadata_db)
@@ -190,7 +191,7 @@ def evaluate_archivist_dirty_check(
     provider = route[0] if route else None
     model = route[1] if route else None
     next_due_at = _compute_next_due_at(state.last_success_at, topic.cadence_hours)
-    now_dt = now or datetime.now()
+    now_dt = now or utc_now()
 
     if state.force_requested_at:
         return ArchivistDirtyCheckResult(
@@ -244,7 +245,9 @@ def evaluate_archivist_dirty_check(
             model=model,
         )
 
-    if next_due_at is not None and now_dt >= _parse_datetime(next_due_at):
+    if next_due_at is not None and _coerce_utc(now_dt) >= _coerce_utc(
+        _parse_datetime(next_due_at)
+    ):
         return ArchivistDirtyCheckResult(
             should_run=True,
             reason="cadence_due",
@@ -286,7 +289,7 @@ def record_archivist_topic_run(
     snapshot = snapshot_archivist_candidates(tuple(candidates))
     provider = route[0] if route else None
     model = route[1] if route else None
-    recorded_at = run_at or _now_iso()
+    recorded_at = run_at or utc_now_iso()
 
     updated = ArchivistTopicState(
         topic_id=topic.id,
@@ -392,11 +395,14 @@ def _parse_datetime(value: str) -> datetime:
         raise ArchivistTopicStateError(f"Invalid archivist timestamp: {value}") from exc
 
 
+def _coerce_utc(value: datetime) -> datetime:
+    """Treat naive datetimes (legacy persisted state) as UTC for comparison."""
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
+
+
 def _optional_text(value: Any) -> str | None:
     if value in (None, ""):
         return None
     return str(value)
-
-
-def _now_iso() -> str:
-    return datetime.now().isoformat()

core/bounded_workers.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from collections.abc import Awaitable, Callable, Sequence
 from typing import TypeVar
 
+logger = logging.getLogger(__name__)
+
 T = TypeVar("T")
 R = TypeVar("R")
 
@@ -89,6 +92,8 @@ async def run_worker() -> None:
 
     if errors:
         errors.sort(key=lambda item: item[0])
+        for index, exc in errors:
+            logger.error("Bounded worker item %d failed: %s", index, exc)
         raise errors[0][1]
 
     return [item for item in results if item is not _MISSING]