admin_rb
89 lines
4.8 KiB
Python
89 lines
4.8 KiB
Python
"""Public-result framing with synthetic JSONL commands; no Ansible or remote calls."""
|
|
from dataclasses import replace
|
|
import json,os,pwd,socket,struct,sys,threading,time
|
|
from pathlib import Path
|
|
import pytest
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui.core.client import CoreClient
|
|
from aim_webgui.core.executor import Executor
|
|
from aim_webgui.core.jsonio import loads
|
|
from aim_webgui.core.reports import encoded
|
|
from aim_webgui.credentials import wire
|
|
from aim_webgui.errors import WebError
|
|
from evidence_fixtures import result,plan
|
|
|
|
|
|
def large_fixture(tmp_path, *, full_budget=False):
|
|
decl={'protocol':'aim_output_v1','schema':'synthetic_rows_v1','scope':'per_host','required':True,'sensitivity':'safe',
|
|
'max_bytes_per_host':1048576,'data_schema':{'type':'object','properties':{'rows':{'type':'array','maxItems':20000,
|
|
'items':{'type':'string','maxLength':8192}}},'required':['rows'],'additionalProperties':False}}
|
|
hosts=['test'+str(i)+'.example' for i in range(16 if full_budget else 4)]
|
|
rows=['\u2603'*2600]*128 if full_budget else ['synthetic-\u2603'*90]*500
|
|
final=result(decl,hosts,report_data={'rows':rows})
|
|
assert 1024*1024<len(encoded(final))<16*1024*1024
|
|
script=tmp_path/'wire_fixture.py';payload=tmp_path/'fixture-result.json';payload.write_bytes(encoded(final))
|
|
script.write_text('''import os,sys,json
|
|
request=json.load(sys.stdin)
|
|
fd=int(sys.argv[sys.argv.index('--credentials-fd')+1])
|
|
with os.fdopen(fd) as stream: credentials=json.load(stream)
|
|
assert credentials['vault_password']=='SYNTHETIC-PRIVATE-ONLY'
|
|
assert 'credentials' not in request
|
|
r=json.load(open('''+repr(str(payload))+'''))
|
|
e={'event_version':'1.0','run_id':'fixture-run','sequence':1,'timestamp':'2026-09-20T10:00:00Z','kind':'result','status':'succeeded','result':r}
|
|
for value in [{'type':'event','event':e},{'type':'response','api_version':'1.0','ok':True,'result':r}]:
|
|
raw=(json.dumps(value,ensure_ascii=False)+'\\n').encode()
|
|
for i in range(0,len(raw),16381):os.write(1,raw[i:i+16381])
|
|
''')
|
|
if full_budget:script.write_text(script.read_text().replace('ensure_ascii=False','ensure_ascii=True'))
|
|
settings=Settings(core_transport='stdio',core_command=(sys.executable,str(script)),core_config=tmp_path/'synthetic.yml')
|
|
return settings,decl,hosts,final
|
|
|
|
|
|
def test_large_result_stdio_and_duplicate_event_is_not_report_sample(tmp_path):
|
|
settings,decl,hosts,expected=large_fixture(tmp_path);events=[]
|
|
r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
|
|
credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'},result_contract=decl,event_sink=events.append)
|
|
assert r['operation_result']==expected['operation_result']
|
|
assert len(events)==1 and 'operation_result'not in events[0] and 'result'not in events[0]
|
|
|
|
|
|
def test_large_result_through_executor_socket_and_unchanged_secret_limits(tmp_path):
|
|
settings,decl,hosts,expected=large_fixture(tmp_path)
|
|
path=tmp_path/'core.sock'
|
|
who=pwd.getpwuid(os.geteuid()).pw_name
|
|
settings=replace(settings,core_transport='unix',core_socket=path,core_executor_user=who,core_client_user=who)
|
|
server=socket.socket(socket.AF_UNIX);server.bind(str(path));path.chmod(0o660);server.listen(1)
|
|
executor=Executor(settings)
|
|
def run():
|
|
conn,_=server.accept();executor.handle(conn)
|
|
thread=threading.Thread(target=run);thread.start()
|
|
try:
|
|
r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
|
|
result_contract=decl,credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'})
|
|
assert r['operation_result']==expected['operation_result']
|
|
finally:thread.join(timeout=10);server.close()
|
|
assert not thread.is_alive()
|
|
assert wire.SECRET_LIMIT==8192 and wire.MAX_MESSAGE==1048576
|
|
|
|
|
|
@pytest.mark.parametrize('raw',[b'{"a":1,"a":2}',b'{"a":NaN}',b'['*49+b'0'+b']'*49,b'{"a":"\xff"}',b'{"a":'])
|
|
def test_strict_decoder_rejects_ambiguous_or_torn_json(raw):
|
|
with pytest.raises((ValueError,UnicodeError)):loads(raw)
|
|
|
|
|
|
def test_frame_bounds_before_allocation_and_torn_frames():
|
|
for body in (struct.pack('!I',33*1024*1024),struct.pack('!I',30)+b'{}'):
|
|
a,b=socket.socketpair()
|
|
try:
|
|
a.sendall(body);a.shutdown(socket.SHUT_WR)
|
|
with pytest.raises(ValueError):wire.receive(b,32*1024*1024,decoder=loads)
|
|
finally:a.close();b.close()
|
|
|
|
|
|
def test_near_run_limit_with_escaped_unicode_and_repeated_final(tmp_path):
|
|
settings,decl,hosts,expected=large_fixture(tmp_path,full_budget=True)
|
|
assert 15*1024*1024<len(encoded(expected))<16*1024*1024
|
|
r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
|
|
result_contract=decl,credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'})
|
|
assert r['operation_result']==expected['operation_result']
|