tegwick
4153ad7305
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929
141 lines
6.8 KiB
Python
141 lines
6.8 KiB
Python
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())
|