143 lines
9.3 KiB
Python
143 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Explicit advanced live receipt. Protected credentials; bounded disposable data.
|
|
|
|
Never changes existing memory/settings, runs scheduled tasks, or prints credentials.
|
|
Research is one ordinary Hermes run. Deletes only objects created by this script.
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import secrets
|
|
import struct
|
|
import time
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
def png():
|
|
def chunk(kind, data):
|
|
return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data))
|
|
pixels = b''.join(b'\x00' + b'\xff\x00\x00' * 16 + b'\x00\x00\xff' * 16 for _ in range(16))
|
|
return b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', 32, 16, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(pixels)) + chunk(b'IEND', b'')
|
|
|
|
|
|
def pdf(text):
|
|
objects = [b'<< /Type /Catalog /Pages 2 0 R >>', b'<< /Type /Pages /Kids [3 0 R] /Count 1 >>',
|
|
b'<< /Type /Page /Parent 2 0 R /MediaBox [0 0 300 200] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>',
|
|
b'<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>']
|
|
stream = ('BT /F1 12 Tf 20 100 Td (' + text + ') Tj ET').encode('ascii')
|
|
objects.append(b'<< /Length ' + str(len(stream)).encode() + b' >>\nstream\n' + stream + b'\nendstream')
|
|
output = bytearray(b'%PDF-1.4\n'); offsets = [0]
|
|
for number, obj in enumerate(objects, 1):
|
|
offsets.append(len(output)); output.extend(str(number).encode() + b' 0 obj\n' + obj + b'\nendobj\n')
|
|
xref = len(output)
|
|
output.extend(b'xref\n0 6\n0000000000 65535 f \n')
|
|
for offset in offsets[1:]: output.extend(f'{offset:010d} 00000 n \n'.encode())
|
|
output.extend(f'trailer\n<< /Size 6 /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n'.encode())
|
|
return bytes(output)
|
|
|
|
|
|
async def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--url', default='https://192.168.86.2:9178')
|
|
parser.add_argument('--credentials', default='.runtime/deployment.json')
|
|
parser.add_argument('--receipt', default='.runtime/chat-features-live-receipt.json')
|
|
args = parser.parse_args()
|
|
source = Path(args.credentials)
|
|
if source.is_symlink() or source.stat().st_mode & 0o077:
|
|
raise RuntimeError('Credentials must be private regular file')
|
|
credentials = json.loads(source.read_text())
|
|
stamp = secrets.token_hex(6)
|
|
receipt = {'url': args.url, 'started_at': time.time(), 'checks': {}, 'cleanup_pending': []}
|
|
attachments, conversations, tasks = [], [], []
|
|
failure = None
|
|
async with httpx.AsyncClient(base_url=args.url, verify=False, timeout=180, follow_redirects=False) as client:
|
|
async def request(method, path, **kwargs):
|
|
response = await client.request(method, path, **kwargs)
|
|
if response.status_code >= 400:
|
|
raise RuntimeError(f'{method} {path.split("?")[0]} status {response.status_code}')
|
|
return response
|
|
async def upload(data, mime, name):
|
|
item = (await request('POST', '/api/attachments', content=data, headers={'Content-Type': mime, 'X-Filename': name})).json()['attachment']
|
|
attachments.append(item['id']); return item
|
|
async def turn(content, ids=(), mode=None):
|
|
conversation = (await request('POST', '/api/conversations', json={'title': 'Advanced qualification ' + stamp})).json()['conversation']
|
|
conversations.append(conversation['id'])
|
|
body = {'content': content, 'attachment_ids': list(ids), 'idempotency_key': secrets.token_hex(16)}
|
|
if mode: body['mode'] = mode
|
|
accepted = (await request('POST', '/api/conversations/' + conversation['id'] + '/messages', json=body)).json()
|
|
rid = accepted['run']['id']
|
|
async with client.stream('GET', '/api/runs/' + rid + '/events') as response:
|
|
if response.status_code != 200: raise RuntimeError('Run events unavailable')
|
|
async for _ in response.aiter_lines(): pass
|
|
detail = (await request('GET', '/api/runs/' + rid)).json()
|
|
if detail['run']['status'] != 'completed': raise RuntimeError('Native Hermes run did not complete')
|
|
return '\n'.join(message['content'] for message in detail['messages'] if message['role'] == 'assistant' and message['status'] == 'completed')
|
|
try:
|
|
login = (await request('POST', '/api/auth/login', json={'username': 'admin', 'password': credentials['admin_password']})).json()
|
|
client.cookies.clear(); client.headers['Authorization'] = 'Bearer ' + login['token']
|
|
caps = (await request('GET', '/api/capabilities')).json()['features']
|
|
receipt['capabilities'] = {key: caps.get(key) for key in ('attachments', 'pdf_extraction', 'image_input', 'scheduled_tasks', 'memory_controls', 'research', 'connectors')}
|
|
text_code, pdf_code = 'TXT' + stamp, 'PDF' + stamp
|
|
text = await upload(('Qualification token: ' + text_code).encode(), 'text/plain', 'qualification.txt')
|
|
assert text['extraction_status'] == 'ready'
|
|
assert text_code in await turn('Read the attached text and return its qualification token only. Do not use tools.', [text['id']])
|
|
receipt['checks']['text_attachment_native_reply'] = True
|
|
document = await upload(pdf('Qualification token ' + pdf_code), 'application/pdf', 'qualification.pdf')
|
|
assert document['extraction_status'] == 'ready'
|
|
assert pdf_code in await turn('Read the attached document and return its qualification token only. Do not use tools.', [document['id']])
|
|
receipt['checks']['pdf_extraction_native_reply'] = True
|
|
image = await upload(png(), 'image/png', 'qualification.png')
|
|
answer = await turn('Name the two panel colors left to right in this attached image. Answer two colors only. Do not use tools.', [image['id']])
|
|
assert 'red' in answer.lower() and 'blue' in answer.lower() and answer.lower().index('red') < answer.lower().index('blue')
|
|
receipt['checks']['image_native_vision_reply'] = True
|
|
memory = (await request('GET', '/api/memory')).json()
|
|
assert {item['section'] for item in memory['sections']} == {'memory', 'user'}
|
|
assert all(len(item['revision']) == 64 for item in memory['sections'])
|
|
receipt['checks']['memory_read_only'] = True
|
|
del memory # Never include existing private memory content in receipt.
|
|
task = (await request('POST', '/api/tasks', json={'name': 'Qualification ' + stamp, 'schedule': '2099-01-01T00:00:00Z',
|
|
'prompt': 'Qualification placeholder; do not execute before scheduled time.', 'paused': True, 'repeat': 1})).json()['task']
|
|
tasks.append(task['id'])
|
|
for action in ('resume', 'pause'):
|
|
await request('POST', '/api/tasks/' + task['id'] + '/' + action)
|
|
await request('PATCH', '/api/tasks/' + task['id'], json={'name': 'Qualification edited ' + stamp, 'enabled': False})
|
|
listed = (await request('GET', '/api/tasks')).json()['tasks']
|
|
assert any(item['id'] == task['id'] for item in listed)
|
|
await request('DELETE', '/api/tasks/' + task['id']); tasks.remove(task['id'])
|
|
receipt['checks']['task_create_pause_resume_edit_delete_no_run'] = True
|
|
research = (await request('GET', '/api/research')).json()
|
|
assert research['available'] and research['mode'] == 'hermes_tools'
|
|
answer = await turn('Use web research only to verify what organization publishes the Python programming language official documentation. '
|
|
'Consult https://docs.python.org/ and cite that source URL. Do not write files, run shell commands, schedule jobs or modify memory.', mode='research')
|
|
assert 'https://docs.python.org' in answer
|
|
receipt['checks']['native_research_cited_reply'] = True
|
|
except Exception as error:
|
|
# Never serialize server bodies/provider exception details.
|
|
failure = type(error).__name__ + ': ' + str(error) if isinstance(error, (AssertionError, RuntimeError)) else type(error).__name__
|
|
receipt['failure'] = failure
|
|
finally:
|
|
for kind, ids in (('tasks', tasks), ('attachments', attachments), ('conversations', conversations)):
|
|
for identifier in ids:
|
|
response = await client.delete('/api/' + kind + '/' + identifier)
|
|
if response.status_code not in (200, 404):
|
|
receipt['cleanup_pending'].append({'kind': kind, 'id': identifier, 'status': response.status_code})
|
|
if 'Authorization' in client.headers:
|
|
await client.post('/api/auth/logout')
|
|
receipt['finished_at'] = time.time()
|
|
target = Path(args.receipt); target.parent.mkdir(parents=True, exist_ok=True)
|
|
if target.is_symlink(): raise RuntimeError('Receipt path cannot be a symlink')
|
|
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, 'w') as stream: json.dump(receipt, stream, indent=2)
|
|
print('Advanced live checks:', ', '.join(receipt['checks']))
|
|
if failure:
|
|
print('Qualification incomplete:', failure)
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == '__main__': asyncio.run(main())
|