feat(auth): sessions end when they should; logins are throttled; Secure behind HTTPS (#5035)
#5035 "Sessions: lifetime applied, role and is_active re-read per request, logout is a POST, failed logins throttled". - steward/core/proxy.py: trusted_value / is_https / client_ip read X-Forwarded-Proto and -For only as far as the trusted proxy hops (Settings → General, default 1, 0 = trust none). - The session cookie is Secure, and HSTS is sent, only on requests that arrived over HTTPS by that rule; plain-HTTP and LAN use are unchanged and nothing redirects. - A login lasts Settings → General's session lifetime from when it began (session.permanent + a login_at stamp checked every request — the cookie's own age slides on every response). A deleted or deactivated user is logged out on their next request; a role change applies at once. - Logout is a POST (a form in the sidebar); GET /logout is gone. - Failed logins pause an address after login.max_failures within login.lockout_minutes (Settings → General; defaults 5 in 15 min), keyed by address so nobody can lock a user out by name. Failures are audited as auth.login_failed with the address. - to_access_cfg projects the new settings into app.config at startup and on save, beside the existing projections. - Integration tests seed real users and a login_at, since every request now re-reads the session's user. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,116 @@
|
||||
"""Sessions: HTTPS-only flags follow the trusted proxy, logins are throttled
|
||||
by address, logout is a POST, and a session speaks for the user as they are."""
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from quart import session
|
||||
|
||||
from steward.auth.middleware import refresh_session_user
|
||||
from steward.auth.throttle import LoginThrottle, login_throttle
|
||||
from steward.core.proxy import trusted_value
|
||||
from steward.models.users import UserRole
|
||||
|
||||
|
||||
@pytest.mark.parametrize("values,hops,want", [
|
||||
(["https"], 0, None),
|
||||
(["https"], 1, "https"),
|
||||
(["http, https"], 1, "https"),
|
||||
(["evil", "https"], 1, "https"),
|
||||
(["https, http"], 2, "https"),
|
||||
(["https"], 2, None),
|
||||
([], 1, None),
|
||||
])
|
||||
def test_trusted_value_reads_from_the_right(values, hops, want):
|
||||
assert trusted_value(values, hops) == want
|
||||
|
||||
|
||||
@pytest.mark.parametrize("hops,headers,secure", [
|
||||
(0, {"X-Forwarded-Proto": "https"}, False),
|
||||
(1, {"X-Forwarded-Proto": "https"}, True),
|
||||
(1, {}, False),
|
||||
])
|
||||
async def test_secure_cookie_and_hsts_follow_the_trusted_proxy(app, client, hops, headers, secure):
|
||||
app.config["TRUSTED_PROXY_HOPS"] = hops
|
||||
response = await client.get("/login", headers=headers)
|
||||
cookies = [c for c in response.headers.getlist("Set-Cookie") if c.startswith("session=")]
|
||||
assert cookies, "the login page sets a session (its CSRF token)"
|
||||
assert ("; Secure" in cookies[0]) is secure
|
||||
assert ("Strict-Transport-Security" in response.headers) is secure
|
||||
|
||||
|
||||
def test_throttle_counts_failures_within_the_window():
|
||||
t = LoginThrottle()
|
||||
for i in range(3):
|
||||
assert t.retry_after("1.2.3.4", 3, 60, now=100 + i) == 0
|
||||
t.failed("1.2.3.4", now=100 + i)
|
||||
assert t.retry_after("1.2.3.4", 3, 60, now=103) > 0
|
||||
assert t.retry_after("5.6.7.8", 3, 60, now=103) == 0
|
||||
assert t.retry_after("1.2.3.4", 3, 60, now=161) == 0 # the first failure aged out
|
||||
t.failed("1.2.3.4", now=161)
|
||||
t.succeeded("1.2.3.4")
|
||||
assert t.retry_after("1.2.3.4", 3, 60, now=162) == 0
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_throttle():
|
||||
login_throttle._failures.clear()
|
||||
yield login_throttle
|
||||
login_throttle._failures.clear()
|
||||
|
||||
|
||||
async def test_login_is_refused_after_repeated_failures(app, client, fresh_throttle):
|
||||
app.config["LOGIN_MAX_FAILURES"] = 2
|
||||
async with client.session_transaction() as sess:
|
||||
sess["_csrf"] = "t"
|
||||
form = {"username": "admin", "password": "wrong"}
|
||||
with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
|
||||
patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=None)):
|
||||
for _ in range(2):
|
||||
assert (await client.post("/login", form=form, headers={"X-CSRF-Token": "t"})).status_code == 400
|
||||
refused = await client.post("/login", form=form, headers={"X-CSRF-Token": "t"})
|
||||
assert refused.status_code == 429
|
||||
assert "Too many failed logins" in await refused.get_data(as_text=True)
|
||||
|
||||
|
||||
async def test_logout_is_a_post(client):
|
||||
async with client.session_transaction() as sess:
|
||||
sess.update(user_id="u", user_role="admin", username="a", _csrf="t")
|
||||
assert (await client.get("/logout")).status_code == 405
|
||||
response = await client.post("/logout", headers={"X-CSRF-Token": "t"})
|
||||
assert response.status_code == 302
|
||||
async with client.session_transaction() as sess:
|
||||
assert "user_id" not in sess
|
||||
|
||||
|
||||
class _DB:
|
||||
def __init__(self, user):
|
||||
self.user = user
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
async def get(self, model, key):
|
||||
return self.user
|
||||
|
||||
|
||||
@pytest.mark.parametrize("user,login_age_s,keeps,role", [
|
||||
(SimpleNamespace(is_active=True, role=UserRole.viewer), 60, True, "viewer"), # demoted
|
||||
(SimpleNamespace(is_active=False, role=UserRole.admin), 60, False, None), # deactivated
|
||||
(None, 60, False, None), # deleted
|
||||
(SimpleNamespace(is_active=True, role=UserRole.admin), 9 * 3600, False, None), # expired
|
||||
])
|
||||
async def test_session_speaks_for_the_user_as_they_are(app, user, login_age_s, keeps, role):
|
||||
app.db_sessionmaker = lambda: _DB(user)
|
||||
app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=8)
|
||||
async with app.test_request_context("/"):
|
||||
session.update(user_id="u", user_role="admin", username="a",
|
||||
login_at=time.time() - login_age_s)
|
||||
await refresh_session_user()
|
||||
assert ("user_id" in session) is keeps
|
||||
assert session.get("user_role") == role
|
||||
Reference in New Issue
Block a user