Proving What AI Didn’t Generate: Building Cryptographic Refusal Logs with CAP-SRP

User Request → AI Generation → Content Created → C2PA Manifest Attached
                                      ↓
                              "This image was created by
                               DALL-E 3 on 2026-01-28"

User Request → Safety Filter → REFUSED → ??? 
                                   ↓
                            No content created
                            No manifest possible
                            No cryptographic proof

┌─────────────────────────────────────────────────────────────────┐
│                     CONTENT LAYER                               │
│   ┌─────────────┐    ┌─────────────┐    ┌─────────────┐        │
│   │ Watermarks  │    │    C2PA     │    │   SynthID   │        │
│   │             │    │  Manifests  │    │             │        │
│   └─────────────┘    └─────────────┘    └─────────────┘        │
│                                                                 │
│   Problem: Can only attest to content that EXISTS              │
└─────────────────────────────────────────────────────────────────┘
                              │
                              │ CAP-SRP bridges this gap
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                     SYSTEM LAYER                                │
│   ┌─────────────────────────────────────────────────────────┐  │
│   │                      CAP-SRP                            │  │
│   │                                                         │  │
│   │  Logs DECISIONS, not content:                          │  │
│   │  • GEN_ATTEMPT  (request received)                     │  │
│   │  • GEN          (content created)                      │  │
│   │  • GEN_DENY     (request refused)                      │  │
│   │  • GEN_ERROR    (system failure)                       │  │
│   │                                                         │  │
│   │  Every attempt has exactly one outcome.                │  │
│   │  Cryptographically provable.                           │  │
│   │  Externally anchored.                                  │  │
│   └─────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

enum EventType {
  GEN_ATTEMPT = "GEN_ATTEMPT",  // Request received (logged BEFORE evaluation)
  GEN = "GEN",                  // Content successfully generated
  GEN_DENY = "GEN_DENY",        // Request refused by safety system
  GEN_ERROR = "GEN_ERROR"       // System error prevented completion
}

┌──────────────────────────────────────────────────────────────────┐
│ Timeline                                                         │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│ t₀ ─────► Request received                                       │
│           │                                                      │
│ t₁ ─────► GEN_ATTEMPT logged ◄─── COMMITMENT POINT               │
│           │                       (before safety evaluation)     │
│           │                                                      │
│ t₂ ─────► Safety evaluation begins                               │
│           │                                                      │
│           ├──► Safe ──────► t₃: Generate content                 │
│           │                      │                               │
│           │                 t₄: GEN logged (links to attempt)    │
│           │                      │                               │
│           │                 t₅: C2PA manifest attached           │
│           │                                                      │
│           └──► Unsafe ────► t₃: GEN_DENY logged                  │
│                                  (links to attempt)              │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

∀ time_window T:
  COUNT(GEN_ATTEMPT) = COUNT(GEN) + COUNT(GEN_DENY) + COUNT(GEN_ERROR)

def detect_hidden_generations(events: List[Event]) -> List[str]:
    """Find attempts without recorded outcomes."""
    attempts = {e.attempt_id for e in events if e.type == "GEN_ATTEMPT"}
    outcomes = {e.attempt_id for e in events if e.type in ["GEN", "GEN_DENY", "GEN_ERROR"]}

    orphan_attempts = attempts - outcomes

    if orphan_attempts:
        return [f"ALERT: Attempt {id} has no recorded outcome" for id in orphan_attempts]
    return []

def detect_fabricated_refusals(events: List[Event]) -> List[str]:
    """Find outcomes without corresponding attempts."""
    attempts = {e.attempt_id for e in events if e.type == "GEN_ATTEMPT"}
    outcomes = {e.attempt_id for e in events if e.type in ["GEN", "GEN_DENY", "GEN_ERROR"]}

    orphan_outcomes = outcomes - attempts

    if orphan_outcomes:
        return [f"ALERT: Outcome {id} has no recorded attempt" for id in orphan_outcomes]
    return []

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional
import hashlib
import uuid

class EventType(Enum):
    GEN_ATTEMPT = "GEN_ATTEMPT"
    GEN = "GEN"
    GEN_DENY = "GEN_DENY"
    GEN_ERROR = "GEN_ERROR"

class RiskCategory(Enum):
    CSAM_RISK = "CSAM_RISK"
    NCII_RISK = "NCII_RISK"
    MINOR_SEXUALIZATION = "MINOR_SEXUALIZATION"
    REAL_PERSON_DEEPFAKE = "REAL_PERSON_DEEPFAKE"
    VIOLENCE_GRAPHIC = "VIOLENCE_GRAPHIC"
    HATE_SPEECH = "HATE_SPEECH"
    SELF_HARM = "SELF_HARM"
    ILLEGAL_ACTIVITY = "ILLEGAL_ACTIVITY"

@dataclass
class CAPEvent:
    event_id: str = field(default_factory=lambda: str(uuid.uuid7()))
    event_type: EventType
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    attempt_id: str  # Links outcomes to attempts

    # For GEN_ATTEMPT
    prompt_hash: Optional[str] = None  # SHA-256 of prompt (never raw text!)
    model_id: Optional[str] = None
    user_context_hash: Optional[str] = None

    # For GEN
    output_hash: Optional[str] = None
    c2pa_manifest_id: Optional[str] = None

    # For GEN_DENY
    denial_reason_code: Optional[str] = None
    risk_category: Optional[RiskCategory] = None
    risk_score: Optional[float] = None
    policy_version: Optional[str] = None

    # For GEN_ERROR
    error_code: Optional[str] = None
    error_category: Optional[str] = None

    # Integrity fields
    previous_hash: Optional[str] = None
    signature: Optional[str] = None

    def compute_hash(self) -> str:
        """Compute SHA-256 hash of event content."""
        content = f"{self.event_id}|{self.event_type.value}|{self.timestamp.isoformat()}|{self.attempt_id}"
        return hashlib.sha256(content.encode()).hexdigest()

from typing import List, Tuple
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
import json

class CAPEventLogger:
    def __init__(self, signing_key: Ed25519PrivateKey, model_id: str):
        self.signing_key = signing_key
        self.model_id = model_id
        self.events: List[CAPEvent] = []
        self.last_hash: Optional[str] = None

    def _sign_event(self, event: CAPEvent) -> str:
        """Sign event with Ed25519."""
        message = event.compute_hash().encode()
        signature = self.signing_key.sign(message)
        return signature.hex()

    def _hash_prompt(self, prompt: str) -> str:
        """Hash prompt - NEVER store raw prompts!"""
        return hashlib.sha256(prompt.encode()).hexdigest()

    def _hash_user_context(self, user_id: str) -> str:
        """Hash user context for privacy."""
        # Add salt in production!
        return hashlib.sha256(user_id.encode()).hexdigest()

    def log_attempt(self, prompt: str, user_id: str) -> CAPEvent:
        """
        Log a generation attempt BEFORE safety evaluation.

        This is the commitment point - must happen before any
        decision is made about whether to generate.
        """
        attempt_id = str(uuid.uuid7())

        event = CAPEvent(
            event_type=EventType.GEN_ATTEMPT,
            attempt_id=attempt_id,
            prompt_hash=self._hash_prompt(prompt),
            model_id=self.model_id,
            user_context_hash=self._hash_user_context(user_id),
            previous_hash=self.last_hash
        )

        event.signature = self._sign_event(event)
        self.last_hash = event.compute_hash()
        self.events.append(event)

        return event

    def log_generation(self, attempt_id: str, output: bytes, 
                       c2pa_manifest_id: Optional[str] = None) -> CAPEvent:
        """Log successful content generation."""
        event = CAPEvent(
            event_type=EventType.GEN,
            attempt_id=attempt_id,
            output_hash=hashlib.sha256(output).hexdigest(),
            c2pa_manifest_id=c2pa_manifest_id,
            previous_hash=self.last_hash
        )

        event.signature = self._sign_event(event)
        self.last_hash = event.compute_hash()
        self.events.append(event)

        return event

    def log_denial(self, attempt_id: str, risk_category: RiskCategory,
                   risk_score: float, policy_version: str) -> CAPEvent:
        """Log a safety-based refusal."""
        event = CAPEvent(
            event_type=EventType.GEN_DENY,
            attempt_id=attempt_id,
            denial_reason_code=risk_category.value,
            risk_category=risk_category,
            risk_score=risk_score,
            policy_version=policy_version,
            previous_hash=self.last_hash
        )

        event.signature = self._sign_event(event)
        self.last_hash = event.compute_hash()
        self.events.append(event)

        return event

    def log_error(self, attempt_id: str, error_code: str, 
                  error_category: str) -> CAPEvent:
        """Log a system error."""
        event = CAPEvent(
            event_type=EventType.GEN_ERROR,
            attempt_id=attempt_id,
            error_code=error_code,
            error_category=error_category,
            previous_hash=self.last_hash
        )

        event.signature = self._sign_event(event)
        self.last_hash = event.compute_hash()
        self.events.append(event)

        return event

    def verify_completeness(self) -> Tuple[bool, List[str]]:
        """Verify the completeness invariant."""
        issues = []

        attempts = {e.attempt_id for e in self.events 
                   if e.event_type == EventType.GEN_ATTEMPT}
        outcomes = {e.attempt_id for e in self.events 
                   if e.event_type in [EventType.GEN, EventType.GEN_DENY, EventType.GEN_ERROR]}

        # Check for orphan attempts (hidden generations)
        orphan_attempts = attempts - outcomes
        for aid in orphan_attempts:
            issues.append(f"VIOLATION: Attempt {aid} has no recorded outcome")

        # Check for orphan outcomes (fabricated refusals)
        orphan_outcomes = outcomes - attempts
        for aid in orphan_outcomes:
            issues.append(f"VIOLATION: Outcome for {aid} has no recorded attempt")

        return len(issues) == 0, issues

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

# Initialize
private_key = Ed25519PrivateKey.generate()
logger = CAPEventLogger(signing_key=private_key, model_id="image-gen-v3")

# Simulate a generation flow
def handle_generation_request(prompt: str, user_id: str):
    # Step 1: Log attempt BEFORE evaluation (commitment point)
    attempt = logger.log_attempt(prompt, user_id)

    # Step 2: Safety evaluation
    risk = evaluate_safety(prompt)  # Your safety system

    # Step 3: Log outcome
    if risk.should_deny:
        logger.log_denial(
            attempt_id=attempt.attempt_id,
            risk_category=RiskCategory.NCII_RISK,
            risk_score=risk.score,
            policy_version="2026-01-28"
        )
        return {"status": "denied", "reason": risk.category}
    else:
        try:
            output = generate_image(prompt)  # Your generation system
            logger.log_generation(
                attempt_id=attempt.attempt_id,
                output=output,
                c2pa_manifest_id=attach_c2pa(output)
            )
            return {"status": "success", "image": output}
        except Exception as e:
            logger.log_error(
                attempt_id=attempt.attempt_id,
                error_code=str(type(e).__name__),
                error_category="GENERATION_FAILURE"
            )
            return {"status": "error", "message": str(e)}

# Later: Verify integrity
is_valid, issues = logger.verify_completeness()
if not is_valid:
    print("INTEGRITY VIOLATIONS DETECTED:")
    for issue in issues:
        print(f"  - {issue}")

from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey, Ed25519PublicKey
)

# Key generation
private_key = Ed25519PrivateKey.generate()
public_key = private_key.public_key()

# Signing
message = b"event data to sign"
signature = private_key.sign(message)

# Verification
try:
    public_key.verify(signature, message)
    print("Signature valid")
except Exception:
    print("Signature INVALID - tampering detected!")

import hashlib
from typing import List, Optional

class MerkleTree:
    def __init__(self):
        self.leaves: List[bytes] = []
        self.tree: List[List[bytes]] = []

    def _hash(self, data: bytes) -> bytes:
        return hashlib.sha256(data).digest()

    def _hash_pair(self, left: bytes, right: bytes) -> bytes:
        return self._hash(left + right)

    def add_leaf(self, data: bytes) -> int:
        """Add a leaf and return its index."""
        leaf_hash = self._hash(data)
        self.leaves.append(leaf_hash)
        self._rebuild_tree()
        return len(self.leaves) - 1

    def _rebuild_tree(self):
        """Rebuild the Merkle tree from leaves."""
        if not self.leaves:
            self.tree = []
            return

        self.tree = [self.leaves.copy()]

        while len(self.tree[-1]) > 1:
            level = self.tree[-1]
            next_level = []

            for i in range(0, len(level), 2):
                left = level[i]
                right = level[i + 1] if i + 1 < len(level) else left
                next_level.append(self._hash_pair(left, right))

            self.tree.append(next_level)

    @property
    def root(self) -> Optional[bytes]:
        """Get the Merkle root."""
        if not self.tree:
            return None
        return self.tree[-1][0]

    def get_proof(self, index: int) -> List[tuple]:
        """
        Get inclusion proof for leaf at index.
        Returns list of (sibling_hash, is_left) tuples.
        """
        if index >= len(self.leaves):
            raise ValueError("Index out of range")

        proof = []
        for level in self.tree[:-1]:
            sibling_index = index ^ 1  # XOR with 1 gives sibling
            if sibling_index < len(level):
                is_left = index % 2 == 1
                proof.append((level[sibling_index], is_left))
            index //= 2

        return proof

    @staticmethod
    def verify_proof(leaf_hash: bytes, proof: List[tuple], root: bytes) -> bool:
        """Verify an inclusion proof."""
        current = leaf_hash
        for sibling, is_left in proof:
            if is_left:
                current = hashlib.sha256(sibling + current).digest()
            else:
                current = hashlib.sha256(current + sibling).digest()
        return current == root

# Usage
tree = MerkleTree()

# Add events
for event in events:
    tree.add_leaf(event.compute_hash().encode())

# Get root for external anchoring
merkle_root = tree.root

# Generate proof for specific event
proof = tree.get_proof(event_index)

# Anyone can verify
is_included = MerkleTree.verify_proof(
    leaf_hash=hashlib.sha256(event.compute_hash().encode()).digest(),
    proof=proof,
    root=merkle_root
)

# Hybrid Ed25519 + ML-DSA (post-quantum)
# Note: ML-DSA libraries still maturing

class HybridSigner:
    def __init__(self, ed25519_key, ml_dsa_key):
        self.ed25519_key = ed25519_key
        self.ml_dsa_key = ml_dsa_key

    def sign(self, message: bytes) -> dict:
        return {
            "ed25519": self.ed25519_key.sign(message).hex(),
            "ml_dsa": self.ml_dsa_key.sign(message).hex(),
            "binding": hashlib.sha256(
                self.ed25519_key.sign(message) + 
                self.ml_dsa_key.sign(message)
            ).hexdigest()
        }

    def verify(self, message: bytes, signature: dict) -> bool:
        # Both must verify
        ed25519_valid = verify_ed25519(message, signature["ed25519"])
        ml_dsa_valid = verify_ml_dsa(message, signature["ml_dsa"])
        return ed25519_valid and ml_dsa_valid

# What we LOG (verifiable, privacy-preserving)
logged_data = {
    "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest(),  # ✓
    "risk_category": "NCII_RISK",  # ✓ Categorical
    "risk_score": 0.94,  # ✓ Confidence
    "policy_version": "2026-01-28"  # ✓ Reference
}

# What we DON'T log
not_logged = {
    "raw_prompt": prompt,  # ✗ Privacy violation
    "user_identity": user.email,  # ✗ Privacy violation  
    "detailed_analysis": safety_report.full_text  # ✗ May reveal prompt
}

# Conceptual ZK proof structure
# "I can prove this prompt matched a dangerous pattern without showing you the prompt"

class RefusalZKProof:
    """
    Proves:
    1. prompt_hash corresponds to some prompt P
    2. P matches safety rule R
    3. R is part of policy version V

    Without revealing:
    - The actual prompt P
    - The specific rule pattern
    - Other policy details
    """

    def __init__(self, prompt_hash: str, policy_version: str):
        self.prompt_hash = prompt_hash
        self.policy_version = policy_version

    def generate_proof(self, prompt: str, matched_rule: SafetyRule) -> bytes:
        # In production: use zk-STARK library (no trusted setup, quantum-resistant)
        # This is a conceptual placeholder
        pass

    def verify_proof(self, proof: bytes) -> bool:
        # Anyone can verify without seeing the prompt
        pass

from cryptography.fernet import Fernet

class GDPRCompliantLogger:
    def __init__(self, base_logger: CAPEventLogger):
        self.base_logger = base_logger
        self.user_keys: Dict[str, bytes] = {}  # user_id -> encryption_key

    def _get_or_create_user_key(self, user_id: str) -> bytes:
        if user_id not in self.user_keys:
            self.user_keys[user_id] = Fernet.generate_key()
        return self.user_keys[user_id]

    def log_attempt_gdpr(self, prompt: str, user_id: str) -> CAPEvent:
        """Log with encrypted user context."""
        key = self._get_or_create_user_key(user_id)
        fernet = Fernet(key)

        # Encrypt user-specific data
        encrypted_context = fernet.encrypt(user_id.encode())

        # Log with encrypted context
        event = self.base_logger.log_attempt(prompt, user_id)
        event.user_context_encrypted = encrypted_context.hex()

        return event

    def handle_deletion_request(self, user_id: str):
        """
        GDPR Article 17: Right to Erasure

        By deleting the key, all encrypted user data becomes 
        cryptographically inaccessible (crypto-shredding).
        """
        if user_id in self.user_keys:
            # Secure deletion of key
            key = self.user_keys[user_id]
            # Overwrite memory (in production, use secure_delete library)
            self.user_keys[user_id] = b'x00' * len(key)
            del self.user_keys[user_id]

            return {"status": "deleted", "method": "crypto_shredding"}

        return {"status": "not_found"}

┌─────────────────────────────────────────────────────────────┐
│                  Your AI Service                            │
│                                                             │
│  ┌───────────┐    ┌───────────┐    ┌───────────────┐       │
│  │  Request  │───►│  Safety   │───►│  Generation   │       │
│  │  Handler  │    │  Filter   │    │    Engine     │       │
│  └───────────┘    └───────────┘    └───────────────┘       │
│        │               │                   │               │
│        │      CAP-SRP Sidecar             │               │
│        ▼               ▼                   ▼               │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                Event Interceptor                    │   │
│  │                                                     │   │
│  │  • Logs GEN_ATTEMPT before safety check             │   │
│  │  • Logs GEN_DENY when filter triggers               │   │
│  │  • Logs GEN when content created                    │   │
│  │  • Logs GEN_ERROR on failures                       │   │
│  └─────────────────────────────────────────────────────┘   │
│                          │                                 │
└──────────────────────────┼─────────────────────────────────┘
                           │
                           ▼
              ┌─────────────────────────────┐
              │   Transparency Log Service  │
              │   (SCITT / External)        │
              └─────────────────────────────┘

# FastAPI sidecar example
from fastapi import FastAPI, Request
from typing import Callable

app = FastAPI()
logger = CAPEventLogger(signing_key=load_key(), model_id="image-gen-v3")

class CAPSRPMiddleware:
    def __init__(self, app: FastAPI, logger: CAPEventLogger):
        self.app = app
        self.logger = logger

    async def __call__(self, request: Request, call_next: Callable):
        if request.url.path == "/generate":
            body = await request.json()

            # COMMITMENT POINT: Log attempt before processing
            attempt = self.logger.log_attempt(
                prompt=body["prompt"],
                user_id=request.headers.get("X-User-ID", "anonymous")
            )

            # Add attempt_id to request state
            request.state.cap_attempt_id = attempt.attempt_id

        response = await call_next(request)

        # Outcome logged by endpoint handler
        return response

app.add_middleware(CAPSRPMiddleware, logger=logger)

@app.post("/generate")
async def generate(request: Request, body: GenerateRequest):
    attempt_id = request.state.cap_attempt_id

    # Safety check
    risk = await safety_evaluator.evaluate(body.prompt)

    if risk.should_deny:
        logger.log_denial(attempt_id, risk.category, risk.score, POLICY_VERSION)
        return {"status": "denied", "reason": risk.category.value}

    try:
        output = await generator.generate(body.prompt)
        logger.log_generation(attempt_id, output, attach_c2pa(output))
        return {"status": "success", "image_url": upload(output)}
    except Exception as e:
        logger.log_error(attempt_id, type(e).__name__, "GENERATION")
        raise

class ComplianceTier(Enum):
    SILVER = "silver"
    GOLD = "gold"
    PLATINUM = "platinum"

class CAPSRPClient:
    def __init__(self, tier: ComplianceTier, config: dict):
        self.tier = tier
        self.anchors = self._setup_anchors(tier, config)

    def _setup_anchors(self, tier: ComplianceTier, config: dict):
        anchors = [TSAnchor(config["tsa_url"])]  # All tiers

        if tier in [ComplianceTier.GOLD, ComplianceTier.PLATINUM]:
            anchors.append(MultiLogAnchor(config["log_urls"]))

        if tier == ComplianceTier.PLATINUM:
            anchors.append(BlockchainAnchor(config["blockchain"]))

        return anchors

    async def anchor_batch(self, merkle_root: bytes):
        """Anchor Merkle root to all configured external services."""
        receipts = []
        for anchor in self.anchors:
            receipt = await anchor.anchor(merkle_root)
            receipts.append(receipt)
        return receipts

# Install the SDK
pip install vap-framework

# Or from source
git clone https://github.com/veritaschain/vap-sdk-python
cd vap-sdk-python
pip install -e .

from vap_sdk import CAPClient, RiskCategory

# Initialize client
client = CAPClient(
    model_id="my-image-generator",
    transparency_service="https://log.veritaschain.org"  # Or self-hosted
)

# In your generation flow
async def generate(prompt: str, user_id: str):
    # 1. Log attempt (commitment point)
    attempt = await client.log_attempt(prompt, user_id)

    # 2. Your safety evaluation
    risk = await your_safety_check(prompt)

    # 3. Log outcome
    if risk.is_unsafe:
        await client.log_denial(attempt.id, risk.category, risk.score)
        return None

    output = await your_generator(prompt)
    await client.log_generation(attempt.id, output)
    return output

ATTEMPTS = GENERATIONS + REFUSALS + ERRORS