"""Synthetic hook transitions, durable outbox, and HTTP security regressions.""" import concurrent.futures import http.client import json from pathlib import Path import sys import tempfile import threading import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from agent_control.store import Store, Invalid, Conflict from agent_control.server import Server def sample(**overrides): data = dict(event_id='failure-1', source='synthetic-test', severity='critical', status='firing', summary='Synthetic worker failure', impact='Task paused', evidence=['synthetic check result'], action='No action taken', next_step='Inspect task') data.update(overrides) return data class HookStoreTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = Path(self.tmp.name) / 'test.sqlite3' self.store = Store(self.path) def tearDown(self): self.tmp.cleanup() def test_persistent_dedup_ack_and_no_queue_or_execution(self): with patch('subprocess.run', side_effect=AssertionError('hook must not execute')): first = self.store.hook(sample(action='shell text is data only')) again = Store(self.path).hook(sample(action='shell text is data only')) self.assertFalse(first['duplicate']) self.assertTrue(again['duplicate']) self.assertEqual(first['notifications'], again['notifications']) self.assertEqual({x['target'] for x in first['notifications']}, {'control', 'human'}) self.assertTrue(all(x['authorization'] is False for x in first['notifications'])) self.assertEqual(len(self.store.state()['events']), 1) self.assertEqual(self.store.state()['hooks']['pending_notifications'], 2) self.assertEqual(self.store.state()['hooks']['active_firing'], 1) nid = first['notifications'][0]['notification_id'] self.assertFalse(self.store.hook_ack({'notification_id':nid})['duplicate']) self.assertTrue(Store(self.path).hook_ack({'notification_id':nid})['duplicate']) self.assertEqual(self.store.hook_notifications()['pending_count'], 1) self.assertTrue(self.store.hook(sample(action='shell text is data only'))['duplicate']) self.assertEqual(self.store.hook_notifications()['pending_count'], 1) self.assertEqual(len(self.store.state()['events']), 1) self.assertEqual(self.store.state()['sessions'], []) def test_info_warning_and_critical_routes(self): for severity, count in [('info',1), ('warning',1), ('critical',2)]: result = self.store.hook(sample(event_id=severity, severity=severity)) self.assertEqual(len(result['notifications']), count) self.assertEqual(result['notifications'][0]['target'], 'control') self.assertEqual(self.store.hook_notifications()['pending_count'], 4) def test_resolution_requires_firing_and_inherits_severity_and_objects(self): with self.assertRaises(Conflict): self.store.hook(sample(status='resolved')) self.store.session(dict(id='session-a', name='Synthetic session')) self.store.task(dict(id='task-a', name='Synthetic task', session_id='session-a')) self.store.hook(sample(task_id='task-a', session_id='session-a')) resolved = self.store.hook(sample(status='resolved', severity='info', summary='Recovered')) self.assertEqual(resolved['hook']['severity'], 'critical') self.assertEqual(resolved['hook']['task_id'], 'task-a') self.assertEqual(resolved['hook']['session_id'], 'session-a') self.assertEqual(len(resolved['notifications']), 2) self.assertTrue(self.store.hook(sample(status='resolved', severity='warning', summary='Recovered'))['duplicate']) self.assertEqual(self.store.state()['hooks']['active_firing'], 0) self.assertEqual(sum(e['kind'].startswith('hook.') for e in self.store.state()['events']), 2) self.assertTrue(self.store.hook(sample(task_id='task-a', session_id='session-a'))['duplicate']) self.assertEqual(self.store.hook_notifications()['pending_count'], 4) # Recurrence uses a new event ID and creates a new lifecycle. self.store.hook(sample(event_id='failure-2')) self.assertEqual(self.store.state()['hooks']['active_firing'], 1) def test_conflicting_duplicates_and_resolution_objects_do_not_mutate(self): self.store.hook(sample()) before = self.store.hook_notifications() for update in ({'summary':'different'}, {'severity':'warning'}, {'evidence':['different']}): with self.assertRaises(Conflict): self.store.hook(sample(**update)) with self.assertRaises(Conflict): self.store.hook(sample(status='resolved',session_id='other')) self.assertEqual(self.store.hook_notifications(), before) self.assertEqual(len(self.store.state()['events']), 1) def test_schema_rejects_unknown_types_lengths_and_unregistered_references(self): invalid = [sample(shell='anything'), sample(severity='urgent'), sample(status='done'), sample(summary=''), sample(summary='x'*1001), sample(impact='x'*2001), sample(source='bad source'), sample(event_id=[]), sample(evidence='not an array'), sample(evidence=['x']*9), sample(evidence=['x'*1001]), sample(evidence=[{}]), sample(task_id='missing'), sample(session_id='missing'), sample(action=None), {k:v for k,v in sample().items() if k!='summary'}] for payload in invalid: with self.subTest(payload=payload): with self.assertRaises(Invalid): self.store.hook(payload) self.assertEqual(self.store.state()['hooks']['recent'], []) self.assertEqual(self.store.hook_notifications()['pending_count'], 0) for payload in ({}, {'notification_id':'missing'}, {'notification_id':'id','shell':'x'}): with self.assertRaises(Invalid): self.store.hook_ack(payload) def test_atomic_concurrent_duplicates_and_ack(self): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _: self.store.hook(sample()), range(16))) self.assertEqual(sum(not r['duplicate'] for r in results), 1) self.assertEqual(self.store.hook_notifications()['pending_count'], 2) nid = results[0]['notifications'][0]['notification_id'] with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _: self.store.hook_ack({'notification_id':nid}), range(16))) self.assertEqual(sum(not r['duplicate'] for r in results), 1) self.assertEqual(len(self.store.state()['events']), 1) def test_bounded_pending_and_recent_with_source_isolation(self): for index in range(105): self.store.hook(sample(event_id='event-'+str(index),severity='info')) result = self.store.hook_notifications() self.assertEqual(len(result['notifications']), 100) self.assertEqual(result['pending_count'], 105) self.assertTrue(result['has_more']) self.assertEqual(len(self.store.state()['hooks']['recent']), 20) self.assertEqual(result['notifications'][0]['hook']['event_id'], 'event-0') self.store.hook(sample(event_id='event-0',source='different-source',severity='info')) self.assertEqual(self.store.hook_notifications()['pending_count'], 106) class HookHTTPTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() root = Path(self.tmp.name) root.chmod(0o700) token = root / 'token' self.secret = 'h'*48 token.write_text(self.secret) token.chmod(0o600) self.metrics = patch('agent_control.server.MetricsCache') mock = self.metrics.start() mock.return_value.snapshot.return_value = {'status':'synthetic'} self.server = Server(dict(bind='127.0.0.1',port=0,runtime_root=str(root),token_file=str(token))) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join() self.metrics.stop() self.tmp.cleanup() def request(self,path,method='GET',payload=None,headers=None): conn = http.client.HTTPConnection(*self.server.server_address,timeout=3) merged = {'Authorization':'Bearer '+self.secret,'X-CSRF-Token':'1','Content-Type':'application/json'} merged.update(headers or {}) conn.request(method,path,json.dumps(payload) if payload is not None else None,merged) response = conn.getresponse() status, body = response.status,json.loads(response.read()) conn.close() return status, body def test_routes_auth_origin_csrf_validation_and_state(self): self.assertEqual(self.request('/api/hooks','POST',sample(),{'Authorization':''})[0],401) self.assertEqual(self.request('/api/hooks','POST',sample(),{'X-CSRF-Token':''})[0],403) self.assertEqual(self.request('/api/hooks','POST',sample(),{'Origin':'https://untrusted.example'})[0],403) self.assertEqual(self.request('/api/hooks','POST',sample(shell='x'))[0],400) self.assertEqual(self.request('/api/hooks','POST',sample(status='resolved'))[0],409) self.assertEqual(self.request('/api/hook-notifications',headers={'Authorization':''})[0],401) status, first = self.request('/api/hooks','POST',sample()) self.assertEqual(status,200) self.assertTrue(self.request('/api/hooks','POST',sample())[1]['duplicate']) status, queue = self.request('/api/hook-notifications') self.assertEqual(status,200) self.assertEqual(len(queue['notifications']),2) nid = first['notifications'][0]['notification_id'] self.assertEqual(self.request('/api/hooks/ack','POST',{'notification_id':nid},{'Authorization':''})[0],401) self.assertEqual(self.request('/api/hooks/ack','POST',{'notification_id':nid,'target':'human'})[0],400) self.assertEqual(self.request('/api/hooks/ack','POST',{'notification_id':nid})[1]['status'],'delivered') self.assertTrue(self.request('/api/hooks/ack','POST',{'notification_id':nid})[1]['duplicate']) self.assertEqual(self.request('/api/state')[1]['hooks']['pending_notifications'],1) self.assertEqual(self.request('/api/hooks','POST',sample(status='resolved',severity='info'))[1]['hook']['severity'],'critical') self.assertEqual(self.request('/api/state')[1]['hooks']['active_firing'],0) self.assertEqual(self.request('/api/hook-notifications?target=human')[0],404) if __name__ == '__main__': unittest.main()