consent-receipt-spec

{
  "version": "KI-CR-v1.1.0",
  "jurisdiction": "EU",
  "consentTimestamp": "2026-03-14T10:30:00.000Z",
  "collectionMethod": "web_form",
  "consentReceiptID": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "language": "en",
  "piiPrincipalId": "user-98765",
  "piiControllers": [
    {
      "piiController": "Cipher Engineering Labs",
      "onBehalf": false,
      "contact": "Data Protection Officer",
      "address": {
        "streetAddress": "100 Technology Drive",
        "locality": "London",
        "region": "Greater London",
        "postalCode": "EC2A 1NT",
        "country": "GB"
      },
      "email": "dpo@cipherengineeringlabs.com",
      "phone": "+44-20-7946-0958",
      "piiControllerUrl": "https://www.cipherengineeringlabs.com"
    }
  ],
  "policyUrl": "https://www.cipherengineeringlabs.com/privacy-policy",
  "services": [
    {
      "service": "Privacy Analytics Platform",
      "purposes": [
        {
          "purpose": "Provide personalized privacy compliance recommendations",
          "purposeCategory": ["core_service"],
          "consentType": "explicit",
          "piiCategory": ["contact_information", "professional_information"],
          "primaryPurpose": true,
          "termination": "Account settings > Privacy > Withdraw consent",
          "thirdPartyDisclosure": false
        },
        {
          "purpose": "Send product updates and feature announcements",
          "purposeCategory": ["marketing"],
          "consentType": "explicit",
          "piiCategory": ["contact_information"],
          "primaryPurpose": false,
          "termination": "Unsubscribe link in email or Account settings",
          "thirdPartyDisclosure": false
        },
        {
          "purpose": "Aggregate usage analytics to improve platform features",
          "purposeCategory": ["analytics"],
          "consentType": "explicit",
          "piiCategory": ["usage_data", "device_information"],
          "primaryPurpose": false,
          "termination": "Account settings > Privacy > Analytics opt-out",
          "thirdPartyDisclosure": true,
          "thirdPartyName": "Cipher Analytics Processing Ltd"
        }
      ]
    }
  ],
  "sensitive": false,
  "spiCat": []
}

"""
Consent receipt generation and verification using JWT (JSON Web Tokens).
Implements Kantara Initiative Consent Receipt Specification v1.1
with cryptographic verification.
"""

import json
import uuid
from datetime import datetime, timezone
from typing import Optional
import jwt  # PyJWT
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.backends import default_backend


class ConsentReceiptIssuer:
    """
    Issue and sign consent receipts as JWTs.
    Uses RS256 (RSA with SHA-256) for signing.
    """

    def __init__(self, private_key_pem: bytes, issuer_name: str):
        """
        Args:
            private_key_pem: PEM-encoded RSA private key
            issuer_name: Name of the issuing organization
        """
        self.private_key = serialization.load_pem_private_key(
            private_key_pem, password=None, backend=default_backend()
        )
        self.public_key = self.private_key.public_key()
        self.issuer_name = issuer_name

    def issue_receipt(
        self,
        principal_id: str,
        services: list[dict],
        jurisdiction: str,
        collection_method: str,
        policy_url: str,
        sensitive: bool = False,
        language: str = "en",
        controller_info: dict = None
    ) -> tuple[str, str]:
        """
        Generate a signed consent receipt.

        Args:
            principal_id: Data subject identifier
            services: List of service/purpose objects
            jurisdiction: Legal jurisdiction code
            collection_method: How consent was collected
            policy_url: URL of the privacy policy
            sensitive: Whether special categories are involved
            language: Language of consent interaction
            controller_info: PII controller details

        Returns:
            Tuple of (receipt_id, signed_jwt_string)
        """
        receipt_id = str(uuid.uuid4())

        if controller_info is None:
            controller_info = {
                "piiController": self.issuer_name,
                "onBehalf": False,
                "contact": "Data Protection Officer",
                "email": f"dpo@{self.issuer_name.lower().replace(' ', '')}.com"
            }

        receipt_payload = {
            "version": "KI-CR-v1.1.0",
            "jurisdiction": jurisdiction,
            "consentTimestamp": datetime.now(timezone.utc).isoformat(),
            "collectionMethod": collection_method,
            "consentReceiptID": receipt_id,
            "language": language,
            "piiPrincipalId": principal_id,
            "piiControllers": [controller_info],
            "policyUrl": policy_url,
            "services": services,
            "sensitive": sensitive,
            "spiCat": [],
            # JWT standard claims
            "iss": self.issuer_name,
            "sub": principal_id,
            "iat": int(datetime.now(timezone.utc).timestamp()),
            "jti": receipt_id,
        }

        # Sign the receipt as a JWT
        signed_token = jwt.encode(
            receipt_payload,
            self.private_key,
            algorithm="RS256",
            headers={"kid": f"{self.issuer_name}-consent-key-1"}
        )

        return receipt_id, signed_token

    def get_public_key_pem(self) -> str:
        """Export the public key for verification by data subjects."""
        return self.public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode("utf-8")


class ConsentReceiptVerifier:
    """
    Verify signed consent receipts.
    Used by data subjects or auditors to confirm receipt authenticity.
    """

    def __init__(self):
        self.trusted_keys: dict[str, bytes] = {}

    def register_issuer_key(self, issuer_name: str, public_key_pem: str):
        """Register a trusted issuer's public key."""
        self.trusted_keys[issuer_name] = public_key_pem.encode("utf-8")

    def verify_receipt(self, token: str) -> tuple[bool, Optional[dict], Optional[str]]:
        """
        Verify a signed consent receipt JWT.

        Returns:
            Tuple of (is_valid, decoded_payload_or_None, error_message_or_None)
        """
        try:
            # Decode without verification first to get the issuer
            unverified = jwt.decode(token, options={"verify_signature": False})
            issuer = unverified.get("iss")

            if issuer not in self.trusted_keys:
                return (False, None, f"Unknown issuer: {issuer}")

            public_key = serialization.load_pem_public_key(
                self.trusted_keys[issuer],
                backend=default_backend()
            )

            # Verify the signature
            decoded = jwt.decode(
                token,
                public_key,
                algorithms=["RS256"],
                issuer=issuer
            )

            # Validate required Kantara fields
            required_fields = [
                "version", "jurisdiction", "consentTimestamp",
                "collectionMethod", "consentReceiptID", "piiPrincipalId",
                "piiControllers", "policyUrl", "services", "sensitive"
            ]

            missing = [f for f in required_fields if f not in decoded]
            if missing:
                return (False, decoded, f"Missing required fields: {missing}")

            return (True, decoded, None)

        except jwt.ExpiredSignatureError:
            return (False, None, "Receipt JWT has expired")
        except jwt.InvalidSignatureError:
            return (False, None, "Invalid signature — receipt may have been tampered with")
        except jwt.DecodeError as e:
            return (False, None, f"Failed to decode JWT: {str(e)}")

ISSUED --> ACTIVE --> WITHDRAWN
  |          |           |
  |          v           v
  |       UPDATED    ARCHIVED
  |          |
  v          v
EXPIRED   ACTIVE (new version)

"""
Manage the lifecycle of consent receipts including
updates, withdrawals, and expiration tracking.
"""

from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum


class ReceiptStatus(Enum):
    ISSUED = "issued"
    ACTIVE = "active"
    UPDATED = "updated"
    WITHDRAWN = "withdrawn"
    EXPIRED = "expired"
    ARCHIVED = "archived"


@dataclass
class ReceiptRecord:
    receipt_id: str
    principal_id: str
    status: ReceiptStatus
    issued_at: datetime
    last_updated: datetime
    withdrawn_at: datetime | None
    jwt_token: str
    version: int
    superseded_by: str | None
    purposes: list[str]


class ConsentReceiptLifecycle:
    """
    Manage consent receipt state transitions and history.
    """

    def __init__(self, receipt_store, issuer: ConsentReceiptIssuer):
        self.store = receipt_store
        self.issuer = issuer

    def activate_receipt(self, receipt_id: str) -> bool:
        """Transition a receipt from ISSUED to ACTIVE."""
        record = self.store.get(receipt_id)
        if record and record.status == ReceiptStatus.ISSUED:
            record.status = ReceiptStatus.ACTIVE
            record.last_updated = datetime.now(timezone.utc)
            self.store.update(record)
            return True
        return False

    def update_receipt(
        self,
        receipt_id: str,
        updated_services: list[dict],
        jurisdiction: str,
        policy_url: str
    ) -> str | None:
        """
        Update an existing receipt by issuing a new version.
        The old receipt is marked as UPDATED and linked to the new one.

        Returns the new receipt ID or None if update failed.
        """
        old_record = self.store.get(receipt_id)
        if not old_record or old_record.status not in [
            ReceiptStatus.ACTIVE, ReceiptStatus.ISSUED
        ]:
            return None

        # Issue new receipt
        new_receipt_id, new_jwt = self.issuer.issue_receipt(
            principal_id=old_record.principal_id,
            services=updated_services,
            jurisdiction=jurisdiction,
            collection_method="consent_update",
            policy_url=policy_url
        )

        # Create new record
        new_record = ReceiptRecord(
            receipt_id=new_receipt_id,
            principal_id=old_record.principal_id,
            status=ReceiptStatus.ACTIVE,
            issued_at=datetime.now(timezone.utc),
            last_updated=datetime.now(timezone.utc),
            withdrawn_at=None,
            jwt_token=new_jwt,
            version=old_record.version + 1,
            superseded_by=None,
            purposes=[p["purpose"] for s in updated_services for p in s.get("purposes", [])]
        )
        self.store.save(new_record)

        # Mark old receipt as updated
        old_record.status = ReceiptStatus.UPDATED
        old_record.superseded_by = new_receipt_id
        old_record.last_updated = datetime.now(timezone.utc)
        self.store.update(old_record)

        return new_receipt_id

    def withdraw_consent(self, receipt_id: str, reason: str = "") -> bool:
        """
        Withdraw consent and mark the receipt accordingly.

        Returns True if withdrawal was successful.
        """
        record = self.store.get(receipt_id)
        if not record or record.status not in [
            ReceiptStatus.ACTIVE, ReceiptStatus.ISSUED
        ]:
            return False

        record.status = ReceiptStatus.WITHDRAWN
        record.withdrawn_at = datetime.now(timezone.utc)
        record.last_updated = datetime.now(timezone.utc)
        self.store.update(record)

        return True

    def get_active_receipts(self, principal_id: str) -> list[ReceiptRecord]:
        """Get all active consent receipts for a data subject."""
        return self.store.find_by_principal(
            principal_id, status=ReceiptStatus.ACTIVE
        )

    def get_receipt_history(self, principal_id: str) -> list[ReceiptRecord]:
        """Get the complete consent receipt history for a data subject."""
        return self.store.find_by_principal(principal_id)