API: tokens_db SQLite CRUD (#78)

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]

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

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"]