0.25.0 — Jetons d'API personnels (kbt_) pour les surfaces machine

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)

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)

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:

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,

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(

app/templates/profile.html

@@ -46,6 +46,43 @@
             {% endif %}
         </section>
 
+        <section class="briefing-card">
+            <h2>🗝️ Jetons d'API (MCP)</h2>
+            <p class="muted">Pour les consommateurs machine — le service d'agents, en premier lieu. Un jeton (préfixe <code>kbt_</code>) authentifie uniquement <code>/mcp</code> et <code>/api/</code>, jamais l'interface web. Il se présente en en-tête <code>Authorization: Bearer kbt_…</code>. Le texte clair n'est affiché qu'à la création ; ensuite, seule son empreinte est conservée.</p>
+            {% if profile_user.id %}
+            {% if nouveau_jeton %}
+            <p>✅ Nouveau jeton — <strong>copie-le maintenant</strong> :</p>
+            <p><code style="font-size:15px">{{ nouveau_jeton }}</code></p>
+            {% endif %}
+            {% if api_tokens %}
+            <ul>
+                {% for t in api_tokens %}
+                <li>
+                    <strong>{{ t.name }}</strong> · créé le {{ t.created_at }}
+                    · {% if t.last_used_at %}dernier usage : {{ t.last_used_at }}{% else %}jamais utilisé{% endif %}
+                    {% if t.revoked_at %} · <span class="chip-warn">révoqué le {{ t.revoked_at }}</span>
+                    {% else %}
+                    <form method="post" action="/profile/tokens/{{ t.id }}/revoke" class="inline-form">
+                        <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
+                        <button type="submit" class="btn-mini">Révoquer</button>
+                    </form>
+                    {% endif %}
+                </li>
+                {% endfor %}
+            </ul>
+            {% else %}
+            <p class="muted">Aucun jeton pour l'instant.</p>
+            {% endif %}
+            <form method="post" action="/profile/tokens" class="admin-form">
+                <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
+                <label>Nom du jeton <input type="text" name="name" placeholder="service-agents" required></label>
+                <button type="submit" class="btn-save btn-inline">Créer un jeton</button>
+            </form>
+            {% else %}
+            <p class="muted">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.</p>
+            {% endif %}
+        </section>
+
         <section class="briefing-card">
             <h2>✦ Liaison draco (SSO)</h2>
             {% if profile_user.draco_sub %}

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 = [

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