Spaces:
Sleeping
Sleeping
File size: 4,540 Bytes
990895d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """Implement password login and signed-session lifecycle routes.
Login attempts are rate-limited in memory by the direct client address.
"""
import time
from collections import defaultdict, deque
from threading import Lock
from fastapi import APIRouter, Depends, Request, status
from app.deps import require_login
from app.models import (
ApiEnvelope,
AuthenticatedData,
ChangePasswordRequest,
LoginRequest,
MeData,
StoredConfig,
err,
ok,
)
router = APIRouter(prefix="/api/auth", tags=["auth"])
_attempts: dict[str, deque[float]] = defaultdict(deque)
_attempts_lock = Lock()
def _client_key(request: Request) -> str:
return request.client.host if request.client else "unknown"
def _rate_limited(request: Request) -> bool:
settings = request.app.state.settings
now = time.monotonic()
cutoff = now - settings.login_rate_window_sec
key = _client_key(request)
with _attempts_lock:
attempts = _attempts[key]
while attempts and attempts[0] <= cutoff:
attempts.popleft()
if len(attempts) >= settings.login_rate_limit:
return True
attempts.append(now)
return False
def _clear_attempts(request: Request) -> None:
with _attempts_lock:
_attempts.pop(_client_key(request), None)
def _set_session(request: Request, config: StoredConfig) -> None:
max_age = request.app.state.settings.session_max_age_sec
request.session.clear()
request.session.update(
{
"auth": True,
"sv": config.session_version,
"exp": int(time.time()) + max_age,
}
)
def _session_is_current(request: Request, config: StoredConfig | None) -> bool:
if config is None:
return False
session = request.session
expiry = session.get("exp")
return (
session.get("auth") is True
and session.get("sv") == config.session_version
and isinstance(expiry, (int, float))
and expiry >= time.time()
)
@router.post("/login", response_model=ApiEnvelope)
def login(request: Request, body: LoginRequest) -> object:
"""Authenticate and establish a signed cookie session."""
config = request.app.state.config_store.load()
if config is None:
return err(
"setup_required",
"Application setup is required",
status.HTTP_503_SERVICE_UNAVAILABLE,
)
if _rate_limited(request):
return err(
"rate_limited",
"Too many login attempts",
status.HTTP_429_TOO_MANY_REQUESTS,
)
if not request.app.state.config_store.password_matches(body.password):
return err(
"unauthorized",
"Invalid credentials",
status.HTTP_401_UNAUTHORIZED,
)
_clear_attempts(request)
_set_session(request, config)
return ok(AuthenticatedData(authenticated=True))
@router.post("/logout", response_model=ApiEnvelope)
def logout(
request: Request,
_config: StoredConfig = Depends(require_login),
) -> dict[str, object]:
"""Clear the current signed session."""
request.session.clear()
return ok(AuthenticatedData(authenticated=False))
@router.get("/me", response_model=ApiEnvelope)
def me(request: Request) -> dict[str, object]:
"""Return public app identity and current session state."""
config = request.app.state.config_store.load()
authenticated = _session_is_current(request, config)
if not authenticated:
request.session.clear()
return ok(
MeData(
authenticated=authenticated,
app_name=request.app.state.settings.app_name,
)
)
@router.post("/change-password", response_model=ApiEnvelope)
def change_password(
request: Request,
body: ChangePasswordRequest,
_config: StoredConfig = Depends(require_login),
) -> object:
"""Rotate the shared password and invalidate older sessions."""
changed = request.app.state.config_store.change_password(
body.old_password,
body.new_password,
)
if not changed:
return err(
"unauthorized",
"Invalid credentials",
status.HTTP_401_UNAUTHORIZED,
)
config = request.app.state.config_store.load()
if config is None:
return err(
"internal",
"Configuration unavailable",
status.HTTP_500_INTERNAL_SERVER_ERROR,
)
_set_session(request, config)
return ok(AuthenticatedData(authenticated=True))
|