diff --git a/CHANGELOG.md b/CHANGELOG.md index a9b0a91..3fda827 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,25 @@ # Changelog +## 0.25.0 — 2026-10-21 + +### Ajouté — jetons d'API personnels pour les surfaces machine (MCP) + +Depuis la 0.24.0, tout Kanbante exige une authentification (session +ou Basic) — y compris `/mcp`. Le service d'agents s'y faisait refuser +en 401 (vu en production le 2026-10-21, panneau Spark : « Source +Kanbante injoignable »). Il manquait une identité machine propre. + +- Table `api_tokens` : jetons personnels préfixés `kbt_`, seule + l'empreinte SHA-256 est conservée ; création, liste et révocation + dans **Mon profil → Jetons d'API (MCP)**, texte clair affiché une + seule fois à la création. +- Le middleware honore `Authorization: Bearer kbt_…` **uniquement** + sur `/mcp` et `/api/` : un jeton ne déverrouille jamais l'interface + web. Révocation et désactivation du compte ont effet immédiat ; + `last_used_at` journalise l'usage. +- Usage visé : créer un jeton « service-agents » et le poser en + `AGENT_MCP_KANBANTE_KEY` dans le `.env` du conteneur agent. + ## 0.24.2 — 2026-10-21 ### Corrigé — boîtes d'édition blanches dans l'administration (#29) diff --git a/app/config.py b/app/config.py index 6406065..bf6e91f 100644 --- a/app/config.py +++ b/app/config.py @@ -104,7 +104,7 @@ from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent -APP_VERSION = "0.24.2" +APP_VERSION = "0.25.0" def _env_bool(name: str, default: bool) -> bool: diff --git a/app/main.py b/app/main.py index fae550d..116fc70 100644 --- a/app/main.py +++ b/app/main.py @@ -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, diff --git a/app/models.py b/app/models.py index 7dd3002..09169ee 100644 --- a/app/models.py +++ b/app/models.py @@ -11,8 +11,10 @@ and upgrades older databases in place (position column, notes updated_at, cleanup of rows carrying an unknown column name). """ +import hashlib import json import os +import secrets import sqlite3 from contextlib import contextmanager from typing import Iterator @@ -134,6 +136,16 @@ def init_db() -> None: last_login_at TEXT ); + CREATE TABLE IF NOT EXISTS api_tokens ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL REFERENCES users(id), + name TEXT NOT NULL DEFAULT '', + token_hash TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + last_used_at TEXT, + revoked_at TEXT + ); + CREATE TABLE IF NOT EXISTS workspaces ( id INTEGER PRIMARY KEY AUTOINCREMENT, slug TEXT NOT NULL UNIQUE, @@ -900,6 +912,73 @@ def get_user_by_username(username: str) -> dict | None: return dict(row) if row else None +# --- Jetons d'API (0.25.0) ------------------------------------------------- +# Jetons personnels pour les consommateurs machine (le service d'agents +# et son client MCP, en premier lieu). Le texte clair (préfixe kbt_) +# n'existe qu'à la création ; la base ne garde que l'empreinte SHA-256. +# Un jeton authentifie uniquement les surfaces machine (/mcp, /api/) — +# jamais l'interface web. + +API_TOKEN_PREFIX = "kbt_" + + +def _hash_api_token(plaintext: str) -> str: + return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() + + +def create_api_token(user_id: int, name: str) -> tuple[dict, str]: + plaintext = API_TOKEN_PREFIX + secrets.token_urlsafe(32) + with get_db() as conn: + cur = conn.execute( + "INSERT INTO api_tokens (user_id, name, token_hash) VALUES (?, ?, ?)", + (user_id, name.strip(), _hash_api_token(plaintext)), + ) + row = conn.execute( + "SELECT * FROM api_tokens WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return dict(row), plaintext + + +def list_api_tokens(user_id: int) -> list[dict]: + with get_db() as conn: + return [ + dict(r) + for r in conn.execute( + "SELECT * FROM api_tokens WHERE user_id = ? ORDER BY id", + (user_id,), + ).fetchall() + ] + + +def revoke_api_token(user_id: int, token_id: int) -> bool: + with get_db() as conn: + cur = conn.execute( + "UPDATE api_tokens SET revoked_at = datetime('now')" + " WHERE id = ? AND user_id = ? AND revoked_at IS NULL", + (token_id, user_id), + ) + return cur.rowcount > 0 + + +def user_by_api_token(plaintext: str) -> dict | None: + if not plaintext or not plaintext.startswith(API_TOKEN_PREFIX): + return None + with get_db() as conn: + row = conn.execute( + "SELECT u.* FROM api_tokens t JOIN users u ON u.id = t.user_id" + " WHERE t.token_hash = ? AND t.revoked_at IS NULL AND u.active = 1", + (_hash_api_token(plaintext),), + ).fetchone() + if row is None: + return None + conn.execute( + "UPDATE api_tokens SET last_used_at = datetime('now')" + " WHERE token_hash = ?", + (_hash_api_token(plaintext),), + ) + return dict(row) + + def get_user_by_draco(sub: str) -> dict | None: with get_db() as conn: row = conn.execute( diff --git a/app/templates/profile.html b/app/templates/profile.html index 2143f69..f4daad4 100644 --- a/app/templates/profile.html +++ b/app/templates/profile.html @@ -46,6 +46,43 @@ {% endif %} +
+

🗝️ Jetons d'API (MCP)

+

Pour les consommateurs machine — le service d'agents, en premier lieu. Un jeton (préfixe kbt_) authentifie uniquement /mcp et /api/, jamais l'interface web. Il se présente en en-tête Authorization: Bearer kbt_…. Le texte clair n'est affiché qu'à la création ; ensuite, seule son empreinte est conservée.

+ {% if profile_user.id %} + {% if nouveau_jeton %} +

✅ Nouveau jeton — copie-le maintenant :

+

{{ nouveau_jeton }}

+ {% endif %} + {% if api_tokens %} + + {% else %} +

Aucun jeton pour l'instant.

+ {% endif %} +
+ + + +
+ {% else %} +

Compte défini par l'environnement (KANBANTE_AUTH_*) : les jetons appartiennent aux comptes de la base — connecte-toi avec ton compte personnel pour en créer un.

+ {% endif %} +
+

✦ Liaison draco (SSO)

{% if profile_user.draco_sub %} diff --git a/pyproject.toml b/pyproject.toml index 889fd96..c09f34f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "kanbante" -version = "0.24.2" +version = "0.25.0" description = "Kanban project manager integrated with Gitea" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_api_tokens.py b/tests/test_api_tokens.py new file mode 100644 index 0000000..2986a5a --- /dev/null +++ b/tests/test_api_tokens.py @@ -0,0 +1,71 @@ +"""Jetons d'API personnels (0.25.0) : modèle + accès MCP par Bearer. + +Le service d'agents se fait refuser par Kanbante depuis la 0.24.0 +(auth partout, cookie/Basic seulement). Les jetons kbt_ lui donnent +une identité machine propre, révocable, limitée à /mcp et /api/. +""" +import pytest +from fastapi.testclient import TestClient + +import app.main as main +from app import accounts, models + + +@pytest.fixture() +def utilisateur(): + # La base de test est partagée par toute la session : nom unique, + # et nettoyage complet au teardown — le pont SSO (test_draco_sso) + # a une sémantique « un seul compte en base » dont hérite la suite. + import uuid + uid = models.create_user( + f"machine-{uuid.uuid4().hex[:8]}", + accounts.hash_password("motdepasse"), role="member") + user = models.get_user(uid) + yield user + with models.get_db() as conn: + conn.execute("DELETE FROM api_tokens WHERE user_id = ?", (uid,)) + conn.execute("DELETE FROM users WHERE id = ?", (uid,)) + + +def test_jeton_cycle_de_vie(utilisateur): + row, clair = models.create_api_token(utilisateur["id"], "service-agents") + assert clair.startswith("kbt_") + # La base ne connaît que l'empreinte, jamais le texte clair. + assert row["token_hash"] != clair and len(row["token_hash"]) == 64 + trouve = models.user_by_api_token(clair) + assert trouve is not None and trouve["username"] == utilisateur["username"] + assert models.list_api_tokens(utilisateur["id"])[0]["last_used_at"] is not None + # Révocation : effet immédiat. + assert models.revoke_api_token(utilisateur["id"], row["id"]) is True + assert models.user_by_api_token(clair) is None + # On ne révoque pas le jeton d'autrui. + row2, _clair2 = models.create_api_token(utilisateur["id"], "autre") + assert models.revoke_api_token(utilisateur["id"] + 999, row2["id"]) is False + + +def test_jeton_invalide_ou_compte_desactive(utilisateur): + _row, clair = models.create_api_token(utilisateur["id"], "x") + assert models.user_by_api_token("kbt_nimportequoi") is None + assert models.user_by_api_token("") is None + assert models.user_by_api_token("tck_autreformat") is None + models.update_user(utilisateur["id"], active=0) + assert models.user_by_api_token(clair) is None + + +def test_mcp_par_jeton_quand_auth_active(monkeypatch, utilisateur): + monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno") + monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse") + _row, clair = models.create_api_token(utilisateur["id"], "service-agents") + corps = {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2024-11-05"}} + with TestClient(main.app) as client: + # Sans jeton : 401 (le refus vu en production le 2026-10-21). + assert client.post("/mcp", json=corps).status_code == 401 + # Avec le jeton Bearer : le handshake passe. + r = client.post("/mcp", json=corps, + headers={"Authorization": f"Bearer {clair}"}) + assert r.status_code == 200 + assert r.json()["result"]["serverInfo"]["name"] == "kanbante" + # Mais un jeton n'ouvre pas l'interface web. + assert client.get("/", headers={"Authorization": f"Bearer {clair}", + "Accept": "application/json"}).status_code == 401