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 AUDIO_CONTENT_TYPES:
        ext = ".mp3"
    return f"{call_id}{ext}"


# Extension → Content-Type. The node used to send nothing but 16 kbps MP3, so
# "audio/mpeg" was hardcoded at every point audio is written or served; it now
# sends FLAC (lossless, for Whisper's benefit — see call_recorder.py's AUDIO_*
# constants) and a stored object mislabelled audio/mpeg will not play in a
# browser. Old .mp3 objects keep working: the map is keyed off the real
# extension, not off what the current node happens to produce.
AUDIO_CONTENT_TYPES = {
    ".flac": "audio/flac",
    ".mp3":  "audio/mpeg",
    ".wav":  "audio/wav",
    ".ogg":  "audio/ogg",
    ".m4a":  "audio/mp4",
    ".aac":  "audio/aac",
}


def content_type_for(name: str) -> str:
    """Content-Type for a stored audio object, by extension. Defaults to MP3."""
    ext = os.path.splitext(name or "")[-1].lower()
    return AUDIO_CONTENT_TYPES.get(ext, "audio/mpeg")


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=content_type_for(safe_name))

    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


# ── Ranged reads for /media playback (server-26#233) ────────────────────────
# A 600 s FLAC call is ~13 MB. /media used to pull the whole object into
# memory (the old download_audio(), removed once nothing called it — #225 S8),
# so the browser waited for all of it before the first sample played. These
# two read only what a request asks for: the
# object size (one metadata GET) and then an inclusive byte span, which the
# router streams out in chunks. One cached client, because the router issues
# several span reads per playback and a fresh Client() per read re-resolves
# credentials every time.
_gcs_client_cache = None


def _gcs_client():
    global _gcs_client_cache
    if _gcs_client_cache is None:
        from google.cloud import storage
        if settings.gcp_credentials_path:
            _gcs_client_cache = storage.Client.from_service_account_json(settings.gcp_credentials_path)
        else:
            _gcs_client_cache = storage.Client()
    return _gcs_client_cache


async def audio_object_size(gcs_uri: str) -> Optional[int]:
    """Size in bytes of a stored audio object, or None if it is missing/unreadable."""
    bucket_name, blob_path = split_gcs_uri(gcs_uri)
    if not bucket_name:
        return None

    def _size() -> Optional[int]:
        blob = _gcs_client().bucket(bucket_name).blob(blob_path)
        blob.reload()
        return blob.size

    try:
        size = await asyncio.to_thread(_size)
    except Exception as e:
        logger.warning(f"GCS metadata read failed for {gcs_uri}: {e}")
        return None
    return int(size) if size is not None else None


async def download_audio_range(gcs_uri: str, start: int, end: int) -> Optional[bytes]:
    """Bytes ``start``..``end`` INCLUSIVE of a stored object (GCS's own convention).

    Reads only that span from GCS — never the whole object.
    """
    bucket_name, blob_path = split_gcs_uri(gcs_uri)
    if not bucket_name:
        return None

    def _read() -> bytes:
        blob = _gcs_client().bucket(bucket_name).blob(blob_path)
        return blob.download_as_bytes(start=start, end=end)

    try:
        return await asyncio.to_thread(_read)
    except Exception as e:
        logger.warning(f"GCS ranged read failed for {gcs_uri} [{start}-{end}]: {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)}