feat(fc3b): /api/credentials blueprint (CRUD + X-Extension-Key auth)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/__init__.py

@@ -19,6 +19,7 @@ def all_blueprints() -> list[Blueprint]:
     from .artist import artist_bp
     from .artists import artists_bp
     from .attachments import attachments_bp
+    from .credentials import credentials_bp
     from .gallery import gallery_bp
     from .import_admin import import_admin_bp
     from .ml_admin import ml_admin_bp
@@ -46,4 +47,5 @@ def all_blueprints() -> list[Blueprint]:
         ml_admin_bp,
         sources_bp,
         platforms_bp,
+        credentials_bp,
     ]

backend/app/api/credentials.py

@@ -0,0 +1,126 @@
+"""FC-3b: /api/credentials — CRUD with X-Extension-Key auth.
+
+Browser requests (no header) are accepted per the homelab posture —
+there is no user-session model in FC. The X-Extension-Key check
+exists to give the extension a revocable path: a request that
+supplies the header must match `app_setting.extension_api_key`.
+"""
+
+from pathlib import Path
+
+from quart import Blueprint, jsonify, request
+from sqlalchemy import select
+
+from ..extensions import get_session
+from ..models import AppSetting, Credential
+from ..services.credential_crypto import CredentialCrypto
+from ..services.credential_service import (
+    CredentialService,
+    EmptyDataError,
+    UnknownPlatformError,
+    WrongAuthTypeError,
+)
+
+credentials_bp = Blueprint("credentials", __name__, url_prefix="/api/credentials")
+
+
+# Production key path; tests pass `cookies_dir=tmp_path` via the
+# service constructor (the API uses the default). The Fernet key
+# file is also at `/images/secrets/credential_key.b64`.
+_KEY_PATH = Path("/images/secrets/credential_key.b64")
+_crypto: CredentialCrypto | None = None
+
+
+def _get_crypto() -> CredentialCrypto:
+    global _crypto
+    if _crypto is None:
+        _crypto = CredentialCrypto(_KEY_PATH)
+    return _crypto
+
+
+def _bad(error: str, *, status: int = 400, detail: str | None = None, **extra):
+    body = {"error": error}
+    if detail is not None:
+        body["detail"] = detail
+    body.update(extra)
+    return jsonify(body), status
+
+
+async def _ext_key_ok(session) -> bool:
+    """If X-Extension-Key is supplied, it must match the stored value.
+    Missing header → True (browser path; accepted per homelab posture).
+    """
+    supplied = request.headers.get("X-Extension-Key")
+    if supplied is None:
+        return True
+    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
+
+
+@credentials_bp.route("", methods=["GET"])
+async def list_credentials():
+    async with get_session() as session:
+        if not await _ext_key_ok(session):
+            return _bad("unauthorized", status=401)
+        records = await CredentialService(session, _get_crypto()).list()
+    return jsonify([r.to_dict() for r in records])
+
+
+@credentials_bp.route("/<platform>", methods=["GET"])
+async def get_credential(platform: str):
+    async with get_session() as session:
+        if not await _ext_key_ok(session):
+            return _bad("unauthorized", status=401)
+        record = await CredentialService(session, _get_crypto()).get(platform)
+    if record is None:
+        return _bad("not_found", status=404)
+    return jsonify(record.to_dict())
+
+
+@credentials_bp.route("", methods=["POST"])
+async def upsert_credential():
+    body = await request.get_json()
+    if not isinstance(body, dict):
+        return _bad("invalid_body", detail="body must be a JSON object")
+    try:
+        platform = body["platform"]
+        credential_type = body["credential_type"]
+        data = body["data"]
+    except KeyError:
+        return _bad("invalid_body", detail="platform, credential_type, data are required")
+    if not isinstance(data, str):
+        return _bad("invalid_body", detail="data must be a string")
+
+    async with get_session() as session:
+        if not await _ext_key_ok(session):
+            return _bad("unauthorized", status=401)
+        existed = (await session.execute(
+            select(Credential).where(Credential.platform == platform)
+        )).scalar_one_or_none() is not None
+        svc = CredentialService(session, _get_crypto())
+        try:
+            record = await svc.upsert(
+                platform=platform, credential_type=credential_type, data=data,
+            )
+        except UnknownPlatformError as exc:
+            return _bad("unknown_platform", detail=str(exc))
+        except WrongAuthTypeError as exc:
+            return _bad("wrong_auth_type", detail=str(exc), expected=exc.expected)
+        except EmptyDataError as exc:
+            return _bad("empty_data", detail=str(exc))
+    return jsonify(record.to_dict()), (200 if existed else 201)
+
+
+@credentials_bp.route("/<platform>", methods=["DELETE"])
+async def delete_credential(platform: str):
+    async with get_session() as session:
+        if not await _ext_key_ok(session):
+            return _bad("unauthorized", status=401)
+        svc = CredentialService(session, _get_crypto())
+        try:
+            await svc.delete(platform)
+        except LookupError:
+            return _bad("not_found", status=404)
+    return "", 204

tests/test_api_credentials.py

@@ -0,0 +1,145 @@
+import pytest
+
+from backend.app import create_app
+from backend.app.models import AppSetting
+
+pytestmark = pytest.mark.integration
+
+
+_NETSCAPE = (
+    "# Netscape HTTP Cookie File\n"
+    ".patreon.com\tTRUE\t/\tTRUE\t1700000000\tsession_id\tabc\n"
+)
+
+
+@pytest.fixture
+async def app():
+    return create_app()
+
+
+@pytest.fixture
+async def client(app):
+    async with app.test_client() as c:
+        yield c
+
+
+@pytest.fixture
+async def ext_key(db):
+    # Seed an extension API key for tests that want to assert the
+    # X-Extension-Key path explicitly.
+    db.add(AppSetting(key="extension_api_key", value="test-ext-key"))
+    await db.commit()
+    return "test-ext-key"
+
+
+@pytest.mark.asyncio
+async def test_list_empty(client):
+    resp = await client.get("/api/credentials")
+    assert resp.status_code == 200
+    assert await resp.get_json() == []
+
+
+@pytest.mark.asyncio
+async def test_create_and_get(client):
+    resp = await client.post("/api/credentials", json={
+        "platform": "patreon",
+        "credential_type": "cookies",
+        "data": _NETSCAPE,
+    })
+    assert resp.status_code == 201
+    body = await resp.get_json()
+    assert body["platform"] == "patreon"
+    assert body["credential_type"] == "cookies"
+    assert "data" not in body  # never echoed
+    assert "encrypted_blob" not in body
+
+    one = await client.get("/api/credentials/patreon")
+    assert one.status_code == 200
+    assert (await one.get_json())["platform"] == "patreon"
+
+
+@pytest.mark.asyncio
+async def test_create_updates_existing(client):
+    a = await client.post("/api/credentials", json={
+        "platform": "patreon", "credential_type": "cookies", "data": _NETSCAPE,
+    })
+    assert a.status_code == 201
+    b = await client.post("/api/credentials", json={
+        "platform": "patreon", "credential_type": "cookies", "data": _NETSCAPE + "x",
+    })
+    # On update, our convention is 200 not 201
+    assert b.status_code == 200
+    listing = await (await client.get("/api/credentials")).get_json()
+    assert len(listing) == 1
+
+
+@pytest.mark.asyncio
+async def test_unknown_platform_400(client):
+    resp = await client.post("/api/credentials", json={
+        "platform": "fanbox", "credential_type": "cookies", "data": "x",
+    })
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "unknown_platform"
+
+
+@pytest.mark.asyncio
+async def test_wrong_auth_type_400(client):
+    resp = await client.post("/api/credentials", json={
+        "platform": "discord", "credential_type": "cookies", "data": "x",
+    })
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "wrong_auth_type"
+    assert body["expected"] == "token"
+
+
+@pytest.mark.asyncio
+async def test_empty_data_400(client):
+    resp = await client.post("/api/credentials", json={
+        "platform": "patreon", "credential_type": "cookies", "data": "   ",
+    })
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "empty_data"
+
+
+@pytest.mark.asyncio
+async def test_invalid_body_400(client):
+    resp = await client.post("/api/credentials", json=[1, 2, 3])
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_get_404(client):
+    resp = await client.get("/api/credentials/patreon")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_delete_204_then_404(client):
+    await client.post("/api/credentials", json={
+        "platform": "patreon", "credential_type": "cookies", "data": _NETSCAPE,
+    })
+    dele = await client.delete("/api/credentials/patreon")
+    assert dele.status_code == 204
+    nope = await client.delete("/api/credentials/patreon")
+    assert nope.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_extension_key_correct_accepts(client, ext_key):
+    resp = await client.post(
+        "/api/credentials",
+        json={"platform": "patreon", "credential_type": "cookies", "data": _NETSCAPE},
+        headers={"X-Extension-Key": ext_key},
+    )
+    assert resp.status_code == 201
+
+
+@pytest.mark.asyncio
+async def test_extension_key_wrong_rejects(client, ext_key):
+    resp = await client.post(
+        "/api/credentials",
+        json={"platform": "patreon", "credential_type": "cookies", "data": _NETSCAPE},
+        headers={"X-Extension-Key": "WRONG"},
+    )
+    assert resp.status_code == 401