feat(fc3b): extension_api_key endpoints + create_app bootstraps credential key

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-20 18:36:28 -04:00
parent 617aa5a90a
commit e77d976ff0
3 changed files with 103 additions and 1 deletions
+9
View File
@@ -1,18 +1,27 @@
"""Quart app factory."""
import logging
from pathlib import Path
from quart import Quart
from .api import all_blueprints
from .config import get_config
from .frontend import frontend_bp
from .services.credential_crypto import CredentialCrypto
_CREDENTIAL_KEY_PATH = Path("/images/secrets/credential_key.b64")
def create_app() -> Quart:
cfg = get_config()
logging.basicConfig(level=cfg.log_level)
# Bootstrap the credential encryption key file (mode 0600) before
# any request can land. FC-3b: file-based system-generated key.
CredentialCrypto(_CREDENTIAL_KEY_PATH)
app = Quart(__name__)
app.secret_key = cfg.secret_key
+41 -1
View File
@@ -1,10 +1,12 @@
"""Settings API: import filters, system stats."""
import secrets
from quart import Blueprint, jsonify, request
from sqlalchemy import func, select
from ..extensions import get_session
from ..models import ImageRecord, ImportBatch, ImportSettings, ImportTask, Tag
from ..models import AppSetting, ImageRecord, ImportBatch, ImportSettings, ImportTask, Tag
settings_bp = Blueprint("settings", __name__, url_prefix="/api")
@@ -137,3 +139,41 @@ async def system_stats():
},
"active_batch": active_batch,
})
# --- FC-3b: extension API key (lives in app_setting) -----------------------
async def _get_or_seed_extension_api_key(session) -> str:
row = (await session.execute(
select(AppSetting).where(AppSetting.key == "extension_api_key")
)).scalar_one_or_none()
if row is None:
row = AppSetting(key="extension_api_key", value=secrets.token_urlsafe(32))
session.add(row)
await session.commit()
await session.refresh(row)
return row.value
@settings_bp.route("/settings/extension_api_key", methods=["GET"])
async def get_extension_api_key():
async with get_session() as session:
key = await _get_or_seed_extension_api_key(session)
return jsonify({"key": key})
@settings_bp.route("/settings/extension_api_key/rotate", methods=["POST"])
async def rotate_extension_api_key():
async with get_session() as session:
row = (await session.execute(
select(AppSetting).where(AppSetting.key == "extension_api_key")
)).scalar_one_or_none()
new_value = secrets.token_urlsafe(32)
if row is None:
row = AppSetting(key="extension_api_key", value=new_value)
session.add(row)
else:
row.value = new_value
await session.commit()
return jsonify({"key": new_value})
+53
View File
@@ -0,0 +1,53 @@
import pytest
from sqlalchemy import select
from backend.app import create_app
from backend.app.models import AppSetting
pytestmark = pytest.mark.integration
@pytest.fixture
async def app():
return create_app()
@pytest.fixture
async def client(app):
async with app.test_client() as c:
yield c
@pytest.mark.asyncio
async def test_get_seeds_a_value(client, db):
# First call seeds the row (idempotent on repeat calls).
resp = await client.get("/api/settings/extension_api_key")
assert resp.status_code == 200
body = await resp.get_json()
assert isinstance(body["key"], str) and len(body["key"]) >= 16
row = (await db.execute(
select(AppSetting.value).where(AppSetting.key == "extension_api_key")
)).scalar_one()
assert row == body["key"]
@pytest.mark.asyncio
async def test_get_is_idempotent(client):
a = (await (await client.get("/api/settings/extension_api_key")).get_json())["key"]
b = (await (await client.get("/api/settings/extension_api_key")).get_json())["key"]
assert a == b
@pytest.mark.asyncio
async def test_rotate_changes_the_value(client, db):
before = (await (await client.get("/api/settings/extension_api_key")).get_json())["key"]
rotated = await client.post("/api/settings/extension_api_key/rotate")
assert rotated.status_code == 200
after = (await rotated.get_json())["key"]
assert after != before
row = (await db.execute(
select(AppSetting.value).where(AppSetting.key == "extension_api_key")
)).scalar_one()
assert row == after