colin/colibre
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases 33 Activity

feat(auth): oauth_identities table + password_hash nullable (#88)

Browse Source 
Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
Colin Maudry
2026-06-24 23:04:18 +02:00
parent 8037f2733e
commit 44031efcd5
2 changed files with 177 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auth/db.py
+82 -2
View File
@@ -10,7 +10,7 @@ USERS_SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
password_hash TEXT,
email_verified INTEGER NOT NULL DEFAULT 0,
pending_email TEXT,
created_at TEXT NOT NULL,
@@ -34,6 +34,15 @@ CREATE TABLE IF NOT EXISTS password_reset_tokens (
created_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS oauth_identities (
provider TEXT NOT NULL,
subject TEXT NOT NULL,
user_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY (provider, subject),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
"""
@@ -66,9 +75,50 @@ def init_schema() -> None:
def _migrate(conn: sqlite3.Connection) -> None:
cols = {row["name"] for row in conn.execute("PRAGMA table_info(users)")}
cols = {row["name"]: row for row in conn.execute("PRAGMA table_info(users)")}
if "pending_email" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN pending_email TEXT")
cols = {row["name"]: row for row in conn.execute("PRAGMA table_info(users)")}
if cols.get("password_hash") and cols["password_hash"]["notnull"] == 1:
_rebuild_users_password_nullable(conn)
def _rebuild_users_password_nullable(conn: sqlite3.Connection) -> None:
# SQLite ne peut pas retirer un NOT NULL via ALTER : on reconstruit la table.
# foreign_keys OFF pour éviter le cascade-delete pendant le DROP.
conn.execute("PRAGMA foreign_keys = OFF")
try:
conn.execute("BEGIN")
conn.execute(
"""
CREATE TABLE users_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
password_hash TEXT,
email_verified INTEGER NOT NULL DEFAULT 0,
pending_email TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
conn.execute(
"INSERT INTO users_new (id, email, password_hash, email_verified, "
"pending_email, created_at, updated_at) "
"SELECT id, email, password_hash, email_verified, pending_email, "
"created_at, updated_at FROM users"
)
conn.execute("DROP TABLE users")
conn.execute("ALTER TABLE users_new RENAME TO users")
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email ON users(email)"
)
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
finally:
conn.execute("PRAGMA foreign_keys = ON")
def _now() -> str:
@@ -198,3 +248,33 @@ def purge_expired_tokens() -> None:
conn = get_conn()
conn.execute("DELETE FROM email_verification_tokens WHERE expires_at <= ?", (now,))
conn.execute("DELETE FROM password_reset_tokens WHERE expires_at <= ?", (now,))
def create_oauth_user(email: str) -> int:
conn = get_conn()
now = _now()
cur = conn.execute(
"INSERT INTO users (email, password_hash, email_verified, created_at, updated_at) "
"VALUES (?, NULL, 1, ?, ?)",
(email.lower(), now, now),
)
return cur.lastrowid
def get_oauth_identity(provider: str, subject: str) -> sqlite3.Row | None:
return (
get_conn()
.execute(
"SELECT * FROM oauth_identities WHERE provider = ? AND subject = ?",
(provider, subject),
)
.fetchone()
)
def link_oauth_identity(provider: str, subject: str, user_id: int) -> None:
get_conn().execute(
"INSERT INTO oauth_identities (provider, subject, user_id, created_at) "
"VALUES (?, ?, ?, ?)",
(provider, subject, user_id, _now()),
)
tests/auth/test_oauth_db.py
+95
View File
@@ -0,0 +1,95 @@
import sqlite3
import pytest
from src.auth import db
def test_oauth_identities_table_created(users_db_path):
db.init_schema()
tables = {
r[0]
for r in db.get_conn().execute(
"SELECT name FROM sqlite_master WHERE type='table'"
)
}
assert "oauth_identities" in tables
def test_password_hash_is_nullable_on_fresh_schema(users_db_path):
db.init_schema()
cols = {r["name"]: r for r in db.get_conn().execute("PRAGMA table_info(users)")}
assert cols["password_hash"]["notnull"] == 0
def test_create_oauth_user(users_db_path):
db.init_schema()
uid = db.create_oauth_user("oauth@example.com")
row = db.get_user_by_id(uid)
assert row["email"] == "oauth@example.com"
assert row["password_hash"] is None
assert row["email_verified"] == 1
def test_link_and_get_oauth_identity(users_db_path):
db.init_schema()
uid = db.create_oauth_user("oauth@example.com")
assert db.get_oauth_identity("linkedin", "sub-123") is None
db.link_oauth_identity("linkedin", "sub-123", uid)
row = db.get_oauth_identity("linkedin", "sub-123")
assert row["user_id"] == uid
def test_oauth_identity_pk_prevents_duplicate(users_db_path):
db.init_schema()
uid = db.create_oauth_user("oauth@example.com")
db.link_oauth_identity("linkedin", "sub-123", uid)
with pytest.raises(sqlite3.IntegrityError):
db.link_oauth_identity("linkedin", "sub-123", uid)
def test_migration_makes_legacy_password_hash_nullable(users_db_path):
# Simule un schéma hérité où password_hash est NOT NULL.
conn = db.get_conn()
conn.executescript(
"""
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
email_verified INTEGER NOT NULL DEFAULT 0,
pending_email TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE email_verification_tokens (
token_hash TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
"""
)
now = db._now()
conn.execute(
"INSERT INTO users (id, email, password_hash, email_verified, created_at, updated_at)"
" VALUES (1, 'legacy@example.com', 'hash', 1, ?, ?)",
(now, now),
)
conn.execute(
"INSERT INTO email_verification_tokens (token_hash, user_id, expires_at, created_at)"
" VALUES ('tok', 1, ?, ?)",
(now, now),
)
db.init_schema() # déclenche _migrate
cols = {r["name"]: r for r in conn.execute("PRAGMA table_info(users)")}
assert cols["password_hash"]["notnull"] == 0
# Données préservées, pas de cascade-delete déclenchée pendant la reconstruction.
assert db.get_user_by_id(1)["email"] == "legacy@example.com"
assert (
conn.execute("SELECT COUNT(*) FROM email_verification_tokens").fetchone()[0]
== 1
)