125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""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 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
|
|
|
|
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$")
|
|
|
|
|
|
def _bad(error: str, *, status: int = 400, **extra):
|
|
body = {"error": error}
|
|
body.update(extra)
|
|
return jsonify(body), status
|
|
|
|
|
|
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()
|
|
return stored is not None and supplied == stored
|
|
|
|
|
|
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("/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")
|
|
|
|
async with get_session() as session:
|
|
if not await _ext_key_required(session):
|
|
return _bad("unauthorized", status=401)
|
|
try:
|
|
result = await ExtensionService(session).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)
|