0.25.0 — Jetons d'API personnels (kbt_) pour les surfaces machine
CI / test (push) Successful in 1m56s
CI / test (push) Successful in 1m56s
Depuis 0.24.0, /mcp exige session ou Basic : le service d'agents s'y faisait refuser en 401 (vu en production, panneau Spark du 2026-10-21). - Table api_tokens (empreinte SHA-256 seule), CRUD dans Mon profil - Bearer honoré par le middleware sur /mcp et /api/ uniquement — jamais l'interface web ; révocation et désactivation immédiates - 235 tests verts (3 nouveaux : cycle de vie, compte désactivé, handshake MCP par jeton quand l'auth est active)
This commit is contained in:
+73
-5
@@ -308,6 +308,18 @@ def _basic_ok(request: Request) -> bool:
|
||||
) and secrets.compare_digest(password, settings.auth_password)
|
||||
|
||||
|
||||
def _bearer_user(request: Request) -> dict | None:
|
||||
"""Jeton d'API personnel (0.25.0) : Authorization: Bearer kbt_…
|
||||
|
||||
Réservé aux surfaces machine (/mcp, /api/) — le middleware ne
|
||||
l'honore que là. La résolution vérifie l'empreinte, la révocation
|
||||
et que le compte est actif, et journalise l'usage (last_used_at)."""
|
||||
header = request.headers.get("authorization", "")
|
||||
if not header.lower().startswith("bearer "):
|
||||
return None
|
||||
return models.user_by_api_token(header[7:].strip())
|
||||
|
||||
|
||||
def _current_user(request: Request) -> dict | None:
|
||||
"""The database user behind the session, or None.
|
||||
|
||||
@@ -367,6 +379,11 @@ class AuthMiddleware(BaseHTTPMiddleware):
|
||||
session_user = None # disabled account: session is dead
|
||||
if session_user is not None or _basic_ok(request):
|
||||
return await call_next(request)
|
||||
# Jetons d'API (0.25.0) : surfaces machine seulement — un jeton
|
||||
# ne déverrouille jamais l'interface web.
|
||||
if (path.startswith("/mcp") or path.startswith("/api/")) \
|
||||
and _bearer_user(request) is not None:
|
||||
return await call_next(request)
|
||||
# API consumers get a plain 401 (no browser prompt, no
|
||||
# redirect they cannot follow); browsers get the login page.
|
||||
if path.startswith("/api/") or "text/html" not in request.headers.get(
|
||||
@@ -1541,11 +1558,8 @@ async def admin_settings_save(request: Request, csrf_token: str = Form("")):
|
||||
)
|
||||
|
||||
|
||||
@app.get("/profile")
|
||||
async def profile_page(request: Request, msg: str = "", err: str = ""):
|
||||
user = _current_user(request)
|
||||
if user is None:
|
||||
return RedirectResponse("/login", status_code=303)
|
||||
def _render_profile(request: Request, user: dict, msg: str = "", err: str = "",
|
||||
nouveau_jeton: str = ""):
|
||||
from app import forge_crypto
|
||||
|
||||
mine = []
|
||||
@@ -1566,10 +1580,64 @@ async def profile_page(request: Request, msg: str = "", err: str = ""):
|
||||
else []
|
||||
),
|
||||
my_connections=mine,
|
||||
api_tokens=(
|
||||
models.list_api_tokens(user["id"])
|
||||
if user.get("id") is not None
|
||||
else []
|
||||
),
|
||||
nouveau_jeton=nouveau_jeton,
|
||||
msg=msg, err=err,
|
||||
)
|
||||
|
||||
|
||||
@app.get("/profile")
|
||||
async def profile_page(request: Request, msg: str = "", err: str = ""):
|
||||
user = _current_user(request)
|
||||
if user is None:
|
||||
return RedirectResponse("/login", status_code=303)
|
||||
return _render_profile(request, user, msg=msg, err=err)
|
||||
|
||||
|
||||
@app.post("/profile/tokens")
|
||||
async def profile_token_create(
|
||||
request: Request,
|
||||
name: str = Form(""),
|
||||
csrf_token: str = Form(""),
|
||||
):
|
||||
user = _current_user(request)
|
||||
if user is None or user.get("id") is None:
|
||||
return RedirectResponse("/profile", status_code=303)
|
||||
if not _csrf_valid(request, csrf_token):
|
||||
return _csrf_error()
|
||||
nom = name.strip() or "Sans nom"
|
||||
_row, plaintext = models.create_api_token(user["id"], nom)
|
||||
models.log_admin(user["id"], "profile.token.create", nom)
|
||||
# Le texte clair n'est montré qu'ici, une seule fois : la base ne
|
||||
# conserve que l'empreinte.
|
||||
return _render_profile(
|
||||
request, user,
|
||||
msg="Jeton créé — copie-le maintenant, il ne sera plus jamais affiché.",
|
||||
nouveau_jeton=plaintext,
|
||||
)
|
||||
|
||||
|
||||
@app.post("/profile/tokens/{token_id}/revoke")
|
||||
async def profile_token_revoke(
|
||||
request: Request,
|
||||
token_id: int,
|
||||
csrf_token: str = Form(""),
|
||||
):
|
||||
user = _current_user(request)
|
||||
if user is None or user.get("id") is None:
|
||||
return RedirectResponse("/profile", status_code=303)
|
||||
if not _csrf_valid(request, csrf_token):
|
||||
return _csrf_error()
|
||||
if models.revoke_api_token(user["id"], token_id):
|
||||
models.log_admin(user["id"], "profile.token.revoke", str(token_id))
|
||||
return RedirectResponse("/profile?msg=Jeton+r%C3%A9voqu%C3%A9", status_code=303)
|
||||
return RedirectResponse("/profile?err=Jeton+introuvable", status_code=303)
|
||||
|
||||
|
||||
@app.post("/profile")
|
||||
async def profile_update(
|
||||
request: Request,
|
||||
|
||||
Reference in New Issue
Block a user