"""Real ASGI API, independent cookie/bearer clients, entirely offline adapters.""" import asyncio import base64 import functools from contextlib import asynccontextmanager import httpx from control.app import Settings, create_app PASSWORD = 'offline-cross-client-password' ORIGIN = 'https://offline.test' INTERNAL = {'X-Control-Secret': 'offline-control-secret-not-live-1234567890'} def offline_async(function): @functools.wraps(function) def run(*args, **kwargs): return asyncio.run(function(*args, **kwargs)) return run class Provisioner: async def ensure(self, user): return {'container_name': user['container_name'], 'hermes_url': 'http://127.0.0.1:12345', 'hermes_api_key': 'offline-placeholder'} async def ready(self, user): return True async def remove(self, user): pass class Features: async def feature_capabilities(self, owner): return {'image_input': True, 'scheduled_tasks': False, 'memory_controls': False} class Runs: available = True approvals = False def __init__(self): self.started = [] self.stopped = [] async def start(self, owner, rid, payload): self.started.append((owner, rid, payload)) return {'run_id': 'offline-' + rid} async def status(self, owner, upstream): return {'status': 'running'} async def events(self, owner, upstream): yield {'event': 'message.delta', 'delta': 'Offline answer'} yield {'event': 'run.completed', 'output': 'Offline answer'} async def stop(self, owner, upstream): self.stopped.append(upstream) @asynccontextmanager async def clients(tmp_path): adapter = Runs() app = create_app(Settings(db_path=str(tmp_path/'offline.sqlite'), admin_password=PASSWORD, allowed_origins=(ORIGIN,), service_secret=INTERNAL['X-Control-Secret'], reconcile_interval=3600), Provisioner(), chat_adapter=adapter, feature_provider=Features()) async with app.router.lifespan_context(app): transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url=ORIGIN) as browser, \ httpx.AsyncClient(transport=transport, base_url=ORIGIN) as native, \ httpx.AsyncClient(transport=transport, base_url=ORIGIN) as anonymous: login = await browser.post('/api/auth/login', headers={'Origin': ORIGIN}, json={'username': 'admin', 'password': PASSWORD}) assert login.status_code == 200 user = login.json()['user'] login2 = await native.post('/api/auth/login', json={'username': 'admin', 'password': PASSWORD}) assert login2.status_code == 200 native_token = login2.json()['token'] native.cookies.clear() native.headers['Authorization'] = 'Bearer ' + native_token browser.headers['Origin'] = ORIGIN # Provisioning adapter is fake and completes locally; no socket is opened. for _ in range(200): if app.state.accounts.get(user['id'])['status'] == 'ready': break await asyncio.sleep(.001) assert app.state.accounts.get(user['id'])['status'] == 'ready' yield app, adapter, browser, native, anonymous, user, native_token async def completed(client, rid): for _ in range(200): response = await client.get('/api/runs/' + rid) assert response.status_code == 200 if response.json()['run']['status'] in ('completed', 'failed', 'cancelled'): return response.json() await asyncio.sleep(.005) raise AssertionError('Offline run did not finish') @offline_async async def test_browser_native_resume_pin_preferences_pages_and_event_cursor(tmp_path): async with clients(tmp_path) as (app, adapter, browser, native, anonymous, user, token): assert (await anonymous.get('/api/conversations')).status_code == 401 response = await browser.post('/api/conversations', json={'title': 'Shared devices'}) assert response.status_code == 200 conversation = response.json()['conversation']; cid = conversation['id'] pinned = await native.patch('/api/conversations/' + cid, json={'revision': conversation['revision'], 'pinned': True}) assert pinned.status_code == 200 listed = (await browser.get('/api/conversations')).json()['conversations'] assert listed[0]['id'] == cid and listed[0]['pinned'] is True saved = await browser.put('/api/preferences', json={'revision': 0, 'palette': 'sage', 'appearance': 'dark'}) assert saved.status_code == 200 assert (await native.get('/api/preferences')).json() == saved.json() operation = await browser.post('/api/conversations/' + cid + '/messages', json={'content': 'From browser', 'idempotency_key': 'browser-turn'}) assert operation.status_code == 202 rid = operation.json()['run']['id'] assert (await completed(native, rid))['run']['status'] == 'completed' # A separate bearer-authenticated device retries the uncertain same turn. replay = await native.post('/api/conversations/' + cid + '/messages', json={'content': 'From browser', 'idempotency_key': 'browser-turn'}) assert replay.status_code == 202 and replay.json()['run']['id'] == rid assert len(adapter.started) == 1 recent = (await native.get('/api/conversations/' + cid, params={'limit': 1})).json() assert recent['messages'][0]['role'] == 'assistant' older = (await browser.get('/api/conversations/' + cid, params={'limit': 1, 'cursor': recent['next_cursor']})).json() assert older['messages'][0]['content'] == 'From browser' event_stream = await native.get('/api/runs/' + rid + '/events') assert event_stream.status_code == 200 and 'event: done' in event_stream.text ids = [int(line[4:]) for line in event_stream.text.splitlines() if line.startswith('id: ')] resumed = await browser.get('/api/runs/' + rid + '/events', headers={'Last-Event-ID': str(ids[-1])}) assert resumed.status_code == 200 and 'data:' not in resumed.text changes = (await native.get('/api/sync')).json()['changes'] assert any(item['kind'] == 'preferences' for item in changes) @offline_async async def test_cookie_csrf_and_independent_session_revocation(tmp_path): async with clients(tmp_path) as (app, adapter, browser, native, anonymous, user, token): browser.headers.pop('Origin') assert (await browser.post('/api/conversations', json={'title': 'No origin'})).status_code == 403 assert (await browser.post('/api/conversations', headers={'Origin': 'https://evil.test'}, json={'title': 'Wrong origin'})).status_code == 403 browser.headers['Origin'] = ORIGIN assert (await native.post('/api/conversations', json={'title': 'Native no origin'})).status_code == 200 assert (await browser.post('/api/auth/logout')).status_code == 200 assert (await browser.get('/api/conversations')).status_code == 401 assert (await native.get('/api/conversations')).status_code == 200 reset = await native.post('/api/admin/users/' + user['id'] + '/password', json={'password': 'offline-replacement-password'}) assert reset.status_code == 200 assert (await native.get('/api/conversations')).status_code == 401 @offline_async async def test_text_attachment_real_admission_provider_payload_and_cross_device_read(tmp_path): async with clients(tmp_path) as (app, adapter, browser, native, anonymous, user, token): upload = await browser.post('/api/attachments', content=b'Offline attachment marker', headers={'Content-Type': 'text/plain', 'X-Filename': 'note.txt'}) assert upload.status_code == 201 attachment = upload.json()['attachment'] cid = (await native.post('/api/conversations', json={'title': 'Attachment'})).json()['conversation']['id'] operation = await native.post('/api/conversations/' + cid + '/messages', json={'content': 'Summarize', 'attachment_ids': [attachment['id']], 'idempotency_key': 'attachment-turn'}) assert operation.status_code == 202, operation.text assert (await completed(browser, operation.json()['run']['id']))['run']['status'] == 'completed' payload = adapter.started[0][2] assert payload['input'][0]['role'] == 'user' assert any('Offline attachment marker' in part.get('text', '') for part in payload['input'][0]['content']) content = await native.get('/api/attachments/' + attachment['id'] + '/content') assert content.content == b'Offline attachment marker' assert content.headers['x-content-type-options'] == 'nosniff' transcript = (await browser.get('/api/conversations/' + cid)).json() assert transcript['messages'][0]['attachments'], 'Saved message must retain attachment selection' png = base64.b64decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jG1sAAAAASUVORK5CYII=') image = await native.post('/api/attachments', content=png, headers={'Content-Type': 'image/png', 'X-Filename': 'pixel.png'}) assert image.status_code == 201 image_turn = await browser.post('/api/conversations/' + cid + '/messages', json={'content': 'Describe image', 'attachment_ids': [image.json()['attachment']['id']], 'idempotency_key': 'image-turn'}) assert image_turn.status_code == 202, image_turn.text assert (await completed(native, image_turn.json()['run']['id']))['run']['status'] == 'completed' image_parts = adapter.started[-1][2]['input'][0]['content'] assert any(part.get('image_url', {}).get('url') == 'data:image/png;base64,' + base64.b64encode(png).decode() for part in image_parts) @offline_async async def test_room_grant_is_bound_to_original_session_and_conversation(tmp_path): async with clients(tmp_path) as (app, adapter, browser, native, anonymous, user, token): cid = (await browser.post('/api/conversations', json={'title': 'Voice resume'})).json()['conversation']['id'] room = 'offline-old-room' grant = await anonymous.post('/internal/rooms', headers=INTERNAL, json={'user_id': user['id'], 'room_name': room, 'session_token': token, 'conversation_id': cid}) assert grant.status_code == 201 assert (await anonymous.get('/internal/rooms/' + room, headers=INTERNAL)).json()['conversation_id'] == cid voice = await anonymous.post('/internal/chat/rooms/' + room + '/messages', headers=INTERNAL, json={'content': 'Voice same conversation', 'modality': 'voice', 'idempotency_key': 'voice-turn'}) assert voice.status_code == 202 await completed(browser, voice.json()['run']['id']) assert 'Voice same conversation' in str((await browser.get('/api/conversations/' + cid)).json()['messages']) assert (await native.post('/api/auth/logout')).status_code == 200 assert (await browser.get('/api/conversations/' + cid)).status_code == 200 assert (await anonymous.get('/internal/rooms/' + room, headers=INTERNAL)).status_code == 404 late = await anonymous.post('/internal/chat/rooms/' + room + '/messages', headers=INTERNAL, json={'content': 'Stale grant', 'modality': 'voice', 'idempotency_key': 'stale'}) assert late.status_code == 404 # Logging in again cannot revive the old grant. native.headers.pop('Authorization') login = await native.post('/api/auth/login', json={'username': 'admin', 'password': PASSWORD}) assert login.status_code == 200 assert (await anonymous.get('/internal/rooms/' + room, headers=INTERNAL)).status_code == 404