From fb3178d34b4cbb477ade168e04fc418678b838a1 Mon Sep 17 00:00:00 2001 From: Shane Maynard Date: Sat, 3 Oct 2026 08:55:45 -0400 Subject: [PATCH] test: qualify chat cancellation and cross-client ownership --- tests/unit/test_chat_cross_client.py | 216 ++++++++++++++++++++++++ tests/unit/test_chat_security_review.py | 188 +++++++++++++++++++++ 2 files changed, 404 insertions(+) create mode 100644 tests/unit/test_chat_cross_client.py create mode 100644 tests/unit/test_chat_security_review.py diff --git a/tests/unit/test_chat_cross_client.py b/tests/unit/test_chat_cross_client.py new file mode 100644 index 0000000..548e657 --- /dev/null +++ b/tests/unit/test_chat_cross_client.py @@ -0,0 +1,216 @@ +"""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 diff --git a/tests/unit/test_chat_security_review.py b/tests/unit/test_chat_security_review.py new file mode 100644 index 0000000..65780f8 --- /dev/null +++ b/tests/unit/test_chat_security_review.py @@ -0,0 +1,188 @@ +"""Offline cancellation/recovery regressions; no HTTP, NAS, secrets or model calls.""" +import asyncio +import functools + +import pytest + + +def offline_async(function): + @functools.wraps(function) + def run(*args, **kwargs): + return asyncio.run(function(*args, **kwargs)) + return run + +from control.accounts import Accounts, token_hash +from control.chat_runs import ChatRuns +from control.chat_store import ChatConflict, ChatStore + + +class BlockedAdapter: + available = True + approvals = False + + def __init__(self, *, block_start=False, fail_stop=False): + self.block_start = block_start + self.fail_stop = fail_stop + self.start_entered = asyncio.Event() + self.start_release = asyncio.Event() + self.stop_entered = asyncio.Event() + self.stream_entered = asyncio.Event() + self.stream_release = asyncio.Event() + self.starts = [] + self.stops = [] + + async def start(self, owner, rid, payload): + self.starts.append(rid) + self.start_entered.set() + if self.block_start: + await self.start_release.wait() + return {'run_id': 'offline-upstream'} + + async def stop(self, owner, upstream): + self.stops.append(upstream) + self.stop_entered.set() + if self.fail_stop: + raise RuntimeError('Offline simulated unavailable stop') + + async def status(self, owner, upstream): + return {'status': 'cancelled' if self.stops and not self.fail_stop else 'running'} + + async def events(self, owner, upstream): + self.stream_entered.set() + await self.stream_release.wait() + yield {'event': 'run.cancelled'} + + +@pytest.fixture +def execution(tmp_path): + accounts = Accounts(str(tmp_path / 'offline.sqlite')) + user = accounts.create('review', 'Offline Review', 'user', 'offline-review-password') + accounts.update(user['id'], status='ready') + token, _ = accounts.issue_session(user['id'], 3600, user['password_hash']) + store = ChatStore(accounts) + cid = store.create_conversation(user['id'])['id'] + operation = store.send(user['id'], cid, 'Offline request', 'offline-key') + yield accounts, store, user['id'], token, cid, operation['run']['id'] + accounts.close() + + +def coordinator(execution, adapter, *, upstream=None): + accounts, store, owner, token, cid, rid = execution + runs = ChatRuns(accounts, store, adapter=adapter) + runs.metadata(owner, rid, {'conversation_id': cid, 'payload': {'input': 'Offline request'}, + 'session_hash': token_hash(token), 'room': None, + 'upstream_id': upstream, 'assistant_id': None}) + return runs + + +@offline_async +async def test_cancel_during_blocked_admission_stops_learned_upstream(execution): + _, store, owner, _, _, rid = execution + adapter = BlockedAdapter(block_start=True) + runs = coordinator(execution, adapter) + runs.schedule(owner, rid) + try: + await asyncio.wait_for(adapter.start_entered.wait(), 2) + await runs.cancel(owner, rid) + assert not runs.tasks[rid].done(), 'Do not abort admission before learning its upstream ID' + adapter.start_release.set() + await asyncio.wait_for(runs.tasks[rid], 2) + assert adapter.starts == [rid] + assert adapter.stops, 'Cancellation must stop the execution admitted during cancellation' + assert store.get_run(owner, rid)['run']['status'] == 'cancelled' + finally: + adapter.start_release.set() + await runs.stop() + + +@offline_async +async def test_stop_failure_keeps_execution_slot_and_durable_intent(execution): + _, store, owner, _, _, rid = execution + adapter = BlockedAdapter(fail_stop=True) + runs = coordinator(execution, adapter, upstream='offline-upstream') + store.update_run(owner, rid, 'running') + runs.schedule(owner, rid) + try: + await asyncio.wait_for(adapter.stream_entered.wait(), 2) + await runs.cancel(owner, rid) + assert runs.metadata(owner, rid).get('stop_requested') is True + assert store.get_run(owner, rid)['run']['status'] not in runs.store.TERMINAL + assert runs.locks[owner].locked(), 'Do not admit new account work while old work is unconfirmed' + assert not runs.tasks[rid].done() + finally: + await runs.stop() + + +@offline_async +async def test_revoked_session_recovery_stops_remote_before_terminalizing(execution): + accounts, store, owner, token, _, rid = execution + adapter = BlockedAdapter() + runs = coordinator(execution, adapter, upstream='offline-upstream') + store.update_run(owner, rid, 'running') + accounts.revoke(token) + try: + await asyncio.wait_for(runs.execute(owner, rid), 2) + assert adapter.starts == [], 'Recovery must never replay a known admission' + assert adapter.stops, 'A revoked session does not mean its remote execution stopped' + assert store.get_run(owner, rid)['run']['status'] == 'cancelled' + finally: + await runs.stop() + + +def test_stale_stream_metadata_cannot_erase_cancellation_intent(execution): + _, _, owner, _, _, rid = execution + runs = coordinator(execution, BlockedAdapter(), upstream='offline-upstream') + stale = runs.metadata(owner, rid) + cancel = runs.metadata(owner, rid) + cancel['stop_requested'] = True + runs.metadata(owner, rid, cancel) + stale['content'] = 'A delta racing with cancellation' + runs.metadata(owner, rid, stale) + assert runs.metadata(owner, rid).get('stop_requested') is True + + +@offline_async +async def test_restart_after_cancel_during_admission_reconciles_remote(execution): + _, store, owner, _, _, rid = execution + adapter = BlockedAdapter(block_start=True, fail_stop=True) + original = coordinator(execution, adapter) + original.schedule(owner, rid) + try: + await asyncio.wait_for(adapter.start_entered.wait(), 2) + await original.cancel(owner, rid) + adapter.start_release.set() + await asyncio.wait_for(adapter.stop_entered.wait(), 2) + finally: + adapter.start_release.set() + await original.stop() + # The remote run exists, its stop failed, and the local projection may already + # be terminal. A restart must still reconcile that durable remote stop intent. + recovered_adapter = BlockedAdapter() + recovered = ChatRuns(execution[0], store, adapter=recovered_adapter) + try: + await recovered.start() + await asyncio.wait_for(recovered_adapter.stop_entered.wait(), 2) + if rid in recovered.tasks: + await asyncio.wait_for(recovered.tasks[rid], 2) + assert recovered_adapter.starts == [] + assert store.get_run(owner, rid)['run']['status'] == 'cancelled' + finally: + await recovered.stop() + + +@offline_async +async def test_delete_cannot_discard_unconfirmed_remote_cancellation(execution): + _, store, owner, _, cid, rid = execution + adapter = BlockedAdapter(fail_stop=True) + runs = coordinator(execution, adapter, upstream='offline-upstream') + store.update_run(owner, rid, 'running') + runs.schedule(owner, rid) + try: + await asyncio.wait_for(adapter.stream_entered.wait(), 2) + with pytest.raises(ChatConflict): + await runs.cancel_conversation(owner, cid) + store.delete_conversation(owner, cid) + assert store.get_run(owner, rid)['run']['status'] not in store.TERMINAL + assert runs.metadata(owner, rid).get('stop_requested') is True + finally: + await runs.stop()