163 lines
5.1 KiB
Python
163 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from host_agent.local_account import LocalAccountStore
|
|
from host_agent.web.auth import SessionManager, attempt_login, change_password
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self, start: datetime) -> None:
|
|
self.current = start
|
|
|
|
def __call__(self) -> datetime:
|
|
return self.current
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.current += timedelta(seconds=seconds)
|
|
|
|
|
|
def test_create_session_then_validate_returns_username() -> None:
|
|
clock = FakeClock(datetime(2026, 1, 1, tzinfo=UTC))
|
|
manager = SessionManager(ttl_seconds=60, now=clock)
|
|
|
|
session_token, csrf_token = manager.create_session("operator")
|
|
state = manager.validate(session_token)
|
|
|
|
assert state is not None
|
|
assert state.username == "operator"
|
|
assert state.csrf_token == csrf_token
|
|
|
|
|
|
def test_validate_unknown_token_returns_none() -> None:
|
|
manager = SessionManager(ttl_seconds=60)
|
|
|
|
assert manager.validate("does-not-exist") is None
|
|
|
|
|
|
def test_validate_after_ttl_elapsed_returns_none() -> None:
|
|
clock = FakeClock(datetime(2026, 1, 1, tzinfo=UTC))
|
|
manager = SessionManager(ttl_seconds=60, now=clock)
|
|
|
|
session_token, _ = manager.create_session("operator")
|
|
clock.advance(61)
|
|
|
|
assert manager.validate(session_token) is None
|
|
|
|
|
|
def test_validate_before_expiry_slides_expiry_forward() -> None:
|
|
clock = FakeClock(datetime(2026, 1, 1, tzinfo=UTC))
|
|
manager = SessionManager(ttl_seconds=60, now=clock)
|
|
|
|
session_token, _ = manager.create_session("operator")
|
|
clock.advance(30)
|
|
first = manager.validate(session_token)
|
|
assert first is not None
|
|
|
|
clock.advance(30)
|
|
second = manager.validate(session_token)
|
|
|
|
assert second is not None
|
|
assert second.expires_at > first.expires_at
|
|
|
|
|
|
def test_validate_csrf_true_for_right_token() -> None:
|
|
manager = SessionManager(ttl_seconds=60)
|
|
session_token, csrf_token = manager.create_session("operator")
|
|
|
|
assert manager.validate_csrf(session_token, csrf_token) is True
|
|
|
|
|
|
def test_validate_csrf_false_for_wrong_token() -> None:
|
|
manager = SessionManager(ttl_seconds=60)
|
|
session_token, _ = manager.create_session("operator")
|
|
|
|
assert manager.validate_csrf(session_token, "wrong-token") is False
|
|
|
|
|
|
def test_validate_csrf_false_for_invalid_session() -> None:
|
|
manager = SessionManager(ttl_seconds=60)
|
|
|
|
assert manager.validate_csrf("does-not-exist", "anything") is False
|
|
|
|
|
|
def test_invalidate_makes_subsequent_validate_return_none() -> None:
|
|
manager = SessionManager(ttl_seconds=60)
|
|
session_token, _ = manager.create_session("operator")
|
|
|
|
manager.invalidate(session_token)
|
|
|
|
assert manager.validate(session_token) is None
|
|
|
|
|
|
def test_attempt_login_true_for_correct_credentials(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
store.create("operator", "correct horse battery staple")
|
|
|
|
assert (
|
|
attempt_login(
|
|
store, username="operator", password="correct horse battery staple"
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_attempt_login_false_for_wrong_password(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
store.create("operator", "correct horse battery staple")
|
|
|
|
assert attempt_login(store, username="operator", password="wrong") is False
|
|
|
|
|
|
def test_attempt_login_false_when_no_account_exists(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
|
|
assert attempt_login(store, username="operator", password="anything") is False
|
|
|
|
|
|
def test_attempt_login_false_for_wrong_username(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
store.create("operator", "correct horse battery staple")
|
|
|
|
assert (
|
|
attempt_login(
|
|
store, username="someone-else", password="correct horse battery staple"
|
|
)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_change_password_succeeds_and_rotates_credential(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
store.create("operator", "old password")
|
|
|
|
assert (
|
|
change_password(
|
|
store, current_password="old password", new_password="new password"
|
|
)
|
|
is True
|
|
)
|
|
assert attempt_login(store, username="operator", password="new password") is True
|
|
assert attempt_login(store, username="operator", password="old password") is False
|
|
|
|
|
|
def test_change_password_fails_with_wrong_current_password(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
store.create("operator", "old password")
|
|
|
|
assert (
|
|
change_password(store, current_password="wrong", new_password="new password")
|
|
is False
|
|
)
|
|
assert attempt_login(store, username="operator", password="old password") is True
|
|
assert attempt_login(store, username="operator", password="new password") is False
|
|
|
|
|
|
def test_change_password_fails_when_no_account_exists(tmp_path) -> None:
|
|
store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
|
|
assert (
|
|
change_password(store, current_password="anything", new_password="new password")
|
|
is False
|
|
)
|