import hashlib import os import re from app import create_app from passwords import hash_password from tests.support import ApiTestCase, app, db_execute class AuthTests(ApiTestCase): def session_token(self, client=None) -> str: return ( (client or self.client) .get_cookie("sid", domain="localhost", path="/api") .value ) def test_register_logs_in_and_me_returns_user(self): response = self.register("Alice") self.assertEqual(response.get_json(), {"id": 1, "username": "Alice"}) me = self.call("GET", "/api/auth/me") self.assertEqual(me.status_code, 200) self.assertEqual(me.get_json(), {"id": 1, "username": "Alice"}) def test_session_cookie_flags(self): cookie = self.register().headers["Set-Cookie"] for flag in ("HttpOnly", "Secure", "SameSite=Lax", "Path=/api"): self.assertIn(flag, cookie) def test_session_token_is_stored_hashed(self): self.register() digest = hashlib.sha256(self.session_token().encode()).digest() self.assertEqual( db_execute( "SELECT count(*) FROM sessions WHERE token_hash = %s", (digest,) )[0][0], 1, ) def test_duplicate_username_is_case_insensitive(self): self.register("Alice") response = self.call( "POST", "/api/auth/register", {"username": "alice", "password": "another long password"}, ) self.assertEqual(response.status_code, 409) self.assertEqual(response.get_json()["field"], "username") def test_register_validation(self): cases = [ ({"username": "al", "password": "long enough password"}, "username"), ({"username": "al ice", "password": "long enough password"}, "username"), ({"username": "alice", "password": "short"}, "password"), ({"username": "alice", "password": "x" * 1025}, "password"), ({"username": "alice", "password": "\ud800" * 12}, "password"), ( { "username": "alice", "password": "long enough password", "is_admin": True, }, "is_admin", ), ] for payload, field in cases: with self.subTest(payload=payload): response = self.call("POST", "/api/auth/register", payload) self.assertEqual(response.status_code, 400) self.assertEqual(response.get_json()["field"], field) def test_login_is_case_insensitive_on_username(self): self.register("Alice") self.client = app.test_client() response = self.call( "POST", "/api/auth/login", {"username": "ALICE", "password": "correct horse battery"}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.get_json(), {"id": 1, "username": "Alice"}) def test_wrong_password_and_unknown_user_are_indistinguishable(self): self.register() wrong = self.call( "POST", "/api/auth/login", {"username": "alice", "password": "not the password"}, ) unknown = self.call( "POST", "/api/auth/login", {"username": "nobody", "password": "not the password"}, ) self.assertEqual(wrong.status_code, 401) self.assertEqual(unknown.status_code, 401) self.assertEqual(wrong.get_json(), unknown.get_json()) self.assertEqual(wrong.get_json(), {"error": "Invalid username or password"}) def test_login_with_malformed_username_is_401_not_500(self): response = self.call( "POST", "/api/auth/login", {"username": "a\x00b", "password": "whatever password"}, ) self.assertEqual(response.status_code, 401) def test_me_without_session_is_401(self): response = self.call("GET", "/api/auth/me") self.assertEqual(response.status_code, 401) self.assertEqual( response.get_json(), {"error": "Not signed in or session expired; log in again"}, ) def test_logout_revokes_token_server_side(self): self.register() token = self.session_token() self.assertEqual(self.call("POST", "/api/auth/logout").status_code, 204) self.client.set_cookie("sid", token, domain="localhost", path="/api") self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401) def test_idle_session_expires(self): self.register() db_execute("UPDATE sessions SET expires_at = now() - interval '1 second'") self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401) def test_absolute_cap_applies_even_inside_idle_window(self): self.register() db_execute( "UPDATE sessions SET created_at = now() - interval '12 hours 1 second'" ) self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401) def test_activity_slides_idle_expiry(self): self.register() db_execute("UPDATE sessions SET expires_at = now() + interval '1 minute'") self.assertEqual(self.call("GET", "/api/auth/me").status_code, 200) remaining = db_execute( "SELECT expires_at - now() > interval '29 minutes' FROM sessions" )[0][0] self.assertTrue(remaining) def test_old_hash_is_upgraded_on_login(self): self.register() pepper = os.environ["PASSWORD_PEPPER"].encode() old = hash_password("correct horse battery", pepper, iterations=1000) db_execute("UPDATE users SET password_hash = %s", (old,)) response = self.call( "POST", "/api/auth/login", {"username": "alice", "password": "correct horse battery"}, ) self.assertEqual(response.status_code, 200) self.assertTrue( db_execute("SELECT password_hash FROM users")[0][0].startswith( "pbkdf2_sha256$600000$" ) ) class AuthByDefaultTests(ApiTestCase): PUBLIC = frozenset({"health", "auth.register", "auth.login", "auth.logout"}) def test_only_the_expected_endpoints_allow_anonymous_access(self): public = { endpoint for endpoint, view in app.view_functions.items() if getattr(view, "allow_anonymous", False) } self.assertEqual(public, self.PUBLIC) def test_every_other_route_rejects_anonymous_requests(self): for rule in app.url_map.iter_rules(): if rule.endpoint in self.PUBLIC or rule.endpoint == "static": continue path = re.sub(r"<[^>]+>", "1", rule.rule) for method in rule.methods - {"HEAD", "OPTIONS"}: with self.subTest(method=method, path=path): self.assertEqual(self.call(method, path).status_code, 401) def test_a_new_route_requires_login_without_any_decorator(self): fresh = create_app() fresh.add_url_rule("/api/new-thing", "new_thing", lambda: {"ok": True}) response = self.call("GET", "/api/new-thing", client=fresh.test_client()) self.assertEqual(response.status_code, 401)