diff --git a/.template.env b/.template.env index af5f046..957c6eb 100644 --- a/.template.env +++ b/.template.env @@ -30,6 +30,9 @@ MATOMO_TRACKING_ENABLED=true # Comptes utilisateurs 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))" APP_BASE_URL=http://localhost:8050 diff --git a/src/api/tokens_db.py b/src/api/tokens_db.py index 068e8aa..c5940f3 100644 --- a/src/api/tokens_db.py +++ b/src/api/tokens_db.py @@ -1,16 +1,21 @@ +import base64 import hashlib +import os import secrets import sqlite3 from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path +from cryptography.fernet import Fernet, InvalidToken + TOKEN_PREFIX = "colibre_" SCHEMA = """ CREATE TABLE IF NOT EXISTS api_tokens ( id INTEGER PRIMARY KEY, token_hash TEXT NOT NULL UNIQUE, + token_enc TEXT, label TEXT NOT NULL, user_id INTEGER, created_at TEXT NOT NULL, @@ -31,6 +36,19 @@ def _hash(token: str) -> str: 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 def _connect(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" ) -> tuple[str, int]: 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: cur = conn.execute( - "INSERT INTO api_tokens (token_hash, label, user_id, kind, created_at) " - "VALUES (?, ?, ?, ?, ?)", - (_hash(token), label, user_id, kind, _utcnow_iso()), + "INSERT INTO api_tokens (token_hash, token_enc, label, user_id, kind, " + "created_at) VALUES (?, ?, ?, ?, ?, ?)", + (_hash(token), token_enc, label, user_id, kind, _utcnow_iso()), ) conn.commit() 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 +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: with _connect(db_path) as conn: conn.execute( diff --git a/src/migrations.py b/src/migrations.py index 6812a83..379593b 100644 --- a/src/migrations.py +++ b/src/migrations.py @@ -78,6 +78,10 @@ _MIGRATIONS: list[tuple[str, str]] = [ "0012_add_token_to_saved_views", "ALTER TABLE saved_views ADD COLUMN token TEXT", ), + ( + "0013_add_token_enc_to_api_tokens", + "ALTER TABLE api_tokens ADD COLUMN token_enc TEXT", + ), ] diff --git a/src/pages/compte/mcp.py b/src/pages/compte/mcp.py index d6af9d8..ec4b7c6 100644 --- a/src/pages/compte/mcp.py +++ b/src/pages/compte/mcp.py @@ -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): statut = "Révoqué" if row["revoked_at"] else "Actif" actions = [] 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( method="POST", action=f"/compte/mcp/revoquer/{row['id']}", + className="d-inline", children=[ _csrf(f"revoke-{row['id']}"), dbc.Button( @@ -159,7 +193,7 @@ def _token_row(row: dict): ), ], ) - ] + ) return html.Tr( [ html.Td(row["label"]), @@ -196,17 +230,25 @@ def layout(**_): alerts.append( dbc.Alert( [ - html.Strong("Votre nouveau jeton (copiez-le maintenant, "), - html.Strong("il ne sera plus affiché) :"), + html.Strong( + "Votre nouveau jeton (vous pourrez le copier à tout " + "moment via « Copier le jeton ») :" + ), html.Pre(html.Code(new_token)), ], color="success", ) ) - tokens = tokens_db.list_user_tokens( - os.environ["USERS_DB_PATH"], current_user.id, "mcp" - ) + db_path = os.environ["USERS_DB_PATH"] + 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 = ( dbc.Table( [ @@ -221,7 +263,7 @@ def layout(**_): ] ) ), - html.Tbody([_token_row(dict(r)) for r in tokens]), + html.Tbody(token_rows), ], striped=True, bordered=False, diff --git a/tests/api/test_tokens_db.py b/tests/api/test_tokens_db.py index 64b34bd..115b7f4 100644 --- a/tests/api/test_tokens_db.py +++ b/tests/api/test_tokens_db.py @@ -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") 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 + + +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 diff --git a/tests/mcp/test_account_page.py b/tests/mcp/test_account_page.py index 8e52dbe..9fbc592 100644 --- a/tests/mcp/test_account_page.py +++ b/tests/mcp/test_account_page.py @@ -44,6 +44,45 @@ def test_prompt_tips_mentions_columns_and_examples(): 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): # Parcourt récursivement les AccordionItem pour collecter leurs `title`. found = []