127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""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
|