user-engine/tests/test_portal_navigation.py
tegwick 655dce7165
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Container Image / build-and-push (push) Successful in 52s
fix: expose operator navigation and protected portal logout
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
2026-09-12 00:37:17 +02:00

104 lines
5.6 KiB
Python

"""Browser navigation and logout must preserve the authenticated authority boundary."""
import unittest
from urllib.parse import urlencode
from test_web import FakeTenantManagement, invoke
from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.oidc import BrowserSession, OIDCClient
from user_engine.service import UserEngineService
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
from user_engine.web import PortalApplication
class PortalNavigationTests(unittest.TestCase):
def setUp(self):
store = InMemoryUserEngineStore()
store.migrate()
self.app = PortalApplication(
UserEngineService(store=store, identity_adapter=FixtureIdentityClaimsAdapter(),
authorization=LocalAuthorizationCheckPort()),
trusted_proxy_secret='synthetic-marker-with-adequate-length', login_url='https://kc.example/login',
tenant_management=FakeTenantManagement(),
)
self.oidc = OIDCClient(issuer='https://kc.example', client_id='portal',
redirect_uri='https://users.example/oidc/callback', audience='portal')
self.app.oidc_client = self.oidc
for key, tenant, roles in [
('operator', 'tenant:platform:root', ['platform-operator']),
('member', 'tenant:trial:demo-company', ['user']),
]:
claims = human_actor_claims(subject=key, tenant=tenant)
claims['roles'] = roles
self.oidc.sessions[key] = BrowserSession(claims, 9999999999, key+'-csrf')
def get(self, path, who='operator', query=''):
return invoke(self.app, path, cookie='ue_session='+who, query=query)
def test_operator_can_reach_administration_without_personal_membership(self):
for path in ['/', '/onboarding', '/platform', '/admin/tenant:trial:demo-company']:
response, body = self.get(path)
self.assertEqual('200 OK', response['status'], path)
self.assertIn(b'href="/platform"', body)
self.assertIn(b'action="/logout"', body)
self.assertIn(b'value="operator-csrf"', body)
_, body = self.get('/onboarding')
self.assertIn(b'no personal tenant memberships', body)
self.assertIn(b'platform operator role', body)
self.assertFalse(self.app.service.store.memberships_for_tenant('tenant:trial:demo-company'))
def test_existing_tenant_user_navigation_preserves_authority(self):
query=urlencode({'tenant':'tenant:trial:demo-company','view':'users'})
response, _ = self.get('/platform/tenant', query=query)
self.assertEqual('/admin/tenant%3Atrial%3Ademo-company',response['headers']['Location'])
denied, _ = self.get('/platform/tenant',who='member',query=query)
self.assertEqual('403 Forbidden',denied['status'])
invalid, _ = self.get('/platform/tenant',query=urlencode({'tenant':'tenant:trial:demo-company','view':'https://evil.test'}))
self.assertEqual('400 Bad Request',invalid['status'])
def test_navigation_does_not_leak_between_operator_member_and_anonymous(self):
self.get('/platform')
_, member = self.get('/onboarding',who='member')
self.assertNotIn(b'href="/platform"',member)
self.assertNotIn(b'operator-csrf',member)
self.assertIn(b'value="member-csrf"',member)
self.assertIn(b'No tenant memberships yet.',member)
_, anonymous = invoke(self.app,'/')
self.assertNotIn(b'action="/logout"',anonymous)
self.assertNotIn(b'href="/platform"',anonymous)
self.assertNotIn(b'member-csrf',anonymous)
def test_get_logout_only_confirms_and_bad_csrf_does_not_end_session(self):
response, body = self.get('/logout')
self.assertEqual('200 OK',response['status'])
self.assertIn(b'Log out of this portal?',body)
self.assertIsNotNone(self.oidc.claims('operator'))
for token in ['', 'wrong', 'member-csrf']:
response,_=invoke(self.app,'/logout',method='POST',cookie='ue_session=operator',form={'csrf_token':token})
self.assertEqual('403 Forbidden',response['status'])
self.assertIsNotNone(self.oidc.claims('operator'))
def test_logout_invalidates_only_the_current_session_and_deletes_cookie(self):
response,_=invoke(self.app,'/logout',method='POST',cookie='ue_session=operator',form={'csrf_token':'operator-csrf'})
self.assertEqual('303 See Other',response['status'])
self.assertEqual('/logged-out',response['headers']['Location'])
cookie=response['headers']['Set-Cookie']
for attribute in ['ue_session=;', 'Path=/', 'Secure', 'HttpOnly', 'SameSite=Lax', 'Max-Age=0']:
self.assertIn(attribute,cookie)
self.assertIsNone(self.oidc.claims('operator'))
self.assertIsNotNone(self.oidc.claims('member'))
denied,_=self.get('/platform')
self.assertEqual('403 Forbidden',denied['status'])
page,body=invoke(self.app,'/logged-out')
self.assertEqual('200 OK',page['status'])
self.assertIn(b'shared NetKingdom sign-in may still be active',body)
self.assertNotIn(b'action="/logout"',body)
def test_expired_session_logout_clears_stale_cookie(self):
self.oidc.sessions['operator'].expires_at=0
response,_=invoke(self.app,'/logout',method='POST',cookie='ue_session=operator',form={})
self.assertEqual('303 See Other',response['status'])
self.assertIn('Max-Age=0',response['headers']['Set-Cookie'])
self.assertNotIn('operator',self.oidc.sessions)
if __name__=='__main__':
unittest.main()