140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
"""Local password accounts and opaque, revocable bearer sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import hmac
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
from uuid import uuid4
|
|
|
|
from app import database
|
|
|
|
PASSWORD_ITERATIONS = 310_000
|
|
SESSION_LIFETIME_SECONDS = 7 * 24 * 60 * 60
|
|
_EMAIL_PATTERN = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$")
|
|
_DUMMY_SALT = bytes.fromhex("4f2bc994cb56a7a98b253a615fb57d2c")
|
|
_DUMMY_DIGEST = hashlib.pbkdf2_hmac(
|
|
"sha256", b"constant-time dummy password", _DUMMY_SALT, PASSWORD_ITERATIONS
|
|
)
|
|
|
|
|
|
def normalize_email(email: str) -> str:
|
|
normalized = email.strip().casefold()
|
|
if len(normalized) > 254 or not _EMAIL_PATTERN.fullmatch(normalized):
|
|
raise ValueError("أدخل بريدًا إلكترونيًا صالحًا.")
|
|
return normalized
|
|
|
|
|
|
def _hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, PASSWORD_ITERATIONS
|
|
)
|
|
return "$".join(
|
|
(
|
|
"pbkdf2_sha256",
|
|
str(PASSWORD_ITERATIONS),
|
|
base64.urlsafe_b64encode(salt).decode("ascii"),
|
|
base64.urlsafe_b64encode(digest).decode("ascii"),
|
|
)
|
|
)
|
|
|
|
|
|
def _verify_password(password: str, encoded: str | None) -> bool:
|
|
if encoded is None:
|
|
salt, expected, iterations = _DUMMY_SALT, _DUMMY_DIGEST, PASSWORD_ITERATIONS
|
|
else:
|
|
try:
|
|
algorithm, rounds, salt_text, digest_text = encoded.split("$", 3)
|
|
iterations = int(rounds)
|
|
if algorithm != "pbkdf2_sha256" or not 100_000 <= iterations <= 1_000_000:
|
|
return False
|
|
salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
|
|
expected = base64.urlsafe_b64decode(digest_text.encode("ascii"))
|
|
except (ValueError, UnicodeError, binascii.Error):
|
|
return False
|
|
actual = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, iterations
|
|
)
|
|
return hmac.compare_digest(actual, expected)
|
|
|
|
|
|
def create_account(email: str, password: str) -> str:
|
|
normalized = normalize_email(email)
|
|
if len(password) < 12 or len(password) > 256:
|
|
raise ValueError("يجب أن تتراوح كلمة المرور بين 12 و256 محرفًا.")
|
|
user_id = str(uuid4())
|
|
password_hash = _hash_password(password)
|
|
try:
|
|
with database._connect() as connection:
|
|
connection.execute("INSERT INTO users(id) VALUES (?)", (user_id,))
|
|
connection.execute(
|
|
"""INSERT INTO user_identities(user_id,provider,provider_subject,email,password_hash)
|
|
VALUES(?, 'password', ?, ?, ?)""",
|
|
(user_id, normalized, normalized, password_hash),
|
|
)
|
|
except sqlite3.IntegrityError as exc:
|
|
raise ValueError("يوجد حساب مسجل بهذا البريد الإلكتروني.") from exc
|
|
return user_id
|
|
|
|
|
|
def authenticate(email: str, password: str) -> tuple[str, str] | None:
|
|
normalized = normalize_email(email)
|
|
with database._connect() as connection:
|
|
row = connection.execute(
|
|
"""SELECT user_id,password_hash FROM user_identities
|
|
WHERE provider='password' AND provider_subject=?""",
|
|
(normalized,),
|
|
).fetchone()
|
|
password_hash = row["password_hash"] if row is not None else None
|
|
valid = _verify_password(password, password_hash)
|
|
if row is None or not valid:
|
|
return None
|
|
return row["user_id"], normalized
|
|
|
|
|
|
def issue_session(user_id: str) -> tuple[str, int]:
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = int(time.time()) + SESSION_LIFETIME_SECONDS
|
|
token_hash = hashlib.sha256(token.encode("ascii")).hexdigest()
|
|
with database._connect() as connection:
|
|
connection.execute("DELETE FROM auth_sessions WHERE expires_at <= ?", (int(time.time()),))
|
|
connection.execute(
|
|
"INSERT INTO auth_sessions(token_hash,user_id,expires_at) VALUES(?,?,?)",
|
|
(token_hash, user_id, expires_at),
|
|
)
|
|
return token, expires_at
|
|
|
|
|
|
def resolve_session(token: str) -> str | None:
|
|
if not token or len(token) > 256:
|
|
return None
|
|
token_hash = hashlib.sha256(token.encode("ascii", errors="ignore")).hexdigest()
|
|
now = int(time.time())
|
|
with database._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT user_id FROM auth_sessions WHERE token_hash=? AND expires_at>?",
|
|
(token_hash, now),
|
|
).fetchone()
|
|
return row["user_id"] if row is not None else None
|
|
|
|
|
|
def revoke_session(token: str) -> None:
|
|
token_hash = hashlib.sha256(token.encode("ascii", errors="ignore")).hexdigest()
|
|
with database._connect() as connection:
|
|
connection.execute("DELETE FROM auth_sessions WHERE token_hash=?", (token_hash,))
|
|
|
|
|
|
def account_email(user_id: str) -> str | None:
|
|
with database._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT email FROM user_identities WHERE user_id=? AND provider='password'",
|
|
(user_id,),
|
|
).fetchone()
|
|
return row["email"] if row is not None else None
|