src/auth/tokens.py : génération et validation des tokens (#73)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,61 @@
|
|||||||
|
import hashlib
|
||||||
|
import secrets
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
|
from src.auth import db
|
||||||
|
|
||||||
|
VERIFICATION_TTL_HOURS = 24
|
||||||
|
RESET_TTL_HOURS = 1
|
||||||
|
|
||||||
|
|
||||||
|
def hash_token(plain: str) -> str:
|
||||||
|
return hashlib.sha256(plain.encode("utf-8")).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _expires_at(hours: int) -> str:
|
||||||
|
return (datetime.now(timezone.utc) + timedelta(hours=hours)).isoformat()
|
||||||
|
|
||||||
|
|
||||||
|
def create_verification_token(
|
||||||
|
user_id: int, expires_in_hours: int = VERIFICATION_TTL_HOURS
|
||||||
|
) -> str:
|
||||||
|
plain = secrets.token_urlsafe(32)
|
||||||
|
db.create_email_verification_token(
|
||||||
|
hash_token(plain), user_id, _expires_at(expires_in_hours)
|
||||||
|
)
|
||||||
|
return plain
|
||||||
|
|
||||||
|
|
||||||
|
def consume_verification_token(plain: str) -> int | None:
|
||||||
|
row = db.find_email_verification_token(hash_token(plain))
|
||||||
|
if row is None:
|
||||||
|
return None
|
||||||
|
user_id = row["user_id"]
|
||||||
|
db.delete_email_verification_tokens_for_user(user_id)
|
||||||
|
return user_id
|
||||||
|
|
||||||
|
|
||||||
|
def create_password_reset_token(
|
||||||
|
user_id: int, expires_in_hours: int = RESET_TTL_HOURS
|
||||||
|
) -> str:
|
||||||
|
db.delete_password_reset_tokens_for_user(user_id)
|
||||||
|
plain = secrets.token_urlsafe(32)
|
||||||
|
db.create_password_reset_token(
|
||||||
|
hash_token(plain), user_id, _expires_at(expires_in_hours)
|
||||||
|
)
|
||||||
|
return plain
|
||||||
|
|
||||||
|
|
||||||
|
def consume_password_reset_token(plain: str) -> int | None:
|
||||||
|
row = db.find_password_reset_token(hash_token(plain))
|
||||||
|
if row is None:
|
||||||
|
return None
|
||||||
|
user_id = row["user_id"]
|
||||||
|
db.delete_password_reset_tokens_for_user(user_id)
|
||||||
|
return user_id
|
||||||
|
|
||||||
|
|
||||||
|
def validate_password_reset_token(plain: str) -> int | None:
|
||||||
|
"""Check token without consuming (used to render the reset form)."""
|
||||||
|
row = db.find_password_reset_token(hash_token(plain))
|
||||||
|
return row["user_id"] if row else None
|
||||||
@@ -0,0 +1,68 @@
|
|||||||
|
from src.auth import db
|
||||||
|
from src.auth.tokens import (
|
||||||
|
consume_password_reset_token,
|
||||||
|
consume_verification_token,
|
||||||
|
create_password_reset_token,
|
||||||
|
create_verification_token,
|
||||||
|
hash_token,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_hash_token_is_stable():
|
||||||
|
t = "abc123"
|
||||||
|
assert hash_token(t) == hash_token(t)
|
||||||
|
assert hash_token(t) != hash_token("abc124")
|
||||||
|
assert len(hash_token(t)) == 64 # sha256 hex
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_verification_token_returns_plain_token(users_db_path):
|
||||||
|
db.init_schema()
|
||||||
|
uid = db.create_user("a@b.c", "h")
|
||||||
|
plain = create_verification_token(uid)
|
||||||
|
assert isinstance(plain, str)
|
||||||
|
assert len(plain) >= 32
|
||||||
|
# stocké en DB sous forme hashée
|
||||||
|
row = db.find_email_verification_token(hash_token(plain))
|
||||||
|
assert row is not None
|
||||||
|
assert row["user_id"] == uid
|
||||||
|
|
||||||
|
|
||||||
|
def test_consume_verification_token_succeeds_once(users_db_path):
|
||||||
|
db.init_schema()
|
||||||
|
uid = db.create_user("a@b.c", "h")
|
||||||
|
plain = create_verification_token(uid)
|
||||||
|
user_id = consume_verification_token(plain)
|
||||||
|
assert user_id == uid
|
||||||
|
# usage unique : les tokens de cet user sont supprimés
|
||||||
|
assert consume_verification_token(plain) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_consume_invalid_verification_token(users_db_path):
|
||||||
|
db.init_schema()
|
||||||
|
assert consume_verification_token("n-existe-pas") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_verification_token_expires(users_db_path, monkeypatch):
|
||||||
|
db.init_schema()
|
||||||
|
uid = db.create_user("a@b.c", "h")
|
||||||
|
plain = create_verification_token(uid, expires_in_hours=-1)
|
||||||
|
assert consume_verification_token(plain) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_password_reset_token_deletes_previous(users_db_path):
|
||||||
|
db.init_schema()
|
||||||
|
uid = db.create_user("a@b.c", "h")
|
||||||
|
old = create_password_reset_token(uid)
|
||||||
|
new = create_password_reset_token(uid)
|
||||||
|
# l'ancien a été supprimé
|
||||||
|
assert consume_password_reset_token(old) is None
|
||||||
|
# le nouveau fonctionne
|
||||||
|
assert consume_password_reset_token(new) == uid
|
||||||
|
|
||||||
|
|
||||||
|
def test_consume_password_reset_token_is_single_use(users_db_path):
|
||||||
|
db.init_schema()
|
||||||
|
uid = db.create_user("a@b.c", "h")
|
||||||
|
plain = create_password_reset_token(uid)
|
||||||
|
assert consume_password_reset_token(plain) == uid
|
||||||
|
assert consume_password_reset_token(plain) is None
|
||||||
Reference in New Issue
Block a user