bvandeusenandClaude Opus 5.5
766eef04c5
Run 8736: test_user_model_fields listed the columns without time_zone, and the session-refresh fixture's users had no time_zone attribute. The refresh test now also asserts a zone set on the Account page reaches the session. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
119 lines
4.7 KiB
Python
119 lines
4.7 KiB
Python
"""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", [
|
|
# demoted, and a time zone set on the Account page since (milestone 490)
|
|
(SimpleNamespace(is_active=True, role=UserRole.viewer, time_zone="Europe/London"), 60, True, "viewer"),
|
|
(SimpleNamespace(is_active=False, role=UserRole.admin, time_zone=None), 60, False, None), # deactivated
|
|
(None, 60, False, None), # deleted
|
|
(SimpleNamespace(is_active=True, role=UserRole.admin, time_zone=None), 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
|
|
assert session.get("tz") == (user.time_zone if keeps else None)
|