121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
import time
|
|
from typing import Optional
|
|
|
|
from fastapi import Depends, HTTPException, Request, WebSocket, status
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
SESSION_COOKIE_NAME = "websurfer_session"
|
|
SESSION_MAX_AGE_SECONDS = 7 * 24 * 3600 # 7 days
|
|
|
|
|
|
def is_auth_enabled() -> bool:
|
|
"""Checks whether authentication is enabled via environment variables."""
|
|
flag = os.getenv("AUTH_ENABLED", "").strip().lower()
|
|
has_credentials = bool(os.getenv("AUTH_USERNAME") and os.getenv("AUTH_PASSWORD"))
|
|
return flag in ("true", "1", "yes") and has_credentials
|
|
|
|
|
|
def get_secret_key() -> str:
|
|
"""Returns the secret key used to HMAC-sign session cookies."""
|
|
return os.getenv("AUTH_SECRET_KEY") or "websurfer-default-secret-key-please-change-in-prod"
|
|
|
|
|
|
def verify_credentials(username: str, password: str) -> bool:
|
|
"""Verifies user credentials using constant-time string comparison."""
|
|
expected_user = os.getenv("AUTH_USERNAME", "")
|
|
expected_pass = os.getenv("AUTH_PASSWORD", "")
|
|
|
|
if not expected_user or not expected_pass:
|
|
return False
|
|
|
|
user_match = secrets.compare_digest(username.strip(), expected_user.strip())
|
|
pass_match = secrets.compare_digest(password, expected_pass)
|
|
return user_match and pass_match
|
|
|
|
|
|
def create_session_token(username: str) -> str:
|
|
"""Generates a tamper-proof signed session token (username:timestamp:signature)."""
|
|
timestamp = str(int(time.time()))
|
|
payload = f"{username}:{timestamp}"
|
|
signature = hmac.new(
|
|
get_secret_key().encode("utf-8"),
|
|
payload.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return f"{payload}:{signature}"
|
|
|
|
|
|
def verify_session_token(token: Optional[str]) -> Optional[str]:
|
|
"""Validates the signature and timestamp of a session token."""
|
|
if not token or ":" not in token:
|
|
return None
|
|
|
|
parts = token.split(":")
|
|
if len(parts) != 3:
|
|
return None
|
|
|
|
username, timestamp_str, signature = parts
|
|
|
|
try:
|
|
token_time = int(timestamp_str)
|
|
# Check token expiration
|
|
if time.time() - token_time > SESSION_MAX_AGE_SECONDS or token_time > time.time() + 300:
|
|
return None
|
|
except ValueError:
|
|
return None
|
|
|
|
payload = f"{username}:{timestamp_str}"
|
|
expected_sig = hmac.new(
|
|
get_secret_key().encode("utf-8"),
|
|
payload.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
if not secrets.compare_digest(signature, expected_sig):
|
|
return None
|
|
|
|
return username
|
|
|
|
|
|
async def get_current_user(request: Request) -> Optional[str]:
|
|
"""FastAPI dependency: verifies user session from cookie or Bearer token."""
|
|
if not is_auth_enabled():
|
|
return "anonymous"
|
|
|
|
# 1. Check session cookie
|
|
token = request.cookies.get(SESSION_COOKIE_NAME)
|
|
|
|
# 2. Check Authorization Bearer header as fallback
|
|
if not token:
|
|
auth_header = request.headers.get("Authorization")
|
|
if auth_header and auth_header.startswith("Bearer "):
|
|
token = auth_header[7:].strip()
|
|
|
|
username = verify_session_token(token)
|
|
if not username:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authentication required",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
return username
|
|
|
|
|
|
def verify_websocket(websocket: WebSocket) -> bool:
|
|
"""Validates WebSocket connection session from cookie or query parameter."""
|
|
if not is_auth_enabled():
|
|
return True
|
|
|
|
# Check cookie from handshake request
|
|
token = websocket.cookies.get(SESSION_COOKIE_NAME)
|
|
|
|
# Check query param as alternative (?token=...)
|
|
if not token:
|
|
token = websocket.query_params.get("token")
|
|
|
|
return verify_session_token(token) is not None
|