diff --git a/src/auth/tokens.py b/src/auth/tokens.py new file mode 100644 index 0000000..40dcea0 --- /dev/null +++ b/src/auth/tokens.py @@ -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 diff --git a/tests/auth/test_tokens.py b/tests/auth/test_tokens.py new file mode 100644 index 0000000..021932b --- /dev/null +++ b/tests/auth/test_tokens.py @@ -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