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

"""
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 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.

    We ignore the client-supplied filename entirely and derive the name from the
    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.
    """
    ext = os.path.splitext(filename)[-1].lower() if filename else ""
    if ext not in (".mp3", ".wav", ".ogg", ".m4a", ".aac", ".flac"):
        ext = ".mp3"
    return f"{call_id}{ext}"


async def upload_audio(data: bytes, filename: str, call_id: str = "") -> Optional[str]:
    """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

    safe_name = _safe_audio_filename(filename, call_id)
    blob_path = f"calls/{safe_name}"

    def _upload() -> None:
        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()
        blob = client.bucket(settings.gcs_bucket).blob(blob_path)
        blob.upload_from_string(data, content_type="audio/mpeg")

    try:
        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)}