Spaces:
Running
Running
| """Canonical, versioned policy registry for crowdsourcing consent receipts.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| from functools import lru_cache | |
| from pathlib import Path | |
| from typing import Any | |
| POLICY_ROOT = Path(__file__).resolve().parents[1] / "consent_policies" | |
| MANIFEST_PATH = POLICY_ROOT / "manifest.json" | |
| def _load_manifest() -> dict[str, Any]: | |
| with MANIFEST_PATH.open(encoding="utf-8") as manifest_file: | |
| return json.load(manifest_file) | |
| def get_consent_policy( | |
| consent_version: str | None = None, | |
| consent_locale: str = "en", | |
| ) -> dict[str, Any]: | |
| """Return one verified policy plus the exact statement shown to participants.""" | |
| manifest = _load_manifest() | |
| version = consent_version or str(manifest["current_version"]) | |
| try: | |
| policy = manifest["policies"][version] | |
| locale_policy = policy["locales"][consent_locale] | |
| except KeyError as exc: | |
| raise ValueError("unknown consent policy version or locale") from exc | |
| canonical_text = str(locale_policy["canonical_text"]) | |
| digest = hashlib.sha256(canonical_text.encode("utf-8")).hexdigest() | |
| expected_digest = str(locale_policy["consent_text_sha256"]) | |
| if digest != expected_digest: | |
| raise RuntimeError( | |
| f"consent policy hash mismatch for {version}/{consent_locale}" | |
| ) | |
| canonical_payload = json.loads(canonical_text) | |
| expected_payload = { | |
| "consent_version": version, | |
| "privacy_notice_version": policy["privacy_notice_version"], | |
| "retention_policy_version": policy["retention_policy_version"], | |
| "withdrawal_policy_version": policy["withdrawal_policy_version"], | |
| "dataset_license": policy["dataset_license"], | |
| "consent_surface": policy["consent_surface"], | |
| "research_purposes": policy["research_purposes"], | |
| "data_recipient_categories": policy["data_recipient_categories"], | |
| "notice": policy["notice"], | |
| } | |
| if canonical_payload != expected_payload: | |
| raise RuntimeError( | |
| f"canonical consent text differs from manifest fields for " | |
| f"{version}/{consent_locale}" | |
| ) | |
| return { | |
| "consent_version": version, | |
| "consent_locale": consent_locale, | |
| "consent_text_sha256": digest, | |
| "canonical_text": canonical_text, | |
| **{ | |
| key: value | |
| for key, value in policy.items() | |
| if key not in {"locales", "allowed_modalities"} | |
| }, | |
| "allowed_modalities": list(policy["allowed_modalities"]), | |
| } | |
| def get_current_consent_policy(consent_locale: str = "en") -> dict[str, Any]: | |
| return get_consent_policy(None, consent_locale) | |
| def validate_consent_policy_payload(payload: dict[str, Any]) -> None: | |
| """Reject receipts whose disclosed policy differs from the registry.""" | |
| policy = get_consent_policy( | |
| str(payload.get("consent_version", "")), | |
| str(payload.get("consent_locale", "")), | |
| ) | |
| scalar_fields = ( | |
| "consent_surface", | |
| "privacy_notice_version", | |
| "dataset_license", | |
| "retention_policy_version", | |
| "withdrawal_policy_version", | |
| "data_controller_name", | |
| "data_controller_contact", | |
| "consent_text_sha256", | |
| ) | |
| for field in scalar_fields: | |
| if payload.get(field) != policy[field]: | |
| raise ValueError(f"{field} does not match the registered consent policy") | |
| for field in ("research_purposes", "data_recipient_categories"): | |
| supplied = payload.get(field, []) | |
| if not isinstance(supplied, list) or not supplied or not all( | |
| isinstance(item, str) and item.strip() for item in supplied | |
| ): | |
| raise ValueError(f"{field} must contain known non-empty strings") | |
| if sorted(supplied) != sorted(policy[field]): | |
| raise ValueError(f"{field} does not match the registered consent policy") | |
| supplied_modalities = payload.get("permitted_modalities", []) | |
| if not isinstance(supplied_modalities, list) or not supplied_modalities or not all( | |
| isinstance(item, str) and item.strip() for item in supplied_modalities | |
| ): | |
| raise ValueError("permitted_modalities must contain known non-empty strings") | |
| permitted_modalities = set(supplied_modalities) | |
| if not permitted_modalities.issubset(set(policy["allowed_modalities"])): | |
| raise ValueError("permitted_modalities contains a modality not in the policy") | |
| if payload.get("raw_audio_consent"): | |
| if not policy.get("raw_audio_collection_allowed", False): | |
| raise ValueError("this consent policy does not authorize raw audio") | |
| if not permitted_modalities.intersection({"speech", "multimodal"}): | |
| raise ValueError( | |
| "raw_audio_consent requires speech or multimodal consent" | |
| ) | |