Serve call audio through c2-core instead of GCS signed URLs

upload_audio() could only sign a URL when GCP_CREDENTIALS_PATH pointed at a
service-account key file. The deployed VM runs on Application Default
Credentials with no key file, so every upload silently took the fallback
branch and returned a bare gs:// URI. That broke two things at once:

  * Browsers cannot fetch a gs:// URI, so no recording was ever playable.
  * _public_url_to_gcs_uri() only matched https://storage.googleapis.com/ and
    returned None for it, so `if gcs_uri:` in the upload path was always false
    and transcription never ran. Nothing was logged, which is why this looked
    like an OpenAI credits problem rather than a storage one.

The fallback also interpolated the client-supplied filename instead of the
call_id-derived safe name, so the URI did not even name the object written.

Calls now store only the canonical gs:// location. A short-lived playback link
is minted per read as an HMAC over (call_id, expiry) keyed by SERVICE_KEY, and
audio is served from the private bucket by the new /media route. An <audio src>
cannot carry an Authorization header, so the link has to be the credential;
that router is therefore public with the check done inline, as enrollment.py
already does. Signing GCS URLs from the VM would have needed a
serviceAccountTokenCreator grant on its own service account — this avoids the
IAM change entirely and keeps the bucket private.

gcs_uri_for_call() reconstructs the object name from call_id, so recordings
made before this fix are reachable again without a data migration.

Frontend rows come straight from Firestore via onSnapshot and never see a
server-minted field, so CallRow fetches the link lazily on expand.

Also removes the last long-lived (1 year) signed URL and the log line that
printed it.

drb-c2-core/app/internal/storage.py

@@ -1,9 +1,40 @@
+"""
+Call-audio storage and playback links.
+
+TWO THINGS THIS MODULE DELIBERATELY DOES NOT DO ANY MORE:
+
+1. It does not return a GCS *signed* URL from the upload path. Signing needs a
+   service-account private key, and the deployed VM runs on Application Default
+   Credentials with no key file (see ansible c2-core.env.j2). The old code
+   silently fell back to returning a bare ``gs://`` URI, which broke two things
+   at once: browsers can't fetch a gs:// URI, so no recording was ever
+   playable, and ``_public_url_to_gcs_uri`` in upload.py returned None for it,
+   so the transcription step was skipped without logging anything at all.
+
+2. It does not store a long-lived URL on the call document. What gets persisted
+   is the canonical ``gs://`` object location; a short-lived playback link is
+   minted on read instead. Nothing durable and nothing loggable is a credential.
+
+Playback goes through c2-core's own /media route rather than GCS directly,
+because an <audio src> cannot carry an Authorization header — so the link
+itself has to be the credential. It is a plain HMAC over (call_id, expiry)
+keyed by SERVICE_KEY, which costs no network round-trip, keeps the bucket
+fully private, and needs no IAM change on the VM's service account.
+"""
 import asyncio
-import datetime
-from typing import Optional
+import hashlib
+import hmac
+import os
+import time
+from typing import Optional, Tuple
 from app.config import settings
 from app.internal.logger import logger
 
+# Domain separation: the audio-link key is derived from SERVICE_KEY rather than
+# being SERVICE_KEY itself, so a leaked playback link can never be replayed as
+# a service-key bearer token against the rest of the API.
+_KEY_CONTEXT = b"drb-audio-link-v1"
+
 
 def _safe_audio_filename(filename: str, call_id: str) -> str:
     """Return a safe GCS object name derived from the call_id.
@@ -12,7 +43,6 @@ def _safe_audio_filename(filename: str, call_id: str) -> str:
     call_id (which we control) to prevent path traversal via crafted filenames.
     The original extension is preserved only if it's a known audio type.
     """
-    import os
     ext = os.path.splitext(filename)[-1].lower() if filename else ""
     if ext not in (".mp3", ".wav", ".ogg", ".m4a", ".aac", ".flac"):
         ext = ".mp3"
@@ -20,38 +50,140 @@ def _safe_audio_filename(filename: str, call_id: str) -> str:
 
 
 async def upload_audio(data: bytes, filename: str, call_id: str = "") -> Optional[str]:
-    """Upload audio bytes to GCS and return a signed URL, or None if disabled."""
+    """Upload audio bytes to GCS and return the canonical gs:// URI, or None if disabled."""
     if not settings.gcs_bucket:
         logger.info("GCS_BUCKET not configured — skipping audio upload.")
         return None
 
-    def _upload() -> str:
+    safe_name = _safe_audio_filename(filename, call_id)
+    blob_path = f"calls/{safe_name}"
+
+    def _upload() -> None:
         from google.cloud import storage
-        from google.oauth2 import service_account as sa
         if settings.gcp_credentials_path:
             client = storage.Client.from_service_account_json(settings.gcp_credentials_path)
-            signing_creds = sa.Credentials.from_service_account_file(settings.gcp_credentials_path)
         else:
             client = storage.Client()
-            signing_creds = None
-        bucket = client.bucket(settings.gcs_bucket)
-        safe_name = _safe_audio_filename(filename, call_id)
-        blob = bucket.blob(f"calls/{safe_name}")
+        blob = client.bucket(settings.gcs_bucket).blob(blob_path)
         blob.upload_from_string(data, content_type="audio/mpeg")
-        if signing_creds:
-            return blob.generate_signed_url(
-                version="v2",
-                expiration=datetime.timedelta(days=365),
-                method="GET",
-                credentials=signing_creds,
-            )
-        # Fallback: return the gs:// URI (no public access)
-        return f"gs://{settings.gcs_bucket}/calls/{filename}"
 
     try:
-        url = await asyncio.to_thread(_upload)
-        logger.info(f"Audio uploaded: {url}")
-        return url
+        await asyncio.to_thread(_upload)
     except Exception as e:
         logger.error(f"GCS upload failed: {e}")
         return None
+
+    gcs_uri = f"gs://{settings.gcs_bucket}/{blob_path}"
+    logger.info(f"Audio uploaded: {gcs_uri}")
+    return gcs_uri
+
+
+async def download_audio(gcs_uri: str) -> Optional[bytes]:
+    """Fetch an object back out of GCS. Server-side read — no signing involved."""
+    bucket_name, blob_path = split_gcs_uri(gcs_uri)
+    if not bucket_name:
+        return None
+
+    def _download() -> bytes:
+        from google.cloud import storage
+        if settings.gcp_credentials_path:
+            client = storage.Client.from_service_account_json(settings.gcp_credentials_path)
+        else:
+            client = storage.Client()
+        return client.bucket(bucket_name).blob(blob_path).download_as_bytes()
+
+    try:
+        return await asyncio.to_thread(_download)
+    except Exception as e:
+        logger.warning(f"GCS download failed for {gcs_uri}: {e}")
+        return None
+
+
+def split_gcs_uri(gcs_uri: str) -> Tuple[Optional[str], Optional[str]]:
+    """``gs://bucket/path/to.mp3`` → ``("bucket", "path/to.mp3")``."""
+    if not gcs_uri or not gcs_uri.startswith("gs://"):
+        return None, None
+    without_scheme = gcs_uri[len("gs://"):]
+    if "/" not in without_scheme:
+        return None, None
+    bucket_name, blob_path = without_scheme.split("/", 1)
+    return bucket_name, blob_path
+
+
+def gcs_uri_for_call(call: dict) -> Optional[str]:
+    """Resolve the audio object for a call document.
+
+    Prefers the canonical ``audio_gcs_uri`` written by /upload. Falls back to
+    reconstructing the object name from the call_id for documents written
+    before this module was fixed: those stored a gs:// URI built from the
+    *client-supplied* filename, which never matched the object actually
+    written (always ``calls/{call_id}.mp3``). Reconstructing rather than
+    trusting the stored value is what makes every pre-existing recording
+    playable again without a data migration.
+    """
+    uri = call.get("audio_gcs_uri")
+    if uri:
+        return uri
+    call_id = call.get("call_id")
+    if call.get("audio_url") and call_id and settings.gcs_bucket:
+        return f"gs://{settings.gcs_bucket}/calls/{call_id}.mp3"
+    return None
+
+
+def _link_key() -> Optional[bytes]:
+    if not settings.service_key:
+        return None
+    return hmac.new(settings.service_key.encode("utf-8"), _KEY_CONTEXT, hashlib.sha256).digest()
+
+
+def sign_audio_link(call_id: str, expires_at: int) -> Optional[str]:
+    key = _link_key()
+    if not key:
+        return None
+    msg = f"{call_id}:{expires_at}".encode("utf-8")
+    return hmac.new(key, msg, hashlib.sha256).hexdigest()
+
+
+def verify_audio_link(call_id: str, expires_at: int, signature: str) -> bool:
+    if expires_at < int(time.time()):
+        return False
+    expected = sign_audio_link(call_id, expires_at)
+    if not expected:
+        return False
+    return hmac.compare_digest(expected, signature)
+
+
+_warned_no_service_key = False
+_warned_no_public_url = False
+
+
+def playback_url(call: dict) -> Optional[str]:
+    """Mint a short-lived playback URL for a call, or None if it has no audio."""
+    global _warned_no_service_key, _warned_no_public_url
+
+    call_id = call.get("call_id")
+    if not call_id or not gcs_uri_for_call(call):
+        return None
+
+    expires_at = int(time.time()) + settings.audio_link_ttl_seconds
+    signature = sign_audio_link(call_id, expires_at)
+    if not signature:
+        if not _warned_no_service_key:
+            logger.error("SERVICE_KEY not set — call audio cannot be served.")
+            _warned_no_service_key = True
+        return None
+
+    # Loud rather than silent: a relative link here would 404 against the
+    # frontend origin, which is the exact failure mode this module exists to
+    # stop repeating. Deploy via ansible so c2-core.env.j2 sets PUBLIC_API_URL.
+    if not settings.public_api_url and not _warned_no_public_url:
+        logger.error("PUBLIC_API_URL not set — call audio links will be relative and will not resolve.")
+        _warned_no_public_url = True
+
+    base = (settings.public_api_url or "").rstrip("/")
+    return f"{base}/media/calls/{call_id}/audio?exp={expires_at}&sig={signature}"
+
+
+def with_playback_url(call: dict) -> dict:
+    """Return the call dict with a freshly minted ``audio_url``."""
+    return {**call, "audio_url": playback_url(call)}