FabledCurator/backend/app/api/extension.py

"""FC-3g: /api/extension — quick-add-source for the Firefox extension
+ install-time manifest for the Settings card.
"""

from __future__ import annotations

import asyncio
import hashlib
import hmac
import re
from pathlib import Path

from quart import Blueprint, jsonify, request
from sqlalchemy import select

from ..extensions import get_session
from ..models import AppSetting
from ..services.extension_service import (
    ExtensionService,
    InvalidUrlError,
    UnknownPlatformError,
)
from ..services.source_service import KNOWN_PLATFORMS
from ._responses import error_response as _bad

extension_bp = Blueprint("extension", __name__, url_prefix="/api/extension")

# Default XPI directory; tests override via monkeypatching this module-
# level constant.
XPI_DIR = Path("/app/frontend/dist/extension")

_XPI_VERSION_RE = re.compile(r"fabledcurator-(?P<version>[\w.-]+)\.xpi$")


async def _ext_key_required(session) -> bool:
    """Unlike /api/credentials (which accepts the browser path with no
    header), quick-add-source writes server state and must be explicitly
    authenticated."""
    supplied = request.headers.get("X-Extension-Key")
    if supplied is None:
        return False
    stored = (await session.execute(
        select(AppSetting.value).where(AppSetting.key == "extension_api_key")
    )).scalar_one_or_none()
    if stored is None:
        return False
    # compare_digest, not `==`: the stored key is a shared secret, and a
    # short-circuiting compare leaks its prefix through timing. Costs nothing
    # here — it is not that this route is exposed (#3072). Compared as BYTES:
    # compare_digest's str form rejects non-ASCII with TypeError, and this
    # header is attacker-supplied, so a str compare would turn a junk key into
    # a 500 instead of a 403.
    return hmac.compare_digest(supplied.encode("utf-8"), stored.encode("utf-8"))


def _extract_version(xpi_name: str) -> str:
    m = _XPI_VERSION_RE.search(xpi_name)
    return m.group("version") if m else "unknown"


def _sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fp:
        for chunk in iter(lambda: fp.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


@extension_bp.route("/probe", methods=["GET"])
async def probe_source():
    """Read-only resolution of a creator-page URL: tells the extension
    whether this URL is already a Source, is for an Artist that exists
    but with a different URL, is brand new, or doesn't match any known
    platform pattern. Drives the content-script chip's color/copy
    BEFORE the operator clicks, so the button can show 'already added'
    without requiring an add-attempt."""
    url = (request.args.get("url") or "").strip()
    if not url:
        return _bad("invalid_body", detail="url query parameter is required")
    async with get_session() as session:
        if not await _ext_key_required(session):
            return _bad("unauthorized", status=401)
        result = await ExtensionService(session).probe(url)
    return jsonify(result)


@extension_bp.route("/quick-add-source", methods=["POST"])
async def quick_add_source():
    body = await request.get_json(silent=True)
    if not isinstance(body, dict):
        return _bad("invalid_body", detail="body must be a JSON object")
    url = body.get("url")
    if not isinstance(url, str) or not url.strip():
        return _bad("invalid_body", detail="url is required")

    from .credentials import _get_crypto

    async with get_session() as session:
        if not await _ext_key_required(session):
            return _bad("unauthorized", status=401)
        try:
            # crypto lets a pixiv add resolve the artist's display name via the
            # stored OAuth token (else it falls back to the numeric id). #130.
            result = await ExtensionService(session, _get_crypto()).quick_add_source(url)
        except UnknownPlatformError as exc:
            return _bad(
                "unknown_platform",
                detail=str(exc),
                known=sorted(KNOWN_PLATFORMS),
            )
        except InvalidUrlError as exc:
            return _bad("invalid_url", detail=str(exc))
    return jsonify(result), (201 if result["created_source"] else 200)


def _read_manifest_sync() -> dict | None:
    """All the filesystem-touching work for /api/extension/manifest,
    in a sync helper so the async route can dispatch it via
    asyncio.to_thread (ASYNC240: no pathlib I/O in async functions)."""
    if not XPI_DIR.is_dir():
        return None
    # Exclude the `fabledcurator-latest.xpi` alias when picking the file to
    # extract a version from — it's a copy of the latest versioned XPI,
    # written at the same mtime by build.yml, and would otherwise tie or
    # win the sort (operator-flagged 2026-05-26: UI displayed "v latest"
    # because `_extract_version("fabledcurator-latest.xpi")` returns
    # the literal "latest"). The alias still serves as `latest_url`.
    versioned = [
        p for p in XPI_DIR.glob("fabledcurator-*.xpi")
        if p.name != "fabledcurator-latest.xpi"
    ]
    if not versioned:
        return None
    versioned.sort(key=lambda p: p.stat().st_mtime)
    latest = versioned[-1]
    return {
        "installed": True,
        "version": _extract_version(latest.name),
        "xpi_url": f"/extension/{latest.name}",
        "latest_url": "/extension/fabledcurator-latest.xpi",
        "sha256": _sha256(latest),
    }


@extension_bp.route("/manifest", methods=["GET"])
async def extension_manifest():
    info = await asyncio.to_thread(_read_manifest_sync)
    if info is None:
        return jsonify({"installed": False}), 404
    return jsonify(info)