278 lines
11 KiB
Python
278 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db.base import AccessBase
|
|
from govoplan_access.backend.db.models import Account, AuthSession, User
|
|
from govoplan_access.backend.security.sessions import authenticate_session_token, hash_session_token
|
|
from govoplan_access.backend.security.passwords import hash_password
|
|
from govoplan_access.backend.api.v1.admin_schemas import AdminSessionItem
|
|
from govoplan_access.backend.api.v1.auth import AccountSessionInfo
|
|
from govoplan_access.backend.api.v1.routes import _require_session_admin_reauthorization
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from fastapi import HTTPException
|
|
from govoplan_access.backend.session_management import (
|
|
MAX_CLIENT_LABEL_LENGTH,
|
|
list_account_sessions,
|
|
revoke_account_session,
|
|
revoke_other_account_sessions,
|
|
)
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables, scope_registry
|
|
|
|
|
|
class SessionManagementTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
create_scope_tables(self.engine)
|
|
AccessBase.metadata.create_all(bind=self.engine)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.session = self.Session()
|
|
self.now = datetime(2026, 8, 19, 20, 0, tzinfo=timezone.utc)
|
|
tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1")
|
|
other_tenant = Tenant(id="tenant-2", slug="tenant-2", name="Tenant 2")
|
|
self.account = Account(
|
|
id="account-1",
|
|
email="person@example.test",
|
|
normalized_email="person@example.test",
|
|
)
|
|
other_account = Account(
|
|
id="account-2",
|
|
email="other@example.test",
|
|
normalized_email="other@example.test",
|
|
)
|
|
user = User(
|
|
id="user-1",
|
|
tenant_id=tenant.id,
|
|
account_id=self.account.id,
|
|
email=self.account.email,
|
|
)
|
|
other_user = User(
|
|
id="user-2",
|
|
tenant_id=tenant.id,
|
|
account_id=other_account.id,
|
|
email=other_account.email,
|
|
)
|
|
self.session.add_all((tenant, other_tenant, self.account, other_account, user, other_user))
|
|
self.session.flush()
|
|
self.tokens = {
|
|
"current": "ms_current",
|
|
"other": "ms_other",
|
|
"expired": "ms_expired",
|
|
"revoked": "ms_revoked",
|
|
"other-account": "ms_other_account",
|
|
}
|
|
self.session.add_all(
|
|
(
|
|
self._auth_session("current", "tenant-1", "user-1", "account-1"),
|
|
self._auth_session("other", "tenant-2", "user-1", "account-1"),
|
|
self._auth_session("expired", "tenant-1", "user-1", "account-1", expires=-1),
|
|
self._auth_session("revoked", "tenant-1", "user-1", "account-1", revoked=True),
|
|
self._auth_session("other-account", "tenant-1", "user-2", "account-2"),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
AccessBase.metadata.drop_all(bind=self.engine)
|
|
scope_registry.metadata.drop_all(bind=self.engine)
|
|
self.engine.dispose()
|
|
|
|
def _auth_session(
|
|
self,
|
|
name: str,
|
|
tenant_id: str,
|
|
user_id: str,
|
|
account_id: str,
|
|
*,
|
|
expires: int = 2,
|
|
revoked: bool = False,
|
|
) -> AuthSession:
|
|
return AuthSession(
|
|
id=f"session-{name}",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
account_id=account_id,
|
|
token_hash=hash_session_token(self.tokens[name]),
|
|
expires_at=self.now + timedelta(hours=expires),
|
|
last_seen_at=self.now - timedelta(minutes=5),
|
|
revoked_at=self.now - timedelta(minutes=1) if revoked else None,
|
|
user_agent="Browser " + ("x" * 500),
|
|
ip_address="192.0.2.55",
|
|
)
|
|
|
|
def test_listing_is_account_scoped_bounded_and_redacted(self) -> None:
|
|
sensitive = {
|
|
"token",
|
|
"token_hash",
|
|
"csrf_token_hash",
|
|
"cookie",
|
|
"ip_address",
|
|
}
|
|
self.assertTrue(sensitive.isdisjoint(AccountSessionInfo.model_fields))
|
|
self.assertTrue(sensitive.isdisjoint(AdminSessionItem.model_fields))
|
|
active = list_account_sessions(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
current_session_id="session-current",
|
|
now=self.now,
|
|
)
|
|
self.assertEqual({"session-current", "session-other"}, {item.id for item in active})
|
|
self.assertTrue(next(item for item in active if item.id == "session-current").current)
|
|
self.assertTrue(all(len(item.client or "") <= MAX_CLIENT_LABEL_LENGTH for item in active))
|
|
self.assertNotIn("192.0.2.55", repr(active))
|
|
self.assertNotIn("token_hash", repr(active))
|
|
|
|
all_states = list_account_sessions(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
current_session_id="session-current",
|
|
include_inactive=True,
|
|
now=self.now,
|
|
)
|
|
self.assertEqual(
|
|
{"active", "expired", "revoked"},
|
|
{item.status for item in all_states},
|
|
)
|
|
tenant_only = list_account_sessions(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
current_session_id="session-current",
|
|
tenant_id="tenant-1",
|
|
include_inactive=True,
|
|
now=self.now,
|
|
)
|
|
self.assertNotIn("session-other", {item.id for item in tenant_only})
|
|
|
|
def test_single_revocation_is_idempotent_and_effective_on_next_request(self) -> None:
|
|
item, changed = revoke_account_session(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
session_id="session-other",
|
|
protected_session_id="session-current",
|
|
now=self.now,
|
|
)
|
|
self.assertTrue(changed)
|
|
self.session.commit()
|
|
self.assertIsNotNone(item)
|
|
self.assertIsNone(
|
|
authenticate_session_token(self.session, self.tokens["other"])
|
|
)
|
|
|
|
repeated, changed = revoke_account_session(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
session_id="session-other",
|
|
protected_session_id="session-current",
|
|
now=self.now,
|
|
)
|
|
self.assertIs(item, repeated)
|
|
self.assertFalse(changed)
|
|
hidden, changed = revoke_account_session(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
session_id="session-other-account",
|
|
now=self.now,
|
|
)
|
|
self.assertIsNone(hidden)
|
|
self.assertFalse(changed)
|
|
|
|
def test_listing_applies_activity_filter_and_limit_before_loading_history(self) -> None:
|
|
statements: list[str] = []
|
|
|
|
def capture_query(connection, cursor, statement, parameters, context, executemany):
|
|
if statement.lstrip().upper().startswith("SELECT") and "access_auth_sessions" in statement:
|
|
statements.append(statement)
|
|
|
|
event.listen(self.engine, "before_cursor_execute", capture_query)
|
|
try:
|
|
for include_inactive in (False, True):
|
|
with self.subTest(include_inactive=include_inactive):
|
|
statements.clear()
|
|
summaries = list_account_sessions(
|
|
self.session,
|
|
account_id="account-1",
|
|
current_session_id="session-current",
|
|
include_inactive=include_inactive,
|
|
limit=1,
|
|
now=self.now,
|
|
)
|
|
self.assertEqual(1, len(summaries))
|
|
self.assertEqual(1, len(statements))
|
|
self.assertIn("LIMIT", statements[0])
|
|
if not include_inactive:
|
|
self.assertEqual("active", summaries[0].status)
|
|
self.assertIn("revoked_at IS NULL", statements[0])
|
|
self.assertIn("expires_at >", statements[0])
|
|
finally:
|
|
event.remove(self.engine, "before_cursor_execute", capture_query)
|
|
|
|
def test_current_session_is_protected_and_revoke_others_skips_expired(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "current session"):
|
|
revoke_account_session(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
session_id="session-current",
|
|
protected_session_id="session-current",
|
|
now=self.now,
|
|
)
|
|
revoked = revoke_other_account_sessions(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
current_session_id="session-current",
|
|
now=self.now,
|
|
)
|
|
self.assertEqual(("session-other",), revoked)
|
|
self.assertIsNone(self.session.get(AuthSession, "session-current").revoked_at)
|
|
self.assertIsNone(self.session.get(AuthSession, "session-expired").revoked_at)
|
|
self.assertEqual(
|
|
(),
|
|
revoke_other_account_sessions(
|
|
self.session,
|
|
account_id=self.account.id,
|
|
current_session_id="session-current",
|
|
now=self.now,
|
|
),
|
|
)
|
|
|
|
def test_administrative_revocation_requires_session_and_current_password(self) -> None:
|
|
self.account.password_hash = hash_password("correct horse")
|
|
membership = self.session.get(User, "user-1")
|
|
current = self.session.get(AuthSession, "session-current")
|
|
principal_ref = PrincipalRef(
|
|
account_id=self.account.id,
|
|
membership_id=membership.id,
|
|
tenant_id=membership.tenant_id,
|
|
scopes=frozenset({"access:membership:update"}),
|
|
auth_method="session",
|
|
session_id=current.id,
|
|
)
|
|
without_session = ApiPrincipal(
|
|
principal=principal_ref,
|
|
account=self.account,
|
|
user=membership,
|
|
)
|
|
with self.assertRaises(HTTPException) as missing:
|
|
_require_session_admin_reauthorization(without_session, "correct horse")
|
|
self.assertEqual(403, missing.exception.status_code)
|
|
|
|
principal = ApiPrincipal(
|
|
principal=principal_ref,
|
|
account=self.account,
|
|
user=membership,
|
|
auth_session=current,
|
|
)
|
|
with self.assertRaises(HTTPException) as incorrect:
|
|
_require_session_admin_reauthorization(principal, "incorrect")
|
|
self.assertEqual(403, incorrect.exception.status_code)
|
|
_require_session_admin_reauthorization(principal, "correct horse")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|