diff --git a/src/api/tokens_db.py b/src/api/tokens_db.py new file mode 100644 index 0000000..76ea1f0 --- /dev/null +++ b/src/api/tokens_db.py @@ -0,0 +1,94 @@ +import hashlib +import secrets +import sqlite3 +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path + +TOKEN_PREFIX = "decpinfo_" + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS api_tokens ( + id INTEGER PRIMARY KEY, + token_hash TEXT NOT NULL UNIQUE, + label TEXT NOT NULL, + user_id INTEGER, + created_at TEXT NOT NULL, + last_used_at TEXT, + count_total INTEGER NOT NULL DEFAULT 0, + revoked_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash); +""" + + +def _utcnow_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _hash(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +@contextmanager +def _connect(db_path): + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + +def init_schema(db_path) -> None: + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + with _connect(db_path) as conn: + conn.executescript(SCHEMA) + conn.commit() + + +def create_token(db_path, label: str, user_id: int | None = None) -> tuple[str, int]: + token = TOKEN_PREFIX + secrets.token_hex(32) + with _connect(db_path) as conn: + cur = conn.execute( + "INSERT INTO api_tokens (token_hash, label, user_id, created_at) " + "VALUES (?, ?, ?, ?)", + (_hash(token), label, user_id, _utcnow_iso()), + ) + conn.commit() + return token, cur.lastrowid + + +def get_token_by_plaintext(db_path, token: str) -> dict | None: + with _connect(db_path) as conn: + row = conn.execute( + "SELECT * FROM api_tokens WHERE token_hash = ?", + (_hash(token),), + ).fetchone() + return dict(row) if row else None + + +def revoke_token(db_path, token_id: int) -> None: + with _connect(db_path) as conn: + conn.execute( + "UPDATE api_tokens SET revoked_at = ? WHERE id = ?", + (_utcnow_iso(), token_id), + ) + conn.commit() + + +def increment_usage(db_path, token_id: int) -> None: + with _connect(db_path) as conn: + conn.execute( + "UPDATE api_tokens " + "SET count_total = count_total + 1, last_used_at = ? " + "WHERE id = ?", + (_utcnow_iso(), token_id), + ) + conn.commit() + + +def list_tokens(db_path) -> list[dict]: + with _connect(db_path) as conn: + rows = conn.execute("SELECT * FROM api_tokens ORDER BY id").fetchall() + return [dict(r) for r in rows] diff --git a/tests/api/conftest.py b/tests/api/conftest.py new file mode 100644 index 0000000..7baca3f --- /dev/null +++ b/tests/api/conftest.py @@ -0,0 +1,12 @@ +import pytest + + +@pytest.fixture +def temp_db(tmp_path, monkeypatch): + """Une SQLite éphémère pour les tests qui modifient la DB.""" + db_path = tmp_path / "users.test.sqlite" + monkeypatch.setenv("USERS_DB_PATH", str(db_path)) + from src.api import tokens_db + + tokens_db.init_schema(db_path) + return db_path diff --git a/tests/api/test_tokens_db.py b/tests/api/test_tokens_db.py new file mode 100644 index 0000000..91ca0b0 --- /dev/null +++ b/tests/api/test_tokens_db.py @@ -0,0 +1,64 @@ +import sqlite3 + +from src.api import tokens_db + + +def test_init_schema_creates_table(temp_db): + with sqlite3.connect(str(temp_db)) as conn: + rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='api_tokens'" + ).fetchall() + assert rows == [("api_tokens",)] + + +def test_create_token_returns_plaintext_and_stores_hash(temp_db): + token, token_id = tokens_db.create_token(temp_db, "test-label") + assert token.startswith("decpinfo_") + assert len(token) == len("decpinfo_") + 64 # 32 octets hex = 64 chars + assert token_id >= 1 + + with sqlite3.connect(str(temp_db)) as conn: + row = conn.execute( + "SELECT token_hash, label, count_total FROM api_tokens WHERE id = ?", + (token_id,), + ).fetchone() + assert row[1] == "test-label" + assert row[2] == 0 + assert row[0] != token # stocké en clair impossible + assert len(row[0]) == 64 # sha256 hex + + +def test_get_token_by_plaintext_returns_row(temp_db): + token, token_id = tokens_db.create_token(temp_db, "x") + row = tokens_db.get_token_by_plaintext(temp_db, token) + assert row is not None + assert row["id"] == token_id + assert row["label"] == "x" + assert row["revoked_at"] is None + + +def test_get_token_unknown_returns_none(temp_db): + assert tokens_db.get_token_by_plaintext(temp_db, "decpinfo_zzz") is None + + +def test_revoke_token_sets_revoked_at(temp_db): + token, token_id = tokens_db.create_token(temp_db, "x") + tokens_db.revoke_token(temp_db, token_id) + row = tokens_db.get_token_by_plaintext(temp_db, token) + assert row["revoked_at"] is not None + + +def test_increment_usage_updates_counter_and_timestamp(temp_db): + token, token_id = tokens_db.create_token(temp_db, "x") + tokens_db.increment_usage(temp_db, token_id) + tokens_db.increment_usage(temp_db, token_id) + row = tokens_db.get_token_by_plaintext(temp_db, token) + assert row["count_total"] == 2 + assert row["last_used_at"] is not None + + +def test_list_tokens_returns_all(temp_db): + tokens_db.create_token(temp_db, "a") + tokens_db.create_token(temp_db, "b") + rows = tokens_db.list_tokens(temp_db) + assert [r["label"] for r in rows] == ["a", "b"]