f81e839ba2
- validate instance names at the host lib and web boundary against
[a-z0-9][a-z0-9_-]{0,63} to prevent path traversal via Server.name
- fail-closed on SECRET_KEY: load_config returns None when env unset,
create_app raises if missing or "dev" outside TESTING
- close login timing oracle by hashing a dummy digest when the user
is not found, equalizing response time
- set SESSION_COOKIE_SECURE outside TESTING
- delete_instance tolerates stop_service and fusermount3 failures so
partially-initialized instances clean up without contract breaks;
drops the is_mount() preflight that violated AGENTS.md
- document claim_next_job's single-process assumption
- clarify emit_step contract via docstring
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
62 lines
2 KiB
Python
62 lines
2 KiB
Python
import time
|
|
|
|
from flask import Blueprint, Response, redirect, render_template, request
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.auth import hash_password, is_safe_next, login_user, logout_user, verify_password
|
|
from l4d2web.db import session_scope
|
|
from l4d2web.models import User
|
|
|
|
|
|
bp = Blueprint("auth", __name__)
|
|
_TIMING_DUMMY_DIGEST = hash_password("__timing_dummy__")
|
|
LOGIN_RATE_LIMIT_WINDOW_SECONDS = 60
|
|
LOGIN_RATE_LIMIT_MAX_ATTEMPTS = 20
|
|
LOGIN_ATTEMPTS_BY_IP: dict[str, list[float]] = {}
|
|
|
|
|
|
def reset_login_rate_limits() -> None:
|
|
LOGIN_ATTEMPTS_BY_IP.clear()
|
|
|
|
|
|
def is_login_rate_limited(remote_addr: str) -> bool:
|
|
now = time.time()
|
|
attempts = LOGIN_ATTEMPTS_BY_IP.setdefault(remote_addr, [])
|
|
cutoff = now - LOGIN_RATE_LIMIT_WINDOW_SECONDS
|
|
attempts[:] = [ts for ts in attempts if ts >= cutoff]
|
|
if len(attempts) >= LOGIN_RATE_LIMIT_MAX_ATTEMPTS:
|
|
return True
|
|
attempts.append(now)
|
|
return False
|
|
|
|
|
|
@bp.get("/login")
|
|
def login_form() -> str:
|
|
next_target = request.args.get("next", "")
|
|
return render_template("login.html", next_target=next_target if is_safe_next(next_target) else "")
|
|
|
|
|
|
@bp.post("/login")
|
|
def login() -> Response:
|
|
remote_addr = request.remote_addr or "unknown"
|
|
if is_login_rate_limited(remote_addr):
|
|
return Response("too many login attempts", status=429)
|
|
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
with session_scope() as db:
|
|
user = db.scalar(select(User).where(User.username == username))
|
|
digest = user.password_digest if user is not None else _TIMING_DUMMY_DIGEST
|
|
password_ok = verify_password(password, digest)
|
|
if user is None or not password_ok:
|
|
return Response("invalid credentials", status=401)
|
|
login_user(user.id)
|
|
LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None)
|
|
next_target = request.form.get("next", "")
|
|
return redirect(next_target if is_safe_next(next_target) else "/dashboard")
|
|
|
|
|
|
@bp.post("/logout")
|
|
def logout() -> Response:
|
|
logout_user()
|
|
return redirect("/login")
|