from __future__ import annotations import hashlib import http.client import os from pathlib import Path import sys import threading import unittest from urllib.parse import urlencode ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.agent_console.auth_gateway import ( # noqa: E402 AuthConfig, LoginAttemptLimiter, SESSION_COOKIE_NAME, SessionTokenCodec, build_handler, clear_session_cookie_header, password_matches, session_cookie_header, ) class AuthGatewayTest(unittest.TestCase): def setUp(self) -> None: self.secret = "s" * 48 self.codec = SessionTokenCodec(self.secret) def test_session_token_round_trip_and_tamper_rejection(self) -> None: token = self.codec.issue("demo-admin", 3600, now=1_000) session = self.codec.verify(token, "demo-admin", now=1_001) self.assertIsNotNone(session) self.assertEqual("demo-admin", session.username) self.assertEqual(4_600, session.expires_at) self.assertIsNone(self.codec.verify(token + "x", "demo-admin", now=1_001)) self.assertIsNone(self.codec.verify(token, "other-user", now=1_001)) def test_expired_session_token_is_rejected(self) -> None: token = self.codec.issue("demo-admin", 300, now=1_000) self.assertIsNone(self.codec.verify(token, "demo-admin", now=1_300)) def test_remember_cookie_has_required_security_attributes(self) -> None: header = session_cookie_header("signed-value", 604_800) self.assertIn(f"{SESSION_COOKIE_NAME}=signed-value", header) self.assertIn("Path=/", header) self.assertIn("Secure", header) self.assertIn("HttpOnly", header) self.assertIn("SameSite=Lax", header) self.assertIn("Max-Age=604800", header) self.assertNotIn("Domain=", header) def test_session_cookie_omits_persistent_max_age(self) -> None: header = session_cookie_header("signed-value", None) self.assertNotIn("Max-Age", header) self.assertIn("HttpOnly", header) def test_logout_cookie_expires_immediately(self) -> None: header = clear_session_cookie_header() self.assertIn("Max-Age=0", header) self.assertIn("Secure", header) self.assertIn("HttpOnly", header) def test_pbkdf2_password_verification(self) -> None: salt = bytes.fromhex("00112233445566778899aabbccddeeff") expected = hashlib.pbkdf2_hmac( "sha256", b"correct-password", salt, 200_000 ).hex() encoded = f"pbkdf2_sha256$200000${salt.hex()}${expected}" self.assertTrue(password_matches("correct-password", encoded)) self.assertFalse(password_matches("wrong-password", encoded)) def test_rate_limiter_blocks_only_after_threshold(self) -> None: limiter = LoginAttemptLimiter(maximum_failures=2, window_seconds=10) limiter.record_failure("client", now=1) self.assertFalse(limiter.blocked("client", now=2)) limiter.record_failure("client", now=3) self.assertTrue(limiter.blocked("client", now=4)) self.assertFalse(limiter.blocked("client", now=20)) def test_environment_config_requires_new_cookie_secret(self) -> None: previous = dict(os.environ) try: os.environ["POC4_LOGIN_USER"] = "demo-admin" os.environ["POC4_LOGIN_PASSWORD_PBKDF2"] = "encoded" os.environ.pop("POC4_LOGIN_COOKIE_SECRET", None) with self.assertRaisesRegex(ValueError, "COOKIE_SECRET"): AuthConfig.from_environment() finally: os.environ.clear() os.environ.update(previous) def test_http_login_check_and_logout_flow_never_uses_url_token(self) -> None: salt = bytes.fromhex("00112233445566778899aabbccddeeff") expected = hashlib.pbkdf2_hmac( "sha256", b"correct-password", salt, 200_000 ).hex() config = AuthConfig( username="demo-admin", password_pbkdf2=( f"pbkdf2_sha256$200000${salt.hex()}${expected}" ), cookie_secret=self.secret, port=8621, ) from http.server import ThreadingHTTPServer server = ThreadingHTTPServer(("127.0.0.1", 0), build_handler(config)) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() connection = http.client.HTTPConnection( "127.0.0.1", server.server_address[1], timeout=3 ) try: connection.request("GET", "/auth/login") login_page = connection.getresponse() body = login_page.read().decode("utf-8") self.assertEqual(200, login_page.status) csrf_header = next( value for name, value in login_page.getheaders() if name.lower() == "set-cookie" and value.startswith("__Host-HMM_LOGIN_CSRF=") ) csrf_value = csrf_header.split("=", 1)[1].split(";", 1)[0] self.assertIn( f'name="csrf" value="{csrf_value}"', body, ) payload = urlencode( { "csrf": csrf_value, "username": "demo-admin", "password": "correct-password", "remember": "yes", } ) connection.request( "POST", "/auth/login", body=payload, headers={ "Content-Type": "application/x-www-form-urlencoded", "Cookie": f"__Host-HMM_LOGIN_CSRF={csrf_value}", }, ) logged_in = connection.getresponse() logged_in.read() self.assertEqual(303, logged_in.status) self.assertEqual("/", logged_in.getheader("Location")) self.assertNotRegex(logged_in.getheader("Location"), r"token|remember") session_header = next( value for name, value in logged_in.getheaders() if name.lower() == "set-cookie" and value.startswith(f"{SESSION_COOKIE_NAME}=") ) session_value = session_header.split("=", 1)[1].split(";", 1)[0] self.assertIn("Secure", session_header) self.assertIn("HttpOnly", session_header) self.assertIn("SameSite=Lax", session_header) connection.request( "GET", "/auth/check", headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}"}, ) check = connection.getresponse() check.read() self.assertEqual(204, check.status) self.assertEqual("demo-admin", check.getheader("X-Auth-User")) connection.request( "GET", "/auth/check", headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}x"}, ) tampered = connection.getresponse() tampered.read() self.assertEqual(401, tampered.status) connection.request( "GET", "/auth/logout", headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}"}, ) logout = connection.getresponse() logout.read() self.assertEqual(303, logout.status) self.assertEqual("/auth/login", logout.getheader("Location")) self.assertTrue( any( name.lower() == "set-cookie" and "Max-Age=0" in value for name, value in logout.getheaders() ) ) finally: connection.close() server.shutdown() server.server_close() thread.join(timeout=3) if __name__ == "__main__": unittest.main()