"""Offline production-image auth gate; only unique disposable resources are mutated.""" import json import os import secrets import subprocess from pathlib import Path import sys import time from uuid import uuid4 sys.path.insert(0,str(Path(__file__).resolve().parents[1])) import httpx from src.config import config from src.services.credentials import token_repository from src.services.task_contract import accepted_task_id from src.lib.postgresql._provisioning import borrow,get_system_pool from src.lib.tracing.runtime import default_health_probes from src.lib.tracing.tracker import tracer def check(condition,message): if not condition:raise RuntimeError(message) def qualify(): suffix=uuid4().hex[:12] names=['auth-qa-'+suffix+'-a','auth-qa-'+suffix+'-b'] admin={'Authorization':'Bearer '+config.system_pat} issued=[];created=[];results={} base=os.getenv('QUALIFY_API_URL','http://127.0.0.1:7860') # This models the HTTP hop *after* HF TLS termination, including its host # and protocol headers. Secure-cookie replay stays entirely inside this # disposable, network-isolated image; no browser TLS checks are disabled. with httpx.Client(base_url=base,headers={'Host':'auth-qualification.local','X-Forwarded-Proto':'https'},timeout=120,trust_env=False,follow_redirects=True) as api: before=api.get('/system/brains-list',headers=admin);before.raise_for_status() identities={(b['name_key'],b['id']) for b in before.json()} check(not any(name in {b[0] for b in identities} for name in names),'Disposable namespace collision') try: for name in names: response=api.post('/system/workspaces',headers=admin,json={'slug':name,'display_name':'Auth qualification'}) check(response.status_code==201,'Workspace creation failed');created.append(name) check('pat' not in response.json(),'Workspace exposes PAT') brains=api.get('/system/brains-list',headers=admin).json() check(all('pat' not in b for b in brains),'Brain listing exposes PAT') results['credential_free_resources']='passed' def issue(permissions): response=api.post('/system/access-tokens',headers=admin,json={'brain_id':names[0],'label':'qualification','permissions':permissions}) check(response.status_code==201,'Explicit token creation failed') value=response.json();issued.append(value['id']);return value scoped=issue(['memory:read','memory:write']);read_only=issue(['memory:read']) headers={'Authorization':'Bearer '+scoped['token']} own='/brains/'+names[0];other='/brains/'+names[1] for credential in (admin,headers): for path in ('/retrieve/entities',own+'/api/retrieve/entities',own+'/agent/capabilities'): response=api.get(path,headers={**credential,'X-Brain-ID':names[0]}) check(response.status_code==200,'HTTP/Agent authentication parity failed') check(api.get(other+'/agent/capabilities',headers=headers).status_code==403,'Cross-workspace scope accepted') check(api.get(own+'/agent/capabilities',headers={'Authorization':'Bearer bpat_unknown'}).status_code==401,'Unknown token accepted') check(api.get(own+'/agent/capabilities').status_code==401,'Missing token accepted') results['http_agent_auth_parity']='passed' listing=api.get('/system/access-tokens',headers=admin).json()['tokens'] check(all('token' not in item and 'token_hash' not in item for item in listing),'Credential listing exposes secrets') with borrow(get_system_pool()) as conn,conn.cursor() as cursor: cursor.execute('SELECT token_hash FROM access_tokens WHERE id=%s',(scoped['id'],)) stored=cursor.fetchone()[0] check(stored != scoped['token'] and len(stored)==64,'Token was not stored as a hash') cursor.execute('SELECT pat,document FROM data_brains WHERE name_key=ANY(%s)',(names,)) check(all(not pat and 'pat' not in doc for pat,doc in cursor.fetchall()),'Resource identity persisted a credential') results['hash_at_rest_metadata_only']='passed' # Emulate an existing legacy row only in our disposable brain. Then # exercise the shipped atomic migration and prove the secret is gone. legacy_secret=secrets.token_urlsafe(36) with borrow(get_system_pool()) as conn,conn.cursor() as cursor: cursor.execute("UPDATE data_brains SET pat=%s,document=jsonb_set(document,'{pat}',to_jsonb(%s::text)) WHERE name_key=%s",(legacy_secret,legacy_secret,names[0])) conn.commit() legacy_headers={'Authorization':'Bearer '+legacy_secret} check(api.get(own+'/agent/capabilities',headers=legacy_headers).status_code==200,'Legacy scoped credential rejected') check(api.get(other+'/agent/capabilities',headers=legacy_headers).status_code==403,'Legacy credential escaped scope') migration=subprocess.run([sys.executable,'-m','scripts.migrate_legacy_credentials','--migrate',names[0]],capture_output=True,text=True) check(migration.returncode==0,'Atomic legacy migration failed') check(legacy_secret not in migration.stdout+ migration.stderr,'Migration printed a credential') check(api.get(own+'/agent/capabilities',headers=legacy_headers).status_code==200,'Migrated legacy credential rejected') with borrow(get_system_pool()) as conn,conn.cursor() as cursor: cursor.execute('SELECT pat,document FROM data_brains WHERE name_key=%s',(names[0],)) pat,document=cursor.fetchone() check(not pat and 'pat' not in document,'Migration retained plaintext') for row in api.get('/system/access-tokens',headers=admin).json()['tokens']: if row['brain_id']==names[0] and row['legacy']: issued.append(row['id']) check(api.delete('/system/access-tokens/'+row['id'],headers=admin).status_code==200,'Legacy rotation revocation failed') check(api.get(own+'/agent/capabilities',headers=legacy_headers).status_code==401,'Revoked migrated credential accepted') results['legacy_scope_atomic_migration_rotation']='passed' for credential,is_system in ((admin,True),(headers,False)): login=api.post('/auth/session',json={'token':credential['Authorization'][7:]}) check(login.status_code==200,'Console session creation failed') cookie=login.headers.get('set-cookie','') check(all(flag in cookie for flag in ['HttpOnly','Secure','SameSite=strict']),'Session cookie flags invalid') check('token' not in {key for key in login.json() if key!='csrf_token'},'Console echoed credential') sid=login.cookies.get('__Host-brainapi-session') browser={'Cookie':'__Host-brainapi-session='+sid} me=api.get('/auth/me',headers=browser);check(me.status_code==200,'Console session resolution failed') check(me.json()['brain_id']==(None if is_system else names[0]),'Console selected a fake/default brain') accessible=api.get('/system/workspaces',headers=browser);check(accessible.status_code==200,'Console discovery failed') if not is_system: check({w['brain_id'] for w in accessible.json()['workspaces']}=={names[0]},'Scoped console sees another workspace') check(api.post('/auth/logout',headers=browser).status_code==403,'Cookie mutation bypassed CSRF') browser['X-CSRF-Token']=login.json()['csrf_token'] check(api.post('/auth/logout',headers={**browser,'Origin':'https://untrusted.example'}).status_code==403,'Cross-origin CSRF failure misclassified') check(api.post('/auth/logout',headers=browser).status_code==200,'Logout failed') check(api.get('/auth/me',headers=browser).status_code==401,'Session survived logout') api.cookies.clear() results['console_session_csrf_logout']='passed' rpc_headers={'Accept':'application/json, text/event-stream'} initialize={'jsonrpc':'2.0','id':1,'method':'initialize','params':{'protocolVersion':'2025-03-26','capabilities':{},'clientInfo':{'name':'offline-auth-qualification','version':'1'}}} for credential in (admin,headers): response=api.post(own+'/mcp',headers={**rpc_headers,**credential},json=initialize) check(response.status_code==200 and 'protocolVersion' in response.text,'MCP initialize authentication failed') check(api.post('/mcp',headers={**rpc_headers,**admin},json=initialize).status_code==200,'Retained MCP rejects canonical administrator') check(api.post(other+'/mcp',headers={**rpc_headers,**headers},json=initialize).status_code==403,'MCP cross-workspace scope accepted') check(api.post(own+'/mcp',headers={**rpc_headers,'Authorization':'Bearer bpat_unknown'},json=initialize).status_code==401,'MCP unknown token accepted') check(api.post(own+'/mcp',headers=rpc_headers,json=initialize).status_code==401,'MCP missing token accepted') readonly_headers={'Authorization':'Bearer '+read_only['token']} store_call={'jsonrpc':'2.0','id':2,'method':'tools/call','params':{'name':'memory_store','arguments':{'text':'must not write'}}} check(api.post(own+'/mcp',headers={**rpc_headers,**readonly_headers},json=store_call).status_code==403,'Read-only MCP credential can write') results['mcp_auth_parity_permissions']='passed' malformed={'data':[{'object':{'name':'only-object','type':'ENTITY'}}]} for path in ('/ingest/structured',own+'/api/ingest/structured'): response=api.post(path,headers={**headers,'X-Brain-ID':names[0]},json=malformed) check(response.status_code==422 and response.json()['error']['code']=='VALIDATION_ERROR','Structured validation did not return 422') try:accepted_task_id(response.status_code,response.json()) except ValueError:pass else:raise RuntimeError('Rejected ingestion may start task polling') results['validation_422_no_polling']='passed' login=api.post('/auth/session',json={'token':scoped['token']});login.raise_for_status() browser={'Cookie':'__Host-brainapi-session='+login.cookies.get('__Host-brainapi-session')} check(api.delete('/system/access-tokens/'+scoped['id'],headers=admin).status_code==200,'Revocation failed') check(api.get(own+'/agent/capabilities',headers=headers).status_code==401,'Revoked API credential accepted') check(api.post(own+'/mcp',headers={**rpc_headers,**headers},json=initialize).status_code==401,'Revoked MCP credential accepted') check(api.get('/auth/me',headers=browser).status_code==401,'Session survived originating token revocation') results['revocation_all_surfaces']='passed' check({p.name for p in default_health_probes()}=={'postgresql','redis','ollama'},'Monitoring inactive backends') for _ in range(11000):tracer.publish('heartbeat','qualification.heartbeat') check(tracer.queue.size==0 and tracer.dropped_events==0,'Default trace sink accumulates events') results['active_probes_bounded_tracing']='passed' health=api.get('/health').json() check(health['persistence']['mode'] in {'durable','ephemeral'},'Persistence state absent') if health['persistence']['mode']=='ephemeral': check(health['status']=='degraded' and not health['persistence']['production_ready'],'Ephemeral storage claims production readiness') results['persistence']=health['persistence'] finally: for token_id in issued: api.delete('/system/access-tokens/'+token_id,headers=admin).raise_for_status() for name in created: api.delete('/system/brains/'+name,headers=admin).raise_for_status() api.patch('/system/workspaces/'+name,headers=admin,json={'archived':True}).raise_for_status() after=api.get('/system/brains-list',headers=admin);after.raise_for_status() check({(b['name_key'],b['id']) for b in after.json()}==identities,'Qualification changed legitimate resources') results['result']='passed' print(json.dumps(results,sort_keys=True)) if __name__=='__main__':qualify()