webmail-client/backend/app/api/avatar.py

"""Sender avatars — Gravatar then the sender domain's favicon (SPEC §9.1).

The browser only ever calls ``GET /api/avatar`` (same origin), so a sender can never
see the user's IP or read activity: the server does the outbound fetch under the
§10.6 SSRF guard and caches the result (both hits and misses) in Redis. A miss
returns 204 No Content: the <img> fails to decode it and the UI falls back to its
initials, without the browser logging an error for every sender who has no avatar
(a 404 did, filling the console). Disable with ``SENDER_AVATARS_ENABLED=false``.
"""

from __future__ import annotations

import hashlib

import httpx
from fastapi import APIRouter, Query, Response

from app.accounts.ssrf import SsrfError, guarded_request
from app.config import get_settings
from app.identity.deps import CurrentUser
from app.redis_client import get_redis_bytes

router = APIRouter(prefix="/avatar", tags=["mail"])

_CACHE_TTL_HIT = 7 * 24 * 3600  # remember a found avatar for a week
_CACHE_TTL_MISS = 24 * 3600  # re-check "no avatar" daily
_MAX_BYTES = 200 * 1024
_NONE = b"none"
# The browser keeps an answer as long as we would, so it stops re-asking per page.
_BROWSER_CACHE = {"Cache-Control": "private, max-age=86400"}


def _no_avatar() -> Response:
    return Response(status_code=204, headers=_BROWSER_CACHE)


def _email_md5(email: str) -> str:
    # Gravatar keys on md5(lowercased email); reused as our cache key. Not security.
    return hashlib.md5(email.encode()).hexdigest()  # noqa: S324


async def _fetch_image(url: str) -> tuple[str, bytes] | None:
    """GET ``url`` under the SSRF guard; return (content_type, bytes) for a 200 image."""
    try:
        resp = await guarded_request("GET", url, timeout=5.0, max_bytes=_MAX_BYTES)
    except (SsrfError, httpx.HTTPError, OSError):
        return None
    if resp.status != 200 or not resp.content:
        return None
    ctype = resp.headers.get("content-type", "").split(";")[0].strip().lower()
    if not ctype.startswith("image/"):
        return None
    return ctype, resp.content


async def _resolve(email: str, md5: str) -> tuple[str, bytes] | None:
    """Gravatar first (``d=404`` so it 404s when the user has none), then the sender
    domain's ``/favicon.ico`` as a brand-ish fallback."""
    gravatar = await _fetch_image(f"https://www.gravatar.com/avatar/{md5}?d=404&s=80")
    if gravatar is not None:
        return gravatar
    domain = email.rsplit("@", 1)[-1]
    if domain:
        favicon = await _fetch_image(f"https://{domain}/favicon.ico")
        if favicon is not None:
            return favicon
    return None


@router.get("")
async def sender_avatar(_user: CurrentUser, email: str = Query(...)) -> Response:
    email = email.strip().lower()
    if not get_settings().sender_avatars_enabled or "@" not in email:
        return _no_avatar()

    # Binary-safe client: the cache holds raw image bytes (and a b"none" marker).
    redis = get_redis_bytes()
    md5 = _email_md5(email)
    key = f"avatar:{md5}"
    cached: bytes | None = await redis.get(key)
    if cached == _NONE:
        return _no_avatar()
    if cached:
        cached_ctype, _, cached_body = cached.partition(b"\n")
        return Response(
            content=cached_body,
            media_type=cached_ctype.decode("ascii", "ignore"),
            headers=_BROWSER_CACHE,
        )

    result = await _resolve(email, md5)
    if result is None:
        await redis.set(key, _NONE, ex=_CACHE_TTL_MISS)
        return _no_avatar()
    ctype, body = result
    await redis.set(key, ctype.encode("ascii", "ignore") + b"\n" + body, ex=_CACHE_TTL_HIT)
    return Response(content=body, media_type=ctype, headers=_BROWSER_CACHE)