fc3g: /api/extension blueprint — quick-add-source + manifest endpoints, registered

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-23 23:38:35 -04:00
parent 46d199450d
commit 2065672a31
2 changed files with 105 additions and 0 deletions
+2
View File
@@ -21,6 +21,7 @@ def all_blueprints() -> list[Blueprint]:
from .attachments import attachments_bp
from .credentials import credentials_bp
from .downloads import downloads_bp
from .extension import extension_bp
from .gallery import gallery_bp
from .import_admin import import_admin_bp
from .migrate import migrate_bp
@@ -53,5 +54,6 @@ def all_blueprints() -> list[Blueprint]:
platforms_bp,
posts_bp,
credentials_bp,
extension_bp,
downloads_bp,
]
+103
View File
@@ -0,0 +1,103 @@
"""FC-3g: /api/extension — quick-add-source for the Firefox extension
+ install-time manifest for the Settings card.
"""
from __future__ import annotations
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)
@extension_bp.route("/manifest", methods=["GET"])
async def extension_manifest():
if not XPI_DIR.is_dir():
return jsonify({"installed": False}), 404
xpis = sorted(XPI_DIR.glob("fabledcurator-*.xpi"), key=lambda p: p.stat().st_mtime)
if not xpis:
return jsonify({"installed": False}), 404
latest = xpis[-1]
return jsonify({
"installed": True,
"version": _extract_version(latest.name),
"xpi_url": f"/extension/{latest.name}",
"latest_url": "/extension/fabledcurator-latest.xpi",
"sha256": _sha256(latest),
})