feat: compose fresh owner facts with explicit binding and evidence

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929

tests/test_owner_facts.py

@@ -0,0 +1,129 @@
+import asyncio
+import time
+from dataclasses import replace
+
+import pytest
+
+from hub_core.security.facts import AccountObservation, TenantObservation, PlatformFacts
+from hub_core.security.identity import AccessFailure
+from test_access_boundary import Owners
+
+
+class Readers:
+    def __init__(self):
+        self.actor = Owners().actor
+        self.account = AccountObservation(self.actor.issuer,self.actor.subject,'owner:platform',
+            True,True,time.time(),'account:1',frozenset({'agent:root'}))
+        self.tenant = TenantObservation('tenant:platform',True,time.time(),'tenant:1')
+        self.calls = []
+
+    async def read(self,**kwargs):
+        self.calls.append(kwargs)
+        return self.tenant if 'identifier' in kwargs else self.account
+
+    def facts(self):
+        return PlatformFacts(accounts=self,tenants=self,account_tenant='owner:platform',
+                             mapping_evidence='review:platform-mapping')
+
+
+def test_live_join_preserves_oldest_observation_and_detects_revocation():
+    reader = Readers()
+    facts = reader.facts()
+    async def run():
+        first = await facts.resolve(reader.actor,'/docs')
+        assert first.checked_at == min(reader.account.checked_at,reader.tenant.checked_at)
+        assert first.root_entitled and first.account_active and first.target_tenant_active
+        assert first.producer_addresses == frozenset({'agent:root'})
+        reader.account = replace(reader.account,root_entitled=False,evidence_id='account:2')
+        second = await facts.resolve(reader.actor,'/docs')
+        assert not second.root_entitled and second.evidence_id != first.evidence_id
+        reader.account = replace(reader.account,account_active=False)
+        reader.tenant = replace(reader.tenant,active=False)
+        third = await facts.resolve(reader.actor,'/docs')
+        assert not third.account_active and not third.actor_tenant_active and not third.target_tenant_active
+    asyncio.run(run())
+    assert len(reader.calls) == 6
+    assert reader.calls[0] == dict(issuer=reader.actor.issuer,subject=reader.actor.subject,tenant='owner:platform')
+
+
+@pytest.mark.parametrize('source,changes',[
+    ('account',{'subject':'other'}),('account',{'issuer':'other'}),
+    ('account',{'tenant':'tenant:platform'}),('tenant',{'identifier':'other'}),
+    ('account',{'account_active':'true'}),('account',{'root_entitled':1}),
+    ('tenant',{'active':None}),('account',{'checked_at':1}),
+    ('tenant',{'checked_at':float('nan')}),('account',{'checked_at':float('inf')}),
+    ('account',{'checked_at':time.time()+100}),('tenant',{'evidence_id':''}),
+    ('account',{'producer_addresses':['agent:root']}),
+])
+def test_malformed_mismatched_or_stale_observation_refuses(source,changes):
+    reader = Readers()
+    setattr(reader,source,replace(getattr(reader,source),**changes))
+    with pytest.raises(AccessFailure,match='owner_facts_unavailable_or_untrusted'):
+        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))
+
+
+def test_owner_failure_has_no_cached_fallback_or_private_error():
+    reader = Readers()
+    facts = reader.facts()
+    asyncio.run(facts.resolve(reader.actor,'/docs'))
+    async def unavailable(**kwargs):
+        raise RuntimeError('private owner details')
+    reader.read = unavailable
+    with pytest.raises(AccessFailure) as error:
+        asyncio.run(facts.resolve(reader.actor,'/docs'))
+    assert error.value.status == 503 and 'private' not in str(error.value)
+
+
+@pytest.mark.parametrize('changes',[{'principal_type':'service'},{'tenant':'tenant:other'}])
+def test_unadmitted_principal_never_calls_owners(changes):
+    reader = Readers()
+    with pytest.raises(AccessFailure) as error:
+        asyncio.run(reader.facts().resolve(replace(reader.actor,**changes),'/docs'))
+    assert error.value.status == 403 and not reader.calls
+
+
+def test_slow_join_does_not_refresh_account_observation(monkeypatch):
+    from types import SimpleNamespace
+    from hub_core.security import facts as module
+    reader = Readers()
+    now = time.time()
+    reader.account = replace(reader.account,checked_at=now-4.5)
+    reader.tenant = replace(reader.tenant,checked_at=now)
+    clock = SimpleNamespace(time=lambda: now)
+    monkeypatch.setattr(module,'time',clock)
+    async def read(**kwargs):
+        if 'identifier' in kwargs:
+            clock.time = lambda: now+1
+            return reader.tenant
+        return reader.account
+    reader.read = read
+    with pytest.raises(AccessFailure):
+        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))
+
+
+@pytest.mark.parametrize('source,changes',[
+    ('account',{'root_entitled':False}),('account',{'account_active':False}),
+    ('tenant',{'active':False}),
+])
+def test_next_authorization_denies_withdrawn_owner_state(source,changes):
+    reader = Readers()
+    owners = Owners()
+    controller = owners.controller()
+    controller.facts = reader.facts()
+    async def run():
+        await controller.authorize('verified-root','hub.read','/docs','first','digest')
+        setattr(reader,source,replace(getattr(reader,source),**changes))
+        with pytest.raises(AccessFailure) as error:
+            await controller.authorize('verified-root','hub.read','/docs','next','digest')
+        assert error.value.status == 403
+        assert len(owners.requests) == 1
+    asyncio.run(run())
+
+
+def test_hung_owner_read_is_bounded():
+    reader = Readers()
+    async def hung(**kwargs):
+        await asyncio.Event().wait()
+    reader.read = hung
+    with pytest.raises(AccessFailure,match='owner_facts_unavailable_or_untrusted'):
+        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))