import asyncio import time from datetime import datetime, timezone from types import SimpleNamespace import httpx import pytest from hub_core.security.account_reader import GRANT, UserEngineAccountReader, normalize_account from hub_core.security.facts import PlatformFacts, TenantObservation from hub_core.security.identity import AccessFailure from hub_core.security import boundary from test_access_boundary import Owners def iso(value): return datetime.fromtimestamp(value,timezone.utc).isoformat() def body(now): return dict(issuer='https://issuer.example',subject='immutable-root',tenant='platform', identity_id='identity:1',user_id='user:1',account_id='account:1',decision_id='decision:1', account_status='active',tenant_account_status='active',observed_at=iso(now),root_grant={ 'subject':{'issuer':'https://issuer.example','subject':'immutable-root'}, 'grant':dict(GRANT),'source':{'record_id':'nk-custody:platform-root','revision':1}, 'observed_at':iso(now),'expires_at':iso(now+900)}) def normalize(value,now): return normalize_account(value,issuer='https://issuer.example',subject='immutable-root', tenant='platform',now=now) def test_current_owner_shape_preserves_time_scope_and_no_grant(): now=float(int(time.time())) value=body(now) value['root_grant']['observed_at']=iso(now-2) value['root_grant']['expires_at']=iso(now+10) result=normalize(value,now) assert result.root_entitled and result.account_active assert result.checked_at == pytest.approx(now-2) assert result.valid_until == pytest.approx(now+10) assert not result.producer_addresses value['root_grant']=None withdrawn=normalize(value,now) assert not withdrawn.root_entitled and withdrawn.valid_until is None assert withdrawn.evidence_id != result.evidence_id @pytest.mark.parametrize('status',['disabled','suspended','deleted','invited','deletion_pending',None]) def test_inactive_or_missing_scoped_account_never_active(status): now=float(int(time.time())); value=body(now); value['tenant_account_status']=status assert not normalize(value,now).account_active @pytest.mark.parametrize('mutate',[ lambda b: b.update(subject='other'), lambda b: b.update(issuer='other'), lambda b: b.update(tenant='other'), lambda b: b.update(decision_id=None), lambda b: b.update(account_status=True), lambda b: b.update(tenant_account_status='unknown'), lambda b: b.update(observed_at=iso(time.time()-6)), lambda b: b.update(observed_at=iso(time.time()+6)), lambda b: b.update(observed_at='2026-10-04T10:00:00'), lambda b: b.pop('root_grant'), lambda b: b['root_grant']['subject'].update(subject='other'), lambda b: b['root_grant']['grant'].update(scope_id='other'), lambda b: b['root_grant']['grant'].update(owner_system='user-engine'), lambda b: b['root_grant']['source'].update(revision=True), lambda b: b['root_grant']['source'].update(revision=0), lambda b: b['root_grant']['source'].update(record_id='other'), lambda b: b['root_grant'].update(observed_at=iso(time.time()-6)), lambda b: b['root_grant'].update(expires_at=iso(time.time()-1)), lambda b: b['root_grant'].update(expires_at=iso(time.time()+901)), lambda b: b['root_grant'].update(revoked_at=iso(time.time())), ]) def test_mismatched_stale_or_incomplete_owner_facts_refuse(mutate): now=float(int(time.time())); value=body(now); mutate(value) with pytest.raises((ValueError,KeyError,TypeError)): normalize(value,now) def test_http_rotates_credential_and_rechecks_withdrawal(tmp_path): token=tmp_path/'workload'; token.write_text('fixture-first') seen=[]; value=body(time.time()) def respond(request): import json assert request.url.path == '/api/v1/platform/account-authority/lookup' assert json.loads(request.content) == dict(issuer=value['issuer'],subject=value['subject'],tenant=value['tenant']) seen.append(request.headers['authorization']) return httpx.Response(200,json=value) async def run(): async with httpx.AsyncClient(transport=httpx.MockTransport(respond),trust_env=False) as client: reader=UserEngineAccountReader(base_url='https://owner.example',token_file=token,client=client) args={key:value[key] for key in ('issuer','subject','tenant')} assert (await reader.read(**args)).root_entitled token.write_text('fixture-second'); value['root_grant']=None assert not (await reader.read(**args)).root_entitled token.unlink() with pytest.raises(AccessFailure): await reader.read(**args) asyncio.run(run()) assert seen == ['Bearer fixture-first','Bearer fixture-second'] @pytest.mark.parametrize('response',[ httpx.Response(302,headers={'Location':'https://other.example'}), httpx.Response(403,text='private details'), httpx.Response(404), httpx.Response(503), httpx.Response(200,content=b'x'*65537), httpx.Response(200,content=b'{"a":1,"a":2}'), httpx.Response(200,json={},headers={'Age':'1'}), ]) def test_http_refuses_redirect_cache_outage_or_invalid_body(tmp_path,response): token=tmp_path/'workload'; token.write_text('fixture') calls=[] def respond(request): calls.append(request); return response async def run(): async with httpx.AsyncClient(transport=httpx.MockTransport(respond),trust_env=False) as client: reader=UserEngineAccountReader(base_url='https://owner.example',token_file=token,client=client) with pytest.raises(AccessFailure,match='account_authority_unavailable_or_untrusted'): await reader.read(issuer='iss',subject='sub',tenant='platform') asyncio.run(run()); assert len(calls)==1 def test_grant_deadline_survives_join_and_blocks_after_audit(monkeypatch): now=float(int(time.time())); value=body(now); value['root_grant']['expires_at']=iso(now+1) account=normalize(value,now) class Accounts: async def read(self,**kwargs): return account class Tenants: async def read(self,**kwargs): return TenantObservation('tenant:platform',True,now,'tenant:1') owners=Owners(); controller=owners.controller() controller.facts=PlatformFacts(accounts=Accounts(),tenants=Tenants(), account_tenant='platform',mapping_evidence='NK-ADR-0017:2') clock=SimpleNamespace(time=lambda:now,monotonic=time.monotonic) monkeypatch.setattr(boundary,'time',clock) original=owners.append async def append(record): await original(record); clock.time=lambda: now+2 owners.append=append async def run(): with pytest.raises(AccessFailure,match='facts_expired_during_authorization'): await controller.authorize('verified-root','hub.read','/docs','correlation','digest') assert controller.readiness_checks()['access_facts']=='unavailable' asyncio.run(run())