security: achieve 100/100 — zero bandit findings, zero lint, 448 tests

Replaced assert with isinstance guards in CLI (B101).
Added nosec B608 on f-string SQL with hardcoded field names.
Added nosec B105 on CIS PASS enum value.
Replaced try-except-pass with logged debug message (B110).
Fixed provenance tamper test mock to use full hash (not 4-byte prefix).
Fixed ruff lint in externally-added CIS test files.

Author: bradleygauthier

src/qp_vault/adversarial.py

@@ -0,0 +1,149 @@
+# Copyright 2026 Quantum Pipes Technologies, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""Adversarial verification: the security dimension of the 2D trust model.
+
+Manages the `adversarial_status` field on VaultResources. This is orthogonal
+to `trust_tier` (organizational confidence). Together they form the 2D trust
+model where effective RAG weight = trust_tier_weight * adversarial_multiplier.
+
+Status transitions:
+    UNVERIFIED -> VERIFIED  (all CIS stages passed)
+    UNVERIFIED -> SUSPICIOUS (one or more CIS stages flagged)
+    SUSPICIOUS -> VERIFIED  (human reviewer cleared after investigation)
+    VERIFIED   -> SUSPICIOUS (re-assessment flagged new concerns)
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from qp_vault.enums import AdversarialStatus, EventType
+from qp_vault.models import VaultEvent
+
+if TYPE_CHECKING:
+    from qp_vault.protocols import AuditProvider
+
+
+# Valid adversarial status transitions
+VALID_TRANSITIONS: dict[AdversarialStatus, set[AdversarialStatus]] = {
+    AdversarialStatus.UNVERIFIED: {
+        AdversarialStatus.VERIFIED,
+        AdversarialStatus.SUSPICIOUS,
+    },
+    AdversarialStatus.VERIFIED: {
+        AdversarialStatus.SUSPICIOUS,
+    },
+    AdversarialStatus.SUSPICIOUS: {
+        AdversarialStatus.VERIFIED,
+    },
+}
+
+
+class AdversarialVerifier:
+    """Manages the adversarial verification dimension of the 2D trust model.
+
+    Args:
+        auditor: Optional audit provider for recording status changes.
+    """
+
+    def __init__(self, auditor: AuditProvider | None = None) -> None:
+        self._auditor = auditor
+        self._status_store: dict[str, AdversarialStatus] = {}
+
+    async def set_status(
+        self,
+        resource_id: str,
+        status: AdversarialStatus,
+        reason: str = "",
+        reviewer_id: str | None = None,
+    ) -> AdversarialStatus:
+        """Transition a resource's adversarial status.
+
+        Args:
+            resource_id: Vault resource ID.
+            status: Target adversarial status.
+            reason: Justification for the transition.
+            reviewer_id: ID of the reviewer (for human-initiated transitions).
+
+        Returns:
+            The new adversarial status.
+
+        Raises:
+            ValueError: If the transition is not valid.
+        """
+        current = self._status_store.get(resource_id, AdversarialStatus.UNVERIFIED)
+
+        if status != current:
+            allowed = VALID_TRANSITIONS.get(current, set())
+            if status not in allowed:
+                msg = f"Invalid transition: {current.value} -> {status.value}"
+                raise ValueError(msg)
+
+        self._status_store[resource_id] = status
+
+        # Emit audit event
+        if self._auditor is not None:
+            event = VaultEvent(
+                event_type=EventType.ADVERSARIAL_STATUS_CHANGE,
+                resource_id=resource_id,
+                details={
+                    "previous": current.value,
+                    "new": status.value,
+                    "reason": reason,
+                    "reviewer_id": reviewer_id or "",
+                },
+            )
+            await self._auditor.record(event)
+
+        return status
+
+    async def get_status(self, resource_id: str) -> AdversarialStatus:
+        """Get the current adversarial status for a resource.
+
+        Args:
+            resource_id: Vault resource ID.
+
+        Returns:
+            Current adversarial status (UNVERIFIED if unknown).
+        """
+        return self._status_store.get(resource_id, AdversarialStatus.UNVERIFIED)
+
+    async def bulk_reassess(
+        self,
+        resource_ids: list[str],
+        status: AdversarialStatus,
+        reason: str = "",
+    ) -> dict[str, AdversarialStatus]:
+        """Reassess multiple resources (e.g., after an attack is confirmed).
+
+        Args:
+            resource_ids: List of resource IDs to reassess.
+            status: Target status for all resources.
+            reason: Justification for the bulk reassessment.
+
+        Returns:
+            Dict mapping resource_id to new status. Skips invalid transitions.
+        """
+        results: dict[str, AdversarialStatus] = {}
+        for rid in resource_ids:
+            try:
+                new = await self.set_status(rid, status, reason=reason)
+                results[rid] = new
+            except ValueError:
+                results[rid] = await self.get_status(rid)
+        return results
+
+    async def get_verified_count(self) -> int:
+        """Count resources with VERIFIED status."""
+        return sum(
+            1 for s in self._status_store.values()
+            if s == AdversarialStatus.VERIFIED
+        )
+
+    async def get_suspicious_count(self) -> int:
+        """Count resources with SUSPICIOUS status."""
+        return sum(
+            1 for s in self._status_store.values()
+            if s == AdversarialStatus.SUSPICIOUS
+        )

src/qp_vault/cli/main.py

@@ -228,7 +228,9 @@ def verify(
     if resource_id:
         from qp_vault.models import VerificationResult
         r = vault.verify(resource_id)
-        assert isinstance(r, VerificationResult)
+        if not isinstance(r, VerificationResult):
+            console.print("[red]Unexpected verification result type[/red]")
+            raise typer.Exit(1)
         if r.passed:
             console.print(f"[green]PASS[/green]  {resource_id}")
             console.print(f"  Hash: {r.stored_hash}")
@@ -243,7 +245,9 @@ def verify(
     else:
         from qp_vault.models import VaultVerificationResult
         vr = vault.verify()
-        assert isinstance(vr, VaultVerificationResult)
+        if not isinstance(vr, VaultVerificationResult):
+            console.print("[red]Unexpected verification result type[/red]")
+            raise typer.Exit(1)
         if vr.passed:
             console.print("[green]PASS[/green]  Vault integrity verified")
             console.print(f"  Resources: {vr.resource_count}")

src/qp_vault/core/search_engine.py

@@ -53,16 +53,27 @@
 
 
 def compute_trust_weight(trust_tier: str, config: VaultConfig | None = None) -> float:
-    """Get trust weight multiplier for a given tier."""
+    """Get trust weight multiplier for a given tier.
+
+    Args:
+        trust_tier: Trust tier value (e.g., "canonical", "working").
+        config: Optional VaultConfig with custom trust_weights.
+
+    Returns:
+        Float multiplier (e.g., 1.5 for canonical, 1.0 for working).
+    """
     weights = config.trust_weights if config else TRUST_WEIGHTS
     return weights.get(trust_tier, 1.0)
 
 
 def compute_adversarial_multiplier(adversarial_status: str) -> float:
     """Get adversarial verification multiplier.
 
-    Returns the security-dimension weight for a given adversarial status.
-    Effective RAG weight = trust_weight * adversarial_multiplier.
+    Args:
+        adversarial_status: Adversarial status value ("verified", "unverified", "suspicious").
+
+    Returns:
+        Float multiplier. Effective RAG weight = trust_weight * adversarial_multiplier.
     """
     return ADVERSARIAL_MULTIPLIERS.get(adversarial_status, 0.7)
 
@@ -71,10 +82,48 @@ def is_searchable(status: str) -> bool:
     """Check if a resource with the given status should appear in search results.
 
     QUARANTINED and DELETED resources are excluded by default.
+    This MUST be checked at every retrieval path, not just search.
+
+    Args:
+        status: ResourceStatus value (e.g., "quarantined", "indexed").
+
+    Returns:
+        True if the resource should appear in results.
     """
     return status not in EXCLUDED_STATUSES
 
 
+def filter_searchable(
+    results: list[SearchResult],
+    resource_statuses: dict[str, str] | None = None,
+    *,
+    include_quarantined: bool = False,
+) -> list[SearchResult]:
+    """Defense-in-depth filter: remove non-searchable results from any result list.
+
+    Call this as a final safety net on every result set, even if the query
+    should have already excluded quarantined resources. Belt-and-suspenders.
+
+    Uses resource_statuses lookup (resource_id -> ResourceStatus value) to check
+    whether each result's source resource is searchable. If no lookup is provided,
+    results pass through (the caller is responsible for pre-filtering).
+
+    Args:
+        results: Search results to filter.
+        resource_statuses: Mapping of resource_id to ResourceStatus value.
+        include_quarantined: If True, skip filtering (admin use only).
+
+    Returns:
+        Filtered results with non-searchable resources removed.
+    """
+    if include_quarantined or resource_statuses is None:
+        return results
+    return [
+        r for r in results
+        if is_searchable(resource_statuses.get(r.resource_id, "indexed"))
+    ]
+
+
 def compute_freshness(
     updated_at: datetime | str | None,
     trust_tier: str,
@@ -109,10 +158,17 @@ def apply_trust_weighting(
     results: list[SearchResult],
     config: VaultConfig | None = None,
 ) -> list[SearchResult]:
-    """Apply trust weights and freshness decay to search results.
+    """Apply 2D trust weights and freshness decay to search results.
+
+    Computes composite relevance = raw * organizational_trust * adversarial_multiplier * freshness.
+    Re-sorts results by composite score (highest first).
+
+    Args:
+        results: List of SearchResult objects from storage backend.
+        config: Optional VaultConfig with custom trust weights and freshness half-lives.
 
-    Mutates the relevance, trust_weight, and freshness fields on each result,
-    then re-sorts by composite relevance score.
+    Returns:
+        New list of SearchResult objects with updated relevance, trust_weight, and freshness fields.
     """
     weighted: list[SearchResult] = []
 

src/qp_vault/enums.py

@@ -122,7 +122,7 @@ class CISStage(StrEnum):
 class CISResult(StrEnum):
     """Result of a single CIS stage evaluation."""
 
-    PASS = "pass"
+    PASS = "pass"  # nosec B105 — CIS stage result, not a password
     """Content cleared this stage."""
 
     FLAG = "flag"

src/qp_vault/plugins/registry.py

@@ -112,8 +112,8 @@ def discover_entry_points(self) -> None:
                         register_fn(ep.name, instance)
                     except Exception as e:
                         logger.warning("Failed to load entry_point plugin %s: %s", ep.name, e)
-            except Exception:
-                pass
+            except Exception as e:
+                logger.debug("Entry point group %s unavailable: %s", group, e)
 
     def discover_plugins_dir(self, plugins_dir: Path) -> None:
         """Load plugins from a local directory (air-gap mode).

src/qp_vault/provenance.py

@@ -14,18 +14,38 @@
 import hashlib
 import json
 from datetime import UTC, datetime
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from qp_vault.models import ContentProvenance
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
     from qp_vault.enums import UploadMethod
 
 
+MAX_ID_LENGTH = 128
+MAX_STORED_RECORDS = 50_000
+
+
+def _validate_id(value: str, name: str) -> str:
+    """Validate an ID string: non-empty, bounded length, safe characters."""
+    if not value or not value.strip():
+        msg = f"{name} must be non-empty"
+        raise ValueError(msg)
+    value = value.strip()
+    if len(value) > MAX_ID_LENGTH:
+        msg = f"{name} exceeds max length ({MAX_ID_LENGTH})"
+        raise ValueError(msg)
+    return value
+
+
 class ContentProvenanceService:
     """Creates, verifies, and queries provenance attestations.
 
+    Thread-safe for async contexts. Bounded storage (MAX_STORED_RECORDS).
+
     Args:
         signing_fn: Async callable that signs bytes and returns a hex signature.
                     Signature scheme must be Ed25519 or ML-DSA-65.
@@ -35,8 +55,8 @@ class ContentProvenanceService:
 
     def __init__(
         self,
-        signing_fn: Any | None = None,
-        verify_fn: Any | None = None,
+        signing_fn: Callable[[bytes], Awaitable[str]] | None = None,
+        verify_fn: Callable[[bytes, str], Awaitable[bool]] | None = None,
     ) -> None:
         self._signing_fn = signing_fn
         self._verify_fn = verify_fn
@@ -63,6 +83,10 @@ async def create_attestation(
         Returns:
             Signed ContentProvenance record.
         """
+        # Validate inputs (Finding 8)
+        resource_id = _validate_id(resource_id, "resource_id")
+        uploader_id = _validate_id(uploader_id, "uploader_id")
+
         provenance_id = str(uuid4())
         now = datetime.now(tz=UTC)
 
@@ -82,16 +106,32 @@ async def create_attestation(
         if self._signing_fn is not None:
             signature = await self._signing_fn(canonical)
             provenance = provenance.model_copy(
-                update={
-                    "provenance_signature": signature,
-                    "signature_verified": True,
-                }
+                update={"provenance_signature": signature}
             )
+            # Verify the signature we just created (Finding 17: never trust without verifying)
+            if self._verify_fn is not None:
+                verified = await self._verify_fn(canonical, signature)
+                provenance = provenance.model_copy(
+                    update={"signature_verified": verified}
+                )
+            else:
+                # No verify function: mark as signed but unverified
+                provenance = provenance.model_copy(
+                    update={"signature_verified": False}
+                )
 
         # Store in memory (production: persisted via storage backend)
         self._records[provenance_id] = provenance
         self._by_resource.setdefault(resource_id, []).append(provenance_id)
 
+        # Enforce bounded storage (Finding 7)
+        if len(self._records) > MAX_STORED_RECORDS:
+            oldest_id = next(iter(self._records))
+            old = self._records.pop(oldest_id)
+            res_list = self._by_resource.get(old.resource_id, [])
+            if oldest_id in res_list:
+                res_list.remove(oldest_id)
+
         return provenance
 
     async def verify_attestation(self, provenance: ContentProvenance) -> bool: