refs #709: replace portal query token with HttpOnly auth
This commit is contained in:
devmrko
217
poc4_active_source_20260714/tests/test_auth_gateway.py
Normal file
217
poc4_active_source_20260714/tests/test_auth_gateway.py
Normal file
@@ -0,0 +1,217 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user