Spaces:
Running
Running
File size: 12,765 Bytes
2f1e7ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """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()
|