156 lines
4.2 KiB
Python
156 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
|
|
from repositories.auth_repository import (
|
|
create_auth_session_record,
|
|
get_auth_session_by_token,
|
|
revoke_auth_session,
|
|
)
|
|
|
|
IDLE_TIMEOUT_SECONDS = 999_999
|
|
SESSION_TOUCH_THROTTLE_SECONDS = 60
|
|
PBKDF2_ITERATIONS = 260_000
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
derived = hashlib.pbkdf2_hmac(
|
|
"sha256",
|
|
password.encode("utf-8"),
|
|
salt,
|
|
PBKDF2_ITERATIONS,
|
|
)
|
|
return (
|
|
f"pbkdf2_sha256${PBKDF2_ITERATIONS}$"
|
|
f"{base64.b64encode(salt).decode()}$"
|
|
f"{base64.b64encode(derived).decode()}"
|
|
)
|
|
|
|
|
|
def verify_password(password: str, stored_hash: str) -> bool:
|
|
try:
|
|
algorithm, iterations_raw, salt_b64, hash_b64 = stored_hash.split("$", 3)
|
|
if algorithm != "pbkdf2_sha256":
|
|
return False
|
|
iterations = int(iterations_raw)
|
|
salt = base64.b64decode(salt_b64)
|
|
expected = base64.b64decode(hash_b64)
|
|
except Exception:
|
|
return False
|
|
|
|
actual = hashlib.pbkdf2_hmac(
|
|
"sha256",
|
|
password.encode("utf-8"),
|
|
salt,
|
|
iterations,
|
|
)
|
|
return hmac.compare_digest(actual, expected)
|
|
|
|
|
|
def create_auth_session(
|
|
user_id: int,
|
|
ip_address: str | None = None,
|
|
user_agent: str | None = None,
|
|
) -> str:
|
|
token = secrets.token_urlsafe(48)
|
|
expires_at = datetime.now(timezone.utc) + timedelta(seconds=IDLE_TIMEOUT_SECONDS)
|
|
create_auth_session_record(
|
|
user_id=user_id,
|
|
session_token=token,
|
|
expires_at=expires_at,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
)
|
|
return token
|
|
|
|
|
|
def _is_api_request(request: Request) -> bool:
|
|
path = request.url.path
|
|
if request.method in {"POST", "PUT", "PATCH", "DELETE"}:
|
|
return True
|
|
if path.endswith("/events") or "/event/" in path or path.endswith("/event"):
|
|
return True
|
|
accept = request.headers.get("accept", "")
|
|
requested_with = request.headers.get("x-requested-with", "")
|
|
return "application/json" in accept or requested_with.lower() == "xmlhttprequest"
|
|
|
|
|
|
def build_not_authenticated_response(request: Request):
|
|
if _is_api_request(request):
|
|
return JSONResponse({"error": "auth_expired"}, status_code=401)
|
|
return RedirectResponse(url="/login?reason=idle", status_code=303)
|
|
|
|
|
|
def revoke_auth_session_by_request(request: Request):
|
|
token = request.cookies.get("auth_token")
|
|
if token:
|
|
revoke_auth_session(token)
|
|
|
|
|
|
def get_current_user_from_request(request: Request):
|
|
token = request.cookies.get("auth_token")
|
|
if not token:
|
|
return None
|
|
|
|
row = get_auth_session_by_token(token)
|
|
if not row:
|
|
return None
|
|
|
|
(
|
|
session_id,
|
|
user_id,
|
|
session_token,
|
|
created_at,
|
|
last_activity_at,
|
|
expires_at,
|
|
revoked_at,
|
|
ip_address,
|
|
user_agent,
|
|
username,
|
|
is_active,
|
|
role,
|
|
) = row
|
|
|
|
if revoked_at is not None or not is_active:
|
|
return None
|
|
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if last_activity_at.tzinfo is None:
|
|
last_activity_at = last_activity_at.replace(tzinfo=timezone.utc)
|
|
if expires_at.tzinfo is None:
|
|
expires_at = expires_at.replace(tzinfo=timezone.utc)
|
|
|
|
idle_seconds = (now - last_activity_at).total_seconds()
|
|
if idle_seconds > IDLE_TIMEOUT_SECONDS or expires_at < now:
|
|
revoke_auth_session(token)
|
|
return None
|
|
|
|
new_expires_at = now + timedelta(seconds=IDLE_TIMEOUT_SECONDS)
|
|
from repositories.auth_repository import touch_auth_session_if_needed
|
|
touch_auth_session_if_needed(
|
|
session_token=token,
|
|
expires_at=new_expires_at,
|
|
throttle_seconds=SESSION_TOUCH_THROTTLE_SECONDS,
|
|
)
|
|
|
|
return {
|
|
"session_id": session_id,
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"created_at": created_at,
|
|
"last_activity_at": last_activity_at,
|
|
"expires_at": expires_at,
|
|
"ip_address": ip_address,
|
|
"user_agent": user_agent,
|
|
"role": role,
|
|
}
|