Spaces:
Running
Running
| """Optional encrypted recording retention. No public audio URLs or exports. | |
| The main research receipt never implicitly authorizes recording retention. | |
| This supplementary policy needs its own affirmative, server-recorded consent. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import base64 | |
| import hashlib | |
| import json | |
| import os | |
| import secrets | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Annotated, Literal | |
| from fastapi import APIRouter, Header, HTTPException, Request, Response | |
| from pydantic import Field | |
| from src.collection_v2 import ( | |
| IdempotencyKey, ParticipantToken, StrictModel, get_store, _new_id, | |
| _utc_now, _raise_api_error, _request_hash, | |
| ) | |
| from src.consent_policy import get_current_consent_policy | |
| router = APIRouter(prefix="/api/v2/raw-audio", tags=["private-raw-audio"]) | |
| MAX_BYTES = 8 * 1024 * 1024 | |
| MAX_PARTICIPANT_BYTES = 128 * 1024 * 1024 | |
| POLICY_VERSION = "private-recording-game-session-2026-10-06" | |
| SESSION_HOURS = 12 | |
| ParticipantId = Annotated[str, Header(alias="X-Participant-ID", pattern=r"^participant_[0-9a-f]{32}$")] | |
| AudioConsentId = Annotated[str, Header(alias="X-Audio-Consent-ID", pattern=r"^acn_[0-9a-f]{32}$")] | |
| AudioSessionId = Annotated[str, Header(alias="X-Audio-Session-ID", pattern=r"^asn_[0-9a-f]{32}$")] | |
| GameId = Annotated[str, Header(alias="X-Game-ID", min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")] | |
| def configuration(): | |
| try: | |
| enabled = os.getenv("RAW_AUDIO_COLLECTION_ENABLED", "").lower() == "true" | |
| days = int(os.getenv("RAW_AUDIO_RETENTION_DAYS", "0")) | |
| key = base64.b64decode(os.getenv("RAW_AUDIO_ENCRYPTION_KEY", ""), validate=True) | |
| from cryptography.hazmat.primitives.ciphers.aead import AESGCM | |
| if not enabled or not 1 <= days <= 3650 or len(key) != 32 or not os.getenv("COLLECTION_DB_PATH", "").strip(): | |
| return None | |
| return days, AESGCM(key) | |
| except (ValueError, ImportError): | |
| return None | |
| def policy(): | |
| configured = configuration() | |
| days = configured[0] if configured else None | |
| notice = ( | |
| f"I separately authorize private storage of recordings for speech, dialect, " | |
| f"and acoustic research for up to {days} days from upload, or earlier if I withdraw. " | |
| "Before each game and after returning from closing, minimizing, or backgrounding the app, " | |
| "I choose whether to store completed recordings automatically for that game session. " | |
| "This applies only to recordings I start with a game microphone; it does not turn on the microphone " | |
| "or authorize background storage. No per-recording Save confirmation follows. " | |
| f"Session permission expires after {SESSION_HOURS} hours at the latest and can be stopped sooner. " | |
| "I acknowledge that my voice may identify me or reveal sensitive traits. " | |
| "Only my own voice may be submitted through this option. Recordings are not public, " | |
| "are not covered by the text dataset's public-release permission, and will not be used " | |
| "to identify people, clone voices, or train models under this permission. " | |
| "I can decline without losing access to games or text contribution. " | |
| "Withdrawal removes active recording payloads; any backup erasure must be separately tracked." | |
| ) if configured else "Private recording storage is not configured. No recordings will be retained." | |
| canonical = json.dumps({"version": POLICY_VERSION, "retention_days": days, "notice": notice}, sort_keys=True, separators=(",", ":")) | |
| return {"enabled": configured is not None, "version": POLICY_VERSION, "retention_days": days, | |
| "notice": notice, "canonical_text": canonical, | |
| "policy_sha256": hashlib.sha256(canonical.encode()).hexdigest(), "max_bytes": MAX_BYTES} | |
| def require_configuration(): | |
| value = configuration() | |
| if value is None: | |
| raise HTTPException(503, "Private recording retention is not enabled and configured") | |
| return value | |
| def authenticate(store, connection, participant_id, participant_token, *, active=True): | |
| store._authenticate_participant(connection, participant_id, participant_token) | |
| if active: | |
| store._ensure_active(connection, participant_id) | |
| def require_audio_consent(store, connection, audio_consent_id, participant_id): | |
| consent = connection.execute("SELECT * FROM raw_audio_consents WHERE audio_consent_id = ? AND participant_id = ?", | |
| (audio_consent_id, participant_id)).fetchone() | |
| if consent is None or consent["revoked_at"]: | |
| raise HTTPException(403, "An active recording-storage consent receipt is required") | |
| if consent["policy_sha256"] != policy()["policy_sha256"]: | |
| raise HTTPException(403, "Review the current recording-storage policy before uploading") | |
| store._require_consent(connection, consent["consent_id"], participant_id, "speech", consent["app_id"]) | |
| require_current_base_consent(connection, participant_id, consent["app_id"], consent["consent_id"]) | |
| return consent | |
| def require_current_base_consent(connection, participant_id, app_id, consent_id): | |
| latest = connection.execute("SELECT consent_id FROM consents WHERE participant_id = ? AND lower(app_id) = lower(?) " | |
| "ORDER BY rowid DESC LIMIT 1", (participant_id, app_id)).fetchone() | |
| if latest is None or latest["consent_id"] != consent_id: | |
| raise HTTPException(403, "Research choices changed; renew recording permission under the latest receipt") | |
| def purge_expired(): | |
| store = get_store() | |
| store.initialize() | |
| with store._connect() as connection: | |
| connection.execute("PRAGMA secure_delete = ON") | |
| connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = ?, " | |
| "erasure_reason = 'expired' WHERE erased_at IS NULL AND expires_at <= ?", (_utc_now(), _utc_now())) | |
| def install_retention_worker(app): | |
| async def start_audio_retention(): | |
| if not os.getenv("COLLECTION_DB_PATH", "").strip(): | |
| return | |
| async def sweep(): | |
| while True: | |
| try: | |
| await asyncio.to_thread(purge_expired) | |
| except Exception: | |
| # No recording content, participant identifiers, paths or secrets in logs. | |
| print("Private recording expiry sweep failed; retry scheduled.") | |
| await asyncio.sleep(60) | |
| app.state.raw_audio_retention_task = asyncio.create_task(sweep()) | |
| async def stop_audio_retention(): | |
| task = getattr(app.state, "raw_audio_retention_task", None) | |
| if task: | |
| task.cancel() | |
| try: | |
| await task | |
| except asyncio.CancelledError: | |
| pass | |
| class AudioConsentCreate(StrictModel): | |
| participant_id: str = Field(pattern=r"^participant_[0-9a-f]{32}$") | |
| consent_id: str = Field(pattern=r"^cns_[0-9a-f]{32}$") | |
| app_id: str = Field(min_length=1, max_length=64) | |
| policy_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") | |
| authorize_recording_storage: Literal[True] | |
| voice_privacy_risks_acknowledged: Literal[True] | |
| class AudioSessionCreate(StrictModel): | |
| audio_consent_id: str = Field(pattern=r"^acn_[0-9a-f]{32}$") | |
| game_id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$") | |
| automatic_storage_confirmed: Literal[True] | |
| own_voice_only_confirmed: Literal[True] | |
| def require_supported_game(game_id): | |
| if game_id.upper() in {"HOME", "BRIDGE", "CONVO-BRIDGE", "CONVO_BRIDGE", "UNKNOWN"}: | |
| raise HTTPException(403, "Shared-device and unassigned recordings cannot be retained") | |
| def require_audio_session(connection, session_id, participant_id, consent_id, game_id): | |
| row = connection.execute("SELECT * FROM raw_audio_sessions WHERE audio_session_id = ?", (session_id,)).fetchone() | |
| if (row is None or row["participant_id"] != participant_id or row["audio_consent_id"] != consent_id | |
| or row["game_id"] != game_id or row["ended_at"] or row["expires_at"] <= _utc_now()): | |
| raise HTTPException(403, "Confirm recording storage for this game session before uploading") | |
| return row | |
| def create_audio_session(payload: AudioSessionCreate, participant_id: ParticipantId, | |
| participant_token: ParticipantToken, idempotency_key: IdempotencyKey): | |
| require_configuration() | |
| require_supported_game(payload.game_id) | |
| store = get_store(); store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, participant_id, participant_token) | |
| require_audio_consent(store, connection, payload.audio_consent_id, participant_id) | |
| request_hash = _request_hash({**payload.model_dump(), "participant_id": participant_id}) | |
| replay = store._idempotent_replay(connection, "audio_session", idempotency_key, request_hash) | |
| if replay: | |
| require_audio_session(connection, replay["audio_session_id"], participant_id, payload.audio_consent_id, payload.game_id) | |
| return replay | |
| now = datetime.now(timezone.utc) | |
| receipt = {"audio_session_id": _new_id("asn"), "participant_id": participant_id, | |
| "audio_consent_id": payload.audio_consent_id, "game_id": payload.game_id, | |
| "created_at": now.isoformat(timespec="milliseconds").replace("+00:00", "Z"), | |
| "expires_at": (now + timedelta(hours=SESSION_HOURS)).isoformat(timespec="milliseconds").replace("+00:00", "Z")} | |
| connection.execute("INSERT INTO raw_audio_sessions VALUES (?, ?, ?, ?, ?, ?, ?, NULL)", | |
| (receipt["audio_session_id"], participant_id, payload.audio_consent_id, payload.game_id, | |
| json.dumps(payload.model_dump(), sort_keys=True), receipt["created_at"], receipt["expires_at"])) | |
| store._record_idempotency(connection, "audio_session", idempotency_key, request_hash, | |
| "audio_session", receipt["audio_session_id"], receipt, receipt["created_at"]) | |
| connection.commit() | |
| return receipt | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def end_audio_session(audio_session_id: str, participant_id: ParticipantId, participant_token: ParticipantToken): | |
| store = get_store(); store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, participant_id, participant_token, active=False) | |
| row = connection.execute("SELECT 1 FROM raw_audio_sessions WHERE audio_session_id = ? AND participant_id = ?", | |
| (audio_session_id, participant_id)).fetchone() | |
| if row is None: raise HTTPException(404, "Recording session not found") | |
| connection.execute("UPDATE raw_audio_sessions SET ended_at = COALESCE(ended_at, ?) WHERE audio_session_id = ?", | |
| (_utc_now(), audio_session_id)) | |
| connection.commit() | |
| return {"audio_session_id": audio_session_id, "status": "ended"} | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def get_audio_policy(): | |
| return policy() | |
| def create_audio_consent(payload: AudioConsentCreate, participant_token: ParticipantToken, idempotency_key: IdempotencyKey): | |
| require_configuration() | |
| statement = policy() | |
| if payload.policy_sha256 != statement["policy_sha256"]: | |
| raise HTTPException(409, "Recording-storage policy changed; review it again") | |
| store = get_store() | |
| store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, payload.participant_id, participant_token) | |
| base = store._require_consent(connection, payload.consent_id, payload.participant_id, "speech", payload.app_id) | |
| require_current_base_consent(connection, payload.participant_id, payload.app_id, payload.consent_id) | |
| if base["consent_version"] != get_current_consent_policy()["consent_version"]: | |
| raise HTTPException(403, "Renew the research consent under the current policy first") | |
| request_hash = _request_hash(payload.model_dump()) | |
| replay = store._idempotent_replay(connection, "audio_consent", idempotency_key, request_hash) | |
| if replay: | |
| # Do not revive a revoked receipt through an idempotent retry. | |
| require_audio_consent(store, connection, replay["audio_consent_id"], payload.participant_id) | |
| return replay | |
| receipt = {"audio_consent_id": _new_id("acn"), "participant_id": payload.participant_id, | |
| "consent_id": payload.consent_id, "policy_sha256": statement["policy_sha256"], | |
| "retention_days": statement["retention_days"], "created_at": _utc_now()} | |
| connection.execute("INSERT INTO raw_audio_consents VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL)", | |
| (receipt["audio_consent_id"], payload.participant_id, payload.consent_id, payload.app_id, | |
| statement["policy_sha256"], statement["canonical_text"], statement["retention_days"], receipt["created_at"])) | |
| store._record_idempotency(connection, "audio_consent", idempotency_key, request_hash, "audio_consent", | |
| receipt["audio_consent_id"], receipt, receipt["created_at"]) | |
| connection.commit() | |
| return receipt | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def detect_audio(data, media_type): | |
| media_type = media_type.split(";", 1)[0].strip().lower() | |
| valid = ((media_type in {"audio/wav", "audio/x-wav"} and len(data) >= 44 and data[:4] == b"RIFF" and data[8:12] == b"WAVE") | |
| or (media_type == "audio/webm" and data[:4] == b"\x1aE\xdf\xa3") | |
| or (media_type == "audio/ogg" and data[:4] == b"OggS") | |
| or (media_type == "audio/mp4" and len(data) >= 12 and data[4:8] == b"ftyp")) | |
| if not valid: | |
| raise HTTPException(415, "A supported WAV, WebM, Ogg or MP4 audio recording is required") | |
| return media_type | |
| async def upload_recording(request: Request, participant_id: ParticipantId, participant_token: ParticipantToken, | |
| audio_consent_id: AudioConsentId, game_id: GameId, idempotency_key: IdempotencyKey, | |
| audio_session_id: AudioSessionId): | |
| days, cipher = require_configuration() | |
| require_supported_game(game_id) | |
| store = get_store() | |
| store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| authenticate(store, connection, participant_id, participant_token) | |
| require_audio_consent(store, connection, audio_consent_id, participant_id) | |
| require_audio_session(connection, audio_session_id, participant_id, audio_consent_id, game_id) | |
| data = bytearray() | |
| async for chunk in request.stream(): | |
| if len(data) + len(chunk) > MAX_BYTES: | |
| raise HTTPException(413, "Recording exceeds the size limit") | |
| data.extend(chunk) | |
| media_type = detect_audio(data, request.headers.get("content-type", "")) | |
| content_hash = hashlib.sha256(data).hexdigest() | |
| request_hash = _request_hash({"participant_id": participant_id, "audio_consent_id": audio_consent_id, | |
| "audio_session_id": audio_session_id, "game_id": game_id, | |
| "media_type": media_type, "content_hash": content_hash}) | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, participant_id, participant_token) | |
| require_audio_consent(store, connection, audio_consent_id, participant_id) | |
| require_audio_session(connection, audio_session_id, participant_id, audio_consent_id, game_id) | |
| replay = store._idempotent_replay(connection, "audio_upload", idempotency_key, request_hash) | |
| if replay: | |
| row = connection.execute("SELECT erased_at, expires_at FROM raw_audio_recordings WHERE audio_id = ?", (replay["audio_id"],)).fetchone() | |
| if row is None or row["erased_at"] or row["expires_at"] <= _utc_now(): | |
| raise HTTPException(410, "This recording has been erased and cannot be replayed") | |
| return replay | |
| used = connection.execute("SELECT COALESCE(SUM(LENGTH(ciphertext)), 0) FROM raw_audio_recordings WHERE participant_id = ?", (participant_id,)).fetchone()[0] | |
| if used + len(data) + 16 > MAX_PARTICIPANT_BYTES: | |
| raise HTTPException(413, "Participant recording storage limit reached") | |
| audio_id, nonce = _new_id("aud"), secrets.token_bytes(12) | |
| now = datetime.now(timezone.utc) | |
| expires = (now + timedelta(days=days)).isoformat(timespec="milliseconds").replace("+00:00", "Z") | |
| ciphertext = cipher.encrypt(nonce, bytes(data), (audio_id + ":" + participant_id).encode()) | |
| connection.execute("INSERT INTO raw_audio_recordings VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL)", | |
| (audio_id, participant_id, audio_consent_id, game_id, media_type, nonce, ciphertext, now.isoformat(timespec="milliseconds").replace("+00:00", "Z"), expires)) | |
| connection.execute("INSERT INTO raw_audio_recording_sessions VALUES (?, ?)", (audio_id, audio_session_id)) | |
| result = {"audio_id": audio_id, "audio_session_id": audio_session_id, "expires_at": expires, | |
| "status": "privately_stored", "public_release": False} | |
| store._record_idempotency(connection, "audio_upload", idempotency_key, request_hash, "audio_recording", audio_id, result, now.isoformat(timespec="milliseconds")) | |
| connection.commit() | |
| return result | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def list_recordings(participant_id: ParticipantId, participant_token: ParticipantToken): | |
| store = get_store(); store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| authenticate(store, connection, participant_id, participant_token, active=False) | |
| rows = connection.execute("SELECT audio_id, game_id, created_at, expires_at, erased_at, erasure_reason " | |
| "FROM raw_audio_recordings WHERE participant_id = ? ORDER BY created_at DESC LIMIT 200", (participant_id,)).fetchall() | |
| return {"recordings": [dict(row) for row in rows]} | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def delete_recording(audio_id: str, participant_id: ParticipantId, participant_token: ParticipantToken): | |
| store = get_store(); store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, participant_id, participant_token, active=False) | |
| connection.execute("PRAGMA secure_delete = ON") | |
| row = connection.execute("SELECT 1 FROM raw_audio_recordings WHERE audio_id = ? AND participant_id = ?", (audio_id, participant_id)).fetchone() | |
| if row is None: raise HTTPException(404, "Recording not found") | |
| connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = COALESCE(erased_at, ?), " | |
| "erasure_reason = COALESCE(erasure_reason, 'participant_request') WHERE audio_id = ?", (_utc_now(), audio_id)) | |
| connection.commit() | |
| return {"audio_id": audio_id, "active_payload": "erased", "backup_erasure": "not_attested"} | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def revoke_audio_consent(audio_consent_id: str, participant_id: ParticipantId, participant_token: ParticipantToken): | |
| store = get_store(); store.initialize() | |
| try: | |
| with store._connect() as connection: | |
| store._begin(connection) | |
| authenticate(store, connection, participant_id, participant_token, active=False) | |
| row = connection.execute("SELECT app_id FROM raw_audio_consents WHERE audio_consent_id = ? AND participant_id = ?", (audio_consent_id, participant_id)).fetchone() | |
| if row is None: raise HTTPException(404, "Recording-storage consent not found") | |
| connection.execute("PRAGMA secure_delete = ON") | |
| # Revoke every receipt for this app, including retries and other tabs. | |
| connection.execute("UPDATE raw_audio_sessions SET ended_at = COALESCE(ended_at, ?) WHERE audio_consent_id IN " | |
| "(SELECT audio_consent_id FROM raw_audio_consents WHERE participant_id = ? AND app_id = ?)", | |
| (_utc_now(), participant_id, row["app_id"])) | |
| connection.execute("UPDATE raw_audio_consents SET revoked_at = COALESCE(revoked_at, ?) WHERE participant_id = ? AND app_id = ?", (_utc_now(), participant_id, row["app_id"])) | |
| connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = COALESCE(erased_at, ?), " | |
| "erasure_reason = COALESCE(erasure_reason, 'audio_consent_revoked') WHERE audio_consent_id IN " | |
| "(SELECT audio_consent_id FROM raw_audio_consents WHERE participant_id = ? AND app_id = ?)", (_utc_now(), participant_id, row["app_id"])) | |
| connection.commit() | |
| return {"audio_consent_id": audio_consent_id, "status": "revoked", "active_payloads": "erased", "backup_erasure": "not_attested"} | |
| except Exception as exc: | |
| _raise_api_error(exc) | |