vee
93 lines
3.2 KiB
Python
Executable File
93 lines
3.2 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from cognitive_agent.auth import (
|
|
AuthConfig,
|
|
SessionCodec,
|
|
SlidingWindowLimiter,
|
|
hash_password,
|
|
is_json_content_type,
|
|
origin_matches_host,
|
|
verify_password,
|
|
)
|
|
|
|
|
|
def auth_config() -> AuthConfig:
|
|
return AuthConfig(
|
|
enabled=True,
|
|
username="demo-user",
|
|
password_hash=hash_password(
|
|
"correct horse battery staple",
|
|
salt=b"0123456789abcdef",
|
|
iterations=100_000,
|
|
),
|
|
session_secret=b"s" * 64,
|
|
session_ttl_seconds=3600,
|
|
login_attempts=3,
|
|
login_window_seconds=60,
|
|
requests_per_minute=10,
|
|
trust_proxy_headers=False,
|
|
)
|
|
|
|
|
|
class AuthenticationUnitTests(unittest.TestCase):
|
|
def test_password_hash_verification(self) -> None:
|
|
encoded = auth_config().password_hash
|
|
|
|
self.assertTrue(verify_password("correct horse battery staple", encoded))
|
|
self.assertFalse(verify_password("wrong password", encoded))
|
|
self.assertNotIn("correct horse battery staple", encoded)
|
|
|
|
def test_signed_session_rejects_tampering_and_expiry(self) -> None:
|
|
codec = SessionCodec(auth_config())
|
|
token = codec.issue("demo-user", now=1000)
|
|
|
|
self.assertEqual(codec.verify(token, now=1001), "demo-user")
|
|
self.assertIsNone(codec.verify(token + "x", now=1001))
|
|
self.assertIsNone(codec.verify(token, now=4600))
|
|
|
|
def test_sliding_window_limiter_returns_retry_after(self) -> None:
|
|
limiter = SlidingWindowLimiter(limit=2, window_seconds=60)
|
|
|
|
limiter.record("client", now=100)
|
|
limiter.record("client", now=110)
|
|
self.assertEqual(limiter.retry_after("client", now=120), 40)
|
|
self.assertEqual(limiter.retry_after("client", now=161), 0)
|
|
|
|
def test_json_content_type_is_exact_but_allows_parameters(self) -> None:
|
|
self.assertTrue(is_json_content_type("application/json"))
|
|
self.assertTrue(is_json_content_type("Application/JSON; charset=utf-8"))
|
|
self.assertFalse(is_json_content_type("text/plain"))
|
|
self.assertFalse(is_json_content_type("application/problem+json"))
|
|
self.assertFalse(is_json_content_type(""))
|
|
|
|
def test_origin_match_uses_scheme_host_and_effective_port(self) -> None:
|
|
self.assertTrue(
|
|
origin_matches_host("http://127.0.0.1:8000", "127.0.0.1:8000", "http")
|
|
)
|
|
self.assertTrue(origin_matches_host("http://example.test", "EXAMPLE.test:80", "http"))
|
|
self.assertTrue(origin_matches_host("http://[::1]:8000", "[::1]:8000", "http"))
|
|
|
|
self.assertFalse(
|
|
origin_matches_host("http://localhost:8000", "127.0.0.1:8000", "http")
|
|
)
|
|
self.assertFalse(
|
|
origin_matches_host("https://127.0.0.1:8000", "127.0.0.1:8000", "http")
|
|
)
|
|
self.assertFalse(
|
|
origin_matches_host("http://127.0.0.1:8001", "127.0.0.1:8000", "http")
|
|
)
|
|
self.assertFalse(origin_matches_host("null", "127.0.0.1:8000", "http"))
|
|
self.assertFalse(
|
|
origin_matches_host(
|
|
"http://127.0.0.1:8000/extra",
|
|
"127.0.0.1:8000",
|
|
"http",
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|