Batman

app/security.py

@@ -0,0 +1,124 @@
+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"