Spark
13182f5d53
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)
72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
"""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
|