A consent receipt is a record of a consent transaction provided to the individual (data subject) as evidence that consent was given. The Kantara Initiative Consent Receipt Specification v1.1 defines a standard, machine-readable format that enables individuals to track and manage their consent across multiple organizations. This skill covers the implementation of consent receipts aligned with the Kantara specification and ISO/IEC 27560:2023.
Consent Receipt Structure
Core Fields (Kantara v1.1)
Field
Type
Required
Description
version
String
Yes
Specification version (e.g., "KI-CR-v1.1.0")
jurisdiction
String
Yes
Legal jurisdiction (ISO 3166-1 alpha-2)
相关技能
consentTimestamp
DateTime
Yes
UTC timestamp of consent grant
collectionMethod
String
Yes
How consent was collected (web form, verbal, paper)
"""
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)}")
Receipt Lifecycle Management
Lifecycle States
ISSUED --> ACTIVE --> WITHDRAWN
| | |
| v v
| UPDATED ARCHIVED
| |
v v
EXPIRED ACTIVE (new version)
Lifecycle Manager
"""
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)