125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import sqlite3
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import VerifyMismatchError
|
|
|
|
from .config import Settings
|
|
|
|
_hasher = PasswordHasher()
|
|
|
|
#: Session cookie name.
|
|
SESSION_COOKIE = "bike_session"
|
|
CSRF_COOKIE = "bike_csrf"
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
try:
|
|
return _hasher.verify(password_hash, password)
|
|
except VerifyMismatchError:
|
|
return False
|
|
|
|
|
|
def _token() -> str:
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC).replace(tzinfo=None)
|
|
|
|
|
|
# --- sessions ---
|
|
def create_session(conn: sqlite3.Connection, user_id: int, settings: Settings) -> str:
|
|
token = _token()
|
|
created = _now()
|
|
expires = created + timedelta(seconds=settings.session_ttl_seconds)
|
|
conn.execute(
|
|
"INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
|
|
(
|
|
token,
|
|
user_id,
|
|
created.isoformat(sep=" "),
|
|
expires.isoformat(sep=" "),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return token
|
|
|
|
|
|
def get_user_id_for_session(conn: sqlite3.Connection, token: str | None) -> int | None:
|
|
if not token:
|
|
return None
|
|
r = conn.execute(
|
|
"SELECT user_id, expires_at FROM sessions WHERE token = ? LIMIT 1", (token,)
|
|
).fetchone()
|
|
if r is None:
|
|
return None
|
|
expires = datetime.fromisoformat(r["expires_at"])
|
|
if expires <= _now():
|
|
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
conn.commit()
|
|
return None
|
|
return int(r["user_id"])
|
|
|
|
|
|
def delete_session(conn: sqlite3.Connection, token: str | None) -> None:
|
|
if not token:
|
|
return
|
|
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
conn.commit()
|
|
|
|
|
|
# --- CSRF ---
|
|
def csrf_token(secret_key: str, session_token: str | None) -> str:
|
|
h = hmac.new(secret_key.encode(), session_token.encode(), hashlib.sha256).hexdigest()
|
|
return h[:32]
|
|
|
|
|
|
def check_csrf(secret_key: str, session_token: str | None, provided: str | None) -> bool:
|
|
if not provided:
|
|
return False
|
|
expected = csrf_token(secret_key, session_token)
|
|
return hmac.compare_digest(expected, provided)
|
|
|
|
|
|
# --- login throttling ---
|
|
def is_throttled(conn: sqlite3.Connection, ip: str, settings: Settings) -> bool:
|
|
window_start = (_now() - timedelta(seconds=settings.login_window_seconds)).isoformat(sep=" ")
|
|
r = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM login_attempts WHERE ip = ? AND ts >= ?",
|
|
(ip, window_start),
|
|
).fetchone()
|
|
return int(r["n"]) >= settings.login_max_attempts
|
|
|
|
|
|
def record_login_attempt(conn: sqlite3.Connection, ip: str) -> None:
|
|
ts = _now().isoformat(sep=" ")
|
|
conn.execute("INSERT INTO login_attempts (ip, ts) VALUES (?, ?)", (ip, ts))
|
|
conn.commit()
|
|
|
|
|
|
def cleanup_old_attempts(conn: sqlite3.Connection, settings: Settings) -> None:
|
|
cutoff = (_now() - timedelta(seconds=settings.login_window_seconds * 2)).isoformat(sep=" ")
|
|
conn.execute("DELETE FROM login_attempts WHERE ts < ?", (cutoff,))
|
|
conn.commit()
|
|
|
|
|
|
def client_ip(request, trust_proxy: bool) -> str:
|
|
"""Best-effort remote IP. Behind a proxy, trust the first X-Forwarded-For hop."""
|
|
if trust_proxy:
|
|
ffor = request.headers.get("x-forwarded-for")
|
|
if ffor:
|
|
return ffor.split(",")[0].strip()
|
|
if request.client is not None:
|
|
return request.client.host or "unknown"
|
|
return "unknown"
|