Jetons MCP chiffrés au repos + bouton « Copier le jeton » sur /compte/mcp

Les jetons MCP sont désormais chiffrés (Fernet dérivée de SECRET_KEY) en plus
du hash conservé pour l'auth, permettant leur ré-affichage/copie à tout moment.
Getter scopé au propriétaire, dégradation propre si clé absente/changée.
Migration 0013 + colonne token_enc. UI alignée sur « Copier le lien » des vues.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Colin Maudry
2026-07-14 23:55:01 +02:00
parent 43d8ba5098
commit f571627a6b
6 changed files with 172 additions and 11 deletions
+3
View File
@@ -30,6 +30,9 @@ MATOMO_TRACKING_ENABLED=true
# Comptes utilisateurs # Comptes utilisateurs
USERS_DB_PATH=./users.sqlite USERS_DB_PATH=./users.sqlite
# Sert aussi à chiffrer les jetons MCP stockés (fonction « Copier le jeton » sur
# /compte/mcp). Sa rotation déconnecte les sessions ET rend les jetons existants
# non ré-affichables (les utilisateurs devront en régénérer).
SECRET_KEY= # à générer : python -c "import secrets; print(secrets.token_hex(32))" SECRET_KEY= # à générer : python -c "import secrets; print(secrets.token_hex(32))"
APP_BASE_URL=http://localhost:8050 APP_BASE_URL=http://localhost:8050
+45 -3
View File
@@ -1,16 +1,21 @@
import base64
import hashlib import hashlib
import os
import secrets import secrets
import sqlite3 import sqlite3
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from cryptography.fernet import Fernet, InvalidToken
TOKEN_PREFIX = "colibre_" TOKEN_PREFIX = "colibre_"
SCHEMA = """ SCHEMA = """
CREATE TABLE IF NOT EXISTS api_tokens ( CREATE TABLE IF NOT EXISTS api_tokens (
id INTEGER PRIMARY KEY, id INTEGER PRIMARY KEY,
token_hash TEXT NOT NULL UNIQUE, token_hash TEXT NOT NULL UNIQUE,
token_enc TEXT,
label TEXT NOT NULL, label TEXT NOT NULL,
user_id INTEGER, user_id INTEGER,
created_at TEXT NOT NULL, created_at TEXT NOT NULL,
@@ -31,6 +36,19 @@ def _hash(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest() return hashlib.sha256(token.encode()).hexdigest()
def _fernet_key() -> bytes | None:
"""Clé Fernet dérivée de SECRET_KEY (séparation de domaine).
Renvoie None si SECRET_KEY est absent : le jeton n'est alors pas chiffré
ni ré-affichable (dégradation propre, sans échec).
"""
secret = os.getenv("SECRET_KEY")
if not secret:
return None
digest = hashlib.sha256(f"mcp-token-enc:{secret}".encode()).digest()
return base64.urlsafe_b64encode(digest)
@contextmanager @contextmanager
def _connect(db_path): def _connect(db_path):
conn = sqlite3.connect(str(db_path)) conn = sqlite3.connect(str(db_path))
@@ -52,11 +70,13 @@ def create_token(
db_path, label: str, user_id: int | None = None, kind: str = "api" db_path, label: str, user_id: int | None = None, kind: str = "api"
) -> tuple[str, int]: ) -> tuple[str, int]:
token = TOKEN_PREFIX + secrets.token_hex(32) token = TOKEN_PREFIX + secrets.token_hex(32)
key = _fernet_key()
token_enc = Fernet(key).encrypt(token.encode()).decode() if key else None
with _connect(db_path) as conn: with _connect(db_path) as conn:
cur = conn.execute( cur = conn.execute(
"INSERT INTO api_tokens (token_hash, label, user_id, kind, created_at) " "INSERT INTO api_tokens (token_hash, token_enc, label, user_id, kind, "
"VALUES (?, ?, ?, ?, ?)", "created_at) VALUES (?, ?, ?, ?, ?, ?)",
(_hash(token), label, user_id, kind, _utcnow_iso()), (_hash(token), token_enc, label, user_id, kind, _utcnow_iso()),
) )
conn.commit() conn.commit()
return token, cur.lastrowid return token, cur.lastrowid
@@ -71,6 +91,28 @@ def get_token_by_plaintext(db_path, token: str) -> dict | None:
return dict(row) if row else None return dict(row) if row else None
def get_token_plaintext_for_user(db_path, token_id: int, user_id: int) -> str | None:
"""Déchiffre et renvoie le jeton en clair, uniquement pour son propriétaire.
Renvoie None si : SECRET_KEY absent, jeton inexistant/appartenant à un autre
utilisateur, jamais chiffré (token_enc NULL), ou indéchiffrable (clé changée).
"""
key = _fernet_key()
if key is None:
return None
with _connect(db_path) as conn:
row = conn.execute(
"SELECT token_enc FROM api_tokens WHERE id = ? AND user_id = ?",
(token_id, user_id),
).fetchone()
if row is None or row["token_enc"] is None:
return None
try:
return Fernet(key).decrypt(row["token_enc"].encode()).decode()
except InvalidToken:
return None
def revoke_token(db_path, token_id: int) -> None: def revoke_token(db_path, token_id: int) -> None:
with _connect(db_path) as conn: with _connect(db_path) as conn:
conn.execute( conn.execute(
+4
View File
@@ -78,6 +78,10 @@ _MIGRATIONS: list[tuple[str, str]] = [
"0012_add_token_to_saved_views", "0012_add_token_to_saved_views",
"ALTER TABLE saved_views ADD COLUMN token TEXT", "ALTER TABLE saved_views ADD COLUMN token TEXT",
), ),
(
"0013_add_token_enc_to_api_tokens",
"ALTER TABLE api_tokens ADD COLUMN token_enc TEXT",
),
] ]
+49 -7
View File
@@ -140,14 +140,48 @@ def prompt_tips():
) )
def _copy_token_button(token: str):
"""Bouton « Copier le jeton » (même design que « Copier le lien » des vues)."""
return dcc.Clipboard(
content=token,
title="Copier le jeton",
className="btn btn-outline-secondary btn-sm d-inline-flex "
"align-items-center me-2",
children=[
html.Img(
src="/assets/copy.svg",
alt="",
style={
"height": "1em",
"verticalAlign": "-0.15em",
"marginRight": "0.35em",
},
),
"Copier le jeton",
],
copied_children="✓ Copié",
)
def _token_row(row: dict): def _token_row(row: dict):
statut = "Révoqué" if row["revoked_at"] else "Actif" statut = "Révoqué" if row["revoked_at"] else "Actif"
actions = [] actions = []
if not row["revoked_at"]: if not row["revoked_at"]:
actions = [ token_plain = row.get("token_plain")
if token_plain:
actions.append(_copy_token_button(token_plain))
else:
actions.append(
html.Span(
"Régénérez pour copier",
className="text-muted small me-2",
)
)
actions.append(
html.Form( html.Form(
method="POST", method="POST",
action=f"/compte/mcp/revoquer/{row['id']}", action=f"/compte/mcp/revoquer/{row['id']}",
className="d-inline",
children=[ children=[
_csrf(f"revoke-{row['id']}"), _csrf(f"revoke-{row['id']}"),
dbc.Button( dbc.Button(
@@ -159,7 +193,7 @@ def _token_row(row: dict):
), ),
], ],
) )
] )
return html.Tr( return html.Tr(
[ [
html.Td(row["label"]), html.Td(row["label"]),
@@ -196,17 +230,25 @@ def layout(**_):
alerts.append( alerts.append(
dbc.Alert( dbc.Alert(
[ [
html.Strong("Votre nouveau jeton (copiez-le maintenant, "), html.Strong(
html.Strong("il ne sera plus affiché) :"), "Votre nouveau jeton (vous pourrez le copier à tout "
"moment via « Copier le jeton ») :"
),
html.Pre(html.Code(new_token)), html.Pre(html.Code(new_token)),
], ],
color="success", color="success",
) )
) )
tokens = tokens_db.list_user_tokens( db_path = os.environ["USERS_DB_PATH"]
os.environ["USERS_DB_PATH"], current_user.id, "mcp" tokens = tokens_db.list_user_tokens(db_path, current_user.id, "mcp")
token_rows = []
for r in tokens:
row = dict(r)
row["token_plain"] = tokens_db.get_token_plaintext_for_user(
db_path, row["id"], current_user.id
) )
token_rows.append(_token_row(row))
table = ( table = (
dbc.Table( dbc.Table(
[ [
@@ -221,7 +263,7 @@ def layout(**_):
] ]
) )
), ),
html.Tbody([_token_row(dict(r)) for r in tokens]), html.Tbody(token_rows),
], ],
striped=True, striped=True,
bordered=False, bordered=False,
+31
View File
@@ -101,3 +101,34 @@ def test_revoke_user_token_already_revoked_returns_false(temp_db):
_, token_id = tokens_db.create_token(temp_db, "x", user_id=1, kind="mcp") _, token_id = tokens_db.create_token(temp_db, "x", user_id=1, kind="mcp")
assert tokens_db.revoke_user_token(temp_db, token_id, 1) is True assert tokens_db.revoke_user_token(temp_db, token_id, 1) is True
assert tokens_db.revoke_user_token(temp_db, token_id, 1) is False assert tokens_db.revoke_user_token(temp_db, token_id, 1) is False
def test_create_token_stores_recoverable_encrypted_token(temp_db, monkeypatch):
monkeypatch.setenv("SECRET_KEY", "s3cr3t-test-key")
token, token_id = tokens_db.create_token(temp_db, "x", user_id=7, kind="mcp")
with sqlite3.connect(str(temp_db)) as conn:
enc = conn.execute(
"SELECT token_enc FROM api_tokens WHERE id = ?", (token_id,)
).fetchone()[0]
assert enc is not None
assert token not in enc # chiffré, jamais en clair dans la base
assert tokens_db.get_token_plaintext_for_user(temp_db, token_id, 7) == token
def test_get_token_plaintext_scoped_to_owner(temp_db, monkeypatch):
monkeypatch.setenv("SECRET_KEY", "s3cr3t-test-key")
_, token_id = tokens_db.create_token(temp_db, "x", user_id=1, kind="mcp")
assert tokens_db.get_token_plaintext_for_user(temp_db, token_id, 999) is None
def test_get_token_plaintext_none_when_no_key(temp_db, monkeypatch):
monkeypatch.delenv("SECRET_KEY", raising=False)
_, token_id = tokens_db.create_token(temp_db, "x", user_id=1, kind="mcp")
assert tokens_db.get_token_plaintext_for_user(temp_db, token_id, 1) is None
def test_get_token_plaintext_none_after_key_rotation(temp_db, monkeypatch):
monkeypatch.setenv("SECRET_KEY", "key-A")
_, token_id = tokens_db.create_token(temp_db, "x", user_id=1, kind="mcp")
monkeypatch.setenv("SECRET_KEY", "key-B") # rotation → indéchiffrable
assert tokens_db.get_token_plaintext_for_user(temp_db, token_id, 1) is None
+39
View File
@@ -44,6 +44,45 @@ def test_prompt_tips_mentions_columns_and_examples():
assert "lien" in text # mention du lien vers la fiche marché assert "lien" in text # mention du lien vers la fiche marché
def test_token_row_has_copy_button_when_token_available():
from src.app import app # noqa: F401
from src.pages.compte.mcp import _token_row
text = str(
_token_row(
{
"id": 42,
"label": "l",
"created_at": "c",
"last_used_at": None,
"revoked_at": None,
"token_plain": "colibre_abc123",
}
)
)
assert "Copier le jeton" in text
assert "colibre_abc123" in text # contenu du presse-papier (dcc.Clipboard)
def test_token_row_no_copy_button_when_token_unavailable():
from src.app import app # noqa: F401
from src.pages.compte.mcp import _token_row
text = str(
_token_row(
{
"id": 42,
"label": "l",
"created_at": "c",
"last_used_at": None,
"revoked_at": None,
"token_plain": None, # ancien jeton haché, non récupérable
}
)
)
assert "Copier le jeton" not in text
def _collect_titles(component): def _collect_titles(component):
# Parcourt récursivement les AccordionItem pour collecter leurs `title`. # Parcourt récursivement les AccordionItem pour collecter leurs `title`.
found = [] found = []