security: protect bootstrap-admin, sleep on bad login, tighten brute-force limits

- bootstrap_admin() now syncs password/active status from env on every startup
  (rotating the env secret immediately takes effect without DB edit)
- user_toggle: block deactivation of the bootstrap admin account
- login_post: tighten rate limits (5 per IP / 3 per login instead of 10/5)
- login_post: add asyncio.sleep(1.0) delay on failed password attempt
This commit is contained in:
Your Name
2026-07-22 15:38:44 +05:00
parent f9076adf96
commit 80c8d04f5b
+22 -4
View File
@@ -384,21 +384,30 @@ def pkce_challenge(verifier: str) -> str:
async def bootstrap_admin() -> None: async def bootstrap_admin() -> None:
"""Create or sync the bootstrap admin user from env secrets on every startup."""
if not settings.admin_bootstrap_login or not settings.admin_bootstrap_password: if not settings.admin_bootstrap_login or not settings.admin_bootstrap_password:
return return
pool = await get_pool() pool = await get_pool()
exists = await pool.fetchval("SELECT 1 FROM admin_users WHERE login=$1", settings.admin_bootstrap_login) login = settings.admin_bootstrap_login.strip().lower()
exists = await pool.fetchval("SELECT 1 FROM admin_users WHERE login=$1", login)
if exists: if exists:
# Always sync password from env so rotating the secret takes effect immediately.
await pool.execute(
"UPDATE admin_users SET password_hash=$1, is_active=TRUE, updated_at=NOW() WHERE login=$2",
hash_password(settings.admin_bootstrap_password),
login,
)
logger.info("Bootstrap admin password synced from env: {}", login)
return return
await pool.execute( await pool.execute(
""" """
INSERT INTO admin_users(login, password_hash, role, is_active) INSERT INTO admin_users(login, password_hash, role, is_active)
VALUES($1, $2, 'admin', TRUE) VALUES($1, $2, 'admin', TRUE)
""", """,
settings.admin_bootstrap_login.strip().lower(), login,
hash_password(settings.admin_bootstrap_password), hash_password(settings.admin_bootstrap_password),
) )
logger.warning("Bootstrap admin user created: {}", settings.admin_bootstrap_login) logger.warning("Bootstrap admin user created: {}", login)
async def get_current_user(request: Request) -> dict | None: async def get_current_user(request: Request) -> dict | None:
@@ -1907,7 +1916,8 @@ async def login_post(request: Request, login: str = Form(...), password: str = F
) )
or 0 or 0
) )
if ip_failures >= 10 or login_failures >= 5: # Stricter limits: 5 bad attempts per IP or 3 per login in 15 minutes → block.
if ip_failures >= 5 or login_failures >= 3:
await pool.execute( await pool.execute(
"INSERT INTO admin_login_attempts(ip, login, success) VALUES($1, $2, FALSE)", "INSERT INTO admin_login_attempts(ip, login, success) VALUES($1, $2, FALSE)",
ip, ip,
@@ -1930,6 +1940,8 @@ async def login_post(request: Request, login: str = Form(...), password: str = F
login_value, login_value,
) )
await audit(None, "auth.login_failed", "admin_login", None, {"login": login_value}, request) await audit(None, "auth.login_failed", "admin_login", None, {"login": login_value}, request)
# Constant-time delay to slow brute-force attempts.
await asyncio.sleep(1.0)
return templates.TemplateResponse("login.html", base_context(request, None, error="Неверный логин или пароль")) return templates.TemplateResponse("login.html", base_context(request, None, error="Неверный логин или пароль"))
await pool.execute("INSERT INTO admin_login_attempts(ip, login, success) VALUES($1, $2, TRUE)", ip, login_value) await pool.execute("INSERT INTO admin_login_attempts(ip, login, success) VALUES($1, $2, TRUE)", ip, login_value)
@@ -3855,6 +3867,12 @@ async def user_toggle(request: Request, user_id: int, csrf_token: str = Form(...
if user["role"] != "admin" or int(user["id"]) == user_id: if user["role"] != "admin" or int(user["id"]) == user_id:
return redirect("/users") return redirect("/users")
pool = await get_pool() pool = await get_pool()
# Prevent deactivating the bootstrap admin account (managed via env secrets).
bootstrap_login = settings.admin_bootstrap_login.strip().lower()
target_login = await pool.fetchval("SELECT login FROM admin_users WHERE id=$1", user_id)
if target_login and target_login == bootstrap_login:
logger.warning("Attempt to deactivate bootstrap admin '{}' was blocked", bootstrap_login)
return redirect("/users")
await pool.execute("UPDATE admin_users SET is_active=NOT is_active, updated_at=NOW() WHERE id=$1", user_id) await pool.execute("UPDATE admin_users SET is_active=NOT is_active, updated_at=NOW() WHERE id=$1", user_id)
await audit(user["id"], "user.toggle", "user", user_id) await audit(user["id"], "user.toggle", "user", user_id)
return redirect("/users") return redirect("/users")