"""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 add_tenant_user(self, tenant, email): response, _ = invoke( self.app, '/admin/'+tenant+'/users', method='POST', cookie='ue_session=operator', form={ 'csrf_token':'operator-csrf', 'display_name':'Demo', 'primary_email':email, 'role':'user', }, ) self.assertEqual('303 See Other', response['status']) def test_existing_tenant_is_selectable_and_short_name_resolves(self): self.add_tenant_user('tenant:trial:demo-company','one@example.test') self.add_tenant_user('tenant:trial:demo-company','two@example.test') response, body = self.get('/platform') self.assertEqual('200 OK', response['status']) self.assertIn(b'href="/admin/tenant%3Atrial%3Ademo-company">demo-company',body) self.assertEqual(('tenant:trial:demo-company',),self.app.service.store.membership_tenants()) response, _ = self.get('/platform/tenant',query=urlencode({'tenant':' demo-company ','view':'users'})) self.assertEqual('/admin/tenant%3Atrial%3Ademo-company',response['headers']['Location']) denied, _ = self.get('/platform',who='member') self.assertEqual('403 Forbidden',denied['status']) def test_ambiguous_or_unknown_short_names_do_not_guess_a_tenant(self): self.add_tenant_user('tenant:trial:demo-company','one@example.test') self.add_tenant_user('tenant:friendly:demo-company','two@example.test') for name, message in [('demo-company',b'Several tenants'),('unknown',b'No tenant with users')]: response,body=self.get('/platform/tenant',query=urlencode({'tenant':name,'view':'users'})) self.assertEqual('200 OK',response['status']) self.assertNotIn('Location',response['headers']) self.assertIn(message,body) self.assertIn(b'role="alert"',body) response,_=self.get('/platform/tenant',query=urlencode({'tenant':'tenant:trial:demo-company','view':'users'})) self.assertEqual('/admin/tenant%3Atrial%3Ademo-company',response['headers']['Location']) 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()