505 lines
28 KiB
Python
505 lines
28 KiB
Python
import concurrent.futures
|
|
import http.client
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from agent_control.store import Store, Invalid, Conflict
|
|
from agent_control.server import Server, EtaClock
|
|
|
|
|
|
def definition(**changes):
|
|
value = dict(id='choice',title='Choose',question='Which approach?',
|
|
options=[dict(id='a',label='Approach A',description='Synthetic option')])
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
def answer(**changes):
|
|
value = dict(option_id='a',comment='Keep this exact comment',idempotency_key='key')
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
class DecisionTests(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_task_progress_independent_timestamp_and_retention(self):
|
|
task = self.store.task(dict(id='t',name='Task',eta_basis='Estimate only'))
|
|
self.assertEqual(task['current_progress'],'')
|
|
self.assertIsNone(task['progress_updated'])
|
|
with patch('agent_control.store.time.time',return_value=100):
|
|
task = self.store.task(dict(current_progress='Validated parser'), 't')
|
|
self.assertEqual(task['progress_updated'],100)
|
|
with patch('agent_control.store.time.time',return_value=200):
|
|
task = self.store.task(dict(status='blocked',decision='Choose next step'),'t')
|
|
self.assertEqual(task['current_progress'],'Validated parser')
|
|
self.assertEqual(task['progress_updated'],100)
|
|
self.assertEqual(task['updated'],200)
|
|
with patch('agent_control.store.time.time',return_value=300):
|
|
task = self.store.task(dict(status='completed',phase='done'),'t')
|
|
self.assertEqual(task['progress_updated'],100)
|
|
self.assertEqual(task['current_progress'],'Validated parser')
|
|
with patch('agent_control.store.time.time',return_value=400):
|
|
task = self.store.task(dict(current_progress=''),'t')
|
|
self.assertEqual(task['progress_updated'],400)
|
|
self.assertEqual(task['current_progress'],'')
|
|
|
|
def test_task_progress_validation_and_legacy_defaults(self):
|
|
self.store.task(dict(id='t',name='Task'))
|
|
for data in [dict(current_progress=None),dict(current_progress=1),
|
|
dict(current_progress='x'*1001),dict(progress_updated=1)]:
|
|
with self.assertRaises(Invalid): self.store.task(data,'t')
|
|
with self.store.db() as db:
|
|
db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('legacy',json.dumps(dict(id='legacy',name='Legacy'))))
|
|
task = next(t for t in self.store.state()['tasks'] if t['id']=='legacy')
|
|
self.assertEqual(task['current_progress'],'')
|
|
self.assertIsNone(task['progress_updated'])
|
|
|
|
|
|
def eta_task(self, **changes):
|
|
data = dict(id='eta',name='Synthetic ETA',status='running',
|
|
eta_min_minutes=5,eta_max_minutes=10,eta_basis='Synthetic remaining checks')
|
|
data.update(changes)
|
|
with patch('agent_control.store.time.time',return_value=1000):
|
|
return self.store.task(data)
|
|
|
|
def eta_notice(self, now=1300):
|
|
self.store.eta_tick(now)
|
|
return self.store.eta_notifications()['notifications'][0]
|
|
|
|
def test_eta_checkpoint_clock_and_persistent_dedup(self):
|
|
task = self.eta_task()
|
|
self.assertEqual(task['eta_revision'],1)
|
|
self.assertEqual(task['eta_estimated_at'],1000)
|
|
self.assertEqual(task['eta_change_reason'],task['eta_basis'])
|
|
self.assertEqual(self.store.eta_tick(1299)['created'],0)
|
|
self.assertEqual(self.store.eta_tick(1300)['created'],1)
|
|
first = self.store.eta_notifications()['notifications'][0]
|
|
self.assertEqual(first['checkpoint'],'half')
|
|
self.assertEqual(first['target'],'control')
|
|
self.assertEqual(first['revision'],1)
|
|
self.assertEqual(Store(self.path).eta_tick(1300)['created'],0)
|
|
self.assertEqual(self.store.eta_tick(1479)['created'],0)
|
|
self.assertEqual(self.store.eta_tick(1480)['created'],1)
|
|
self.assertEqual(self.store.eta_tick(1600)['created'],1)
|
|
self.assertEqual(self.store.state()['tasks'][0]['eta_review_status'],'overdue')
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],3)
|
|
|
|
def test_eta_heartbeat_keep_and_ack_never_move_estimate(self):
|
|
self.eta_task()
|
|
nid = self.eta_notice()['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=1310):
|
|
task = self.store.task(dict(eta_min_minutes=5,eta_max_minutes=10,current_progress='Running checks'),'eta')
|
|
self.assertEqual(task['eta_estimated_at'],1000)
|
|
self.assertEqual(task['eta_revision'],1)
|
|
self.assertEqual(self.store.eta_notifications()['notifications'][0]['task']['current_progress'],'Running checks')
|
|
ack = dict(notification_id=nid)
|
|
self.assertFalse(self.store.eta_ack(ack)['duplicate'])
|
|
self.assertTrue(self.store.eta_ack(ack)['duplicate'])
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],0)
|
|
body = dict(result='keep',reason='Half the checks passed',current_progress='Verified outputs')
|
|
with patch('agent_control.store.time.time',return_value=1320):
|
|
result = self.store.eta_complete(nid,body)
|
|
self.assertEqual(result['eta_estimated_at'],1000)
|
|
self.assertEqual(result['eta_revision'],1)
|
|
self.assertEqual(result['eta_last_reviewed_at'],1320)
|
|
self.assertEqual(result['eta_review_reason'],body['reason'])
|
|
self.assertEqual(result['progress_updated'],1320)
|
|
self.assertEqual(Store(self.path).eta_complete(nid,body),result)
|
|
with self.assertRaises(Conflict):
|
|
self.store.eta_complete(nid,dict(body,reason='Changed answer'))
|
|
|
|
def test_eta_revise_changes_version_and_cancels_old_notifications(self):
|
|
self.eta_task()
|
|
self.store.eta_tick(1480)
|
|
notices = self.store.eta_notifications()['notifications']
|
|
body = dict(result='revise',reason='New failing test found',eta_min_minutes=5,eta_max_minutes=20)
|
|
with patch('agent_control.store.time.time',return_value=1481):
|
|
task = self.store.eta_complete(notices[0]['notification_id'],body)
|
|
self.assertEqual((task['eta_revision'],task['eta_estimated_at']),(2,1481))
|
|
self.assertEqual(task['eta_extension_count'],1)
|
|
self.assertEqual(task['eta_change_reason'],body['reason'])
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],0)
|
|
with self.assertRaises(Conflict):
|
|
self.store.eta_complete(notices[1]['notification_id'],dict(result='keep',reason='Late response'))
|
|
self.assertEqual(self.store.eta_tick(2081)['created'],1)
|
|
self.assertEqual(self.store.eta_notifications()['notifications'][0]['revision'],2)
|
|
with self.store.db() as db:
|
|
self.assertEqual(db.execute('SELECT count(*) FROM eta_reviews WHERE cancelled_at IS NOT NULL').fetchone()[0],2)
|
|
|
|
def test_eta_unknown_does_not_extend(self):
|
|
self.eta_task()
|
|
nid = self.eta_notice()['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=1310):
|
|
task = self.store.eta_complete(nid,dict(result='unknown',reason='Waiting on real evidence'))
|
|
self.assertEqual(task['eta_review_status'],'unknown')
|
|
self.assertEqual(task['eta_estimated_at'],1000)
|
|
self.assertEqual(task['eta_revision'],1)
|
|
self.assertEqual(task['eta_max_minutes'],10)
|
|
|
|
def test_eta_extension_limit_persists_and_cannot_be_bypassed_by_task_patch(self):
|
|
task = self.eta_task()
|
|
for index in range(3):
|
|
checkpoint = task['eta_estimated_at'] + task['eta_max_minutes']*30
|
|
nid = self.eta_notice(checkpoint)['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=checkpoint+1):
|
|
task = self.store.eta_complete(nid,dict(result='revise',reason='Measured additional work',
|
|
eta_min_minutes=5,eta_max_minutes=20+index*10))
|
|
self.assertEqual(task['eta_extension_count'],3)
|
|
old = dict(task)
|
|
checkpoint = task['eta_estimated_at'] + task['eta_max_minutes']*30
|
|
nid = self.eta_notice(checkpoint)['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=checkpoint+1):
|
|
with self.assertRaises(Conflict):
|
|
self.store.eta_complete(nid,dict(result='revise',reason='Fourth extension',
|
|
eta_min_minutes=5,eta_max_minutes=50))
|
|
with self.assertRaises(Conflict):
|
|
self.store.task(dict(eta_max_minutes=50,eta_change_reason='Try task patch'),'eta')
|
|
task = Store(self.path).state()['tasks'][0]
|
|
self.assertEqual(task['eta_review_status'],'needs_decision')
|
|
self.assertEqual(task['eta_revision'],old['eta_revision'])
|
|
self.assertEqual(task['eta_estimated_at'],old['eta_estimated_at'])
|
|
self.assertEqual(task['eta_max_minutes'],old['eta_max_minutes'])
|
|
self.assertEqual(task['eta_extension_count'],3)
|
|
self.store.eta_tick(checkpoint+10000)
|
|
self.assertEqual(self.store.state()['tasks'][0]['eta_review_status'],'needs_decision')
|
|
|
|
def test_eta_same_range_revise_and_unknown_cannot_clear_decision_limit(self):
|
|
self.eta_task()
|
|
nid = self.eta_notice()['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=1400):
|
|
task = self.store.eta_complete(nid,dict(result='revise',reason='Same measured range',
|
|
eta_min_minutes=5,eta_max_minutes=10))
|
|
self.assertEqual(task['eta_revision'],1)
|
|
self.assertEqual(task['eta_estimated_at'],1000)
|
|
self.assertEqual(task['eta_review_status'],'reviewed')
|
|
self.store.eta_tick(1480)
|
|
nid = self.store.eta_notifications()['notifications'][0]['notification_id']
|
|
with self.store.db() as db:
|
|
task['eta_review_status']='needs_decision'
|
|
self.store._put(db,'tasks',task)
|
|
task = self.store.eta_complete(nid,dict(result='unknown',reason='Still awaiting evidence'))
|
|
self.assertEqual(task['eta_review_status'],'needs_decision')
|
|
|
|
def test_eta_invalid_changes_zero_and_managed_fields(self):
|
|
with self.assertRaises(Invalid): self.eta_task(eta_min_minutes=0,eta_max_minutes=0)
|
|
self.eta_task()
|
|
for data in [dict(eta_max_minutes=11),dict(eta_max_minutes=11,eta_change_reason=''),
|
|
dict(eta_revision=9),dict(eta_estimated_at=1),dict(eta_extension_count=0),
|
|
dict(eta_max_minutes=True),dict(eta_max_minutes=float('inf'))]:
|
|
with self.assertRaises(Invalid): self.store.task(data,'eta')
|
|
self.assertEqual(self.store.state()['tasks'][0]['eta_revision'],1)
|
|
task = self.store.task(dict(status='completed',phase='done',eta_min_minutes=0,
|
|
eta_max_minutes=0,eta_change_reason='Checks finished'),'eta')
|
|
self.assertEqual(task['eta_max_minutes'],0)
|
|
self.assertEqual(self.store.eta_tick(100000)['created'],0)
|
|
|
|
def test_eta_legacy_and_no_estimate_do_not_invent_start(self):
|
|
self.store.task(dict(id='none',name='No estimate'))
|
|
with self.store.db() as db:
|
|
db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('legacy',json.dumps(
|
|
dict(id='legacy',name='Legacy',status='running',eta_min_minutes=5,eta_max_minutes=10,
|
|
eta_basis='Legacy explanation'))))
|
|
self.assertEqual(self.store.eta_tick(100000)['created'],0)
|
|
task = next(t for t in self.store.state()['tasks'] if t['id']=='legacy')
|
|
self.assertIsNone(task['eta_estimated_at'])
|
|
self.assertEqual(task['eta_revision'],0)
|
|
with patch('agent_control.store.time.time',return_value=2000):
|
|
task = self.store.task(dict(current_progress='Reading baseline'),'legacy')
|
|
self.assertIsNone(task['eta_estimated_at'])
|
|
with patch('agent_control.store.time.time',return_value=2001):
|
|
task = self.store.task(dict(eta_min_minutes=5,eta_max_minutes=10),'legacy')
|
|
self.assertEqual(task['eta_revision'],1)
|
|
self.assertEqual(task['eta_estimated_at'],2001)
|
|
|
|
def test_eta_finished_and_revised_work_does_not_notify_old_reviews(self):
|
|
for index,status in enumerate(['completed','cancelled','failed']):
|
|
tid = 'finished-'+str(index)
|
|
self.eta_task(id=tid)
|
|
notices = self.store.eta_tick(1300)
|
|
notice = next(n for n in self.store.eta_notifications()['notifications'] if n['task']['id']==tid)
|
|
self.store.task(dict(status=status,phase='done'),'finished-'+str(index))
|
|
self.assertFalse(any(n['task']['id']==tid for n in self.store.eta_notifications()['notifications']))
|
|
with self.assertRaises(Conflict):
|
|
self.store.eta_complete(notice['notification_id'],dict(result='keep',reason='Late result'))
|
|
self.assertEqual(self.store.eta_tick(100000)['created'],0)
|
|
|
|
def test_eta_invalid_completion_does_not_mutate(self):
|
|
self.eta_task()
|
|
nid = self.eta_notice()['notification_id']
|
|
for body in [dict(result='keep',reason=''),dict(result='bad',reason='Reason'),
|
|
dict(result='keep',reason='Reason',eta_max_minutes=10),
|
|
dict(result='revise',reason='Reason',eta_max_minutes=20),
|
|
dict(result='revise',reason='Reason',eta_min_minutes=0,eta_max_minutes=0),
|
|
dict(result='revise',reason='Reason',eta_min_minutes=None,eta_max_minutes=None),
|
|
dict(result='unknown',reason='Reason',current_progress=None),
|
|
dict(result='keep',reason='Reason',authorization=True)]:
|
|
with self.subTest(body=body):
|
|
with self.assertRaises(Invalid): self.store.eta_complete(nid,body)
|
|
self.assertEqual(self.store.state()['tasks'][0]['eta_revision'],1)
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],1)
|
|
|
|
def test_eta_concurrent_ticks_and_completions(self):
|
|
self.eta_task()
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
list(pool.map(lambda _:self.store.eta_tick(1300),range(12)))
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],1)
|
|
nid = self.store.eta_notifications()['notifications'][0]['notification_id']
|
|
body = dict(result='keep',reason='Measured progress')
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(lambda _:self.store.eta_complete(nid,body),range(12)))
|
|
self.assertTrue(all(r==results[0] for r in results))
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],0)
|
|
|
|
def test_eta_background_clock_without_state_poll_and_stop(self):
|
|
self.eta_task()
|
|
seen = threading.Event()
|
|
original = self.store.eta_tick
|
|
def tick():
|
|
original(1300)
|
|
seen.set()
|
|
with patch.object(self.store,'eta_tick',side_effect=tick):
|
|
clock = EtaClock(self.store,interval=0.01)
|
|
try:
|
|
self.assertTrue(seen.wait(2))
|
|
self.assertEqual(clock.interval,0.01)
|
|
self.assertEqual(self.store.eta_notifications()['pending_count'],1)
|
|
finally:
|
|
clock.close()
|
|
self.assertFalse(clock.thread.is_alive())
|
|
|
|
def test_empty_and_legacy_task(self):
|
|
self.assertEqual(self.store.state()['decisions'],[])
|
|
self.assertEqual(self.store.decision_notifications(),dict(notifications=[],pending_count=0))
|
|
self.store.session(dict(id='s',name='Synthetic'))
|
|
self.store.task(dict(id='t',name='Task',session_id='s',decision='Legacy question'))
|
|
self.store.decision(definition(task_id='t'))
|
|
self.assertEqual(self.store.state()['tasks'][0]['decision'],'Legacy question')
|
|
|
|
def test_normalized_creation_and_immutable_definition(self):
|
|
first = self.store.decision(definition())
|
|
self.assertEqual(first['status'],'open')
|
|
self.assertIsNone(first['response'])
|
|
explicit = definition(impact='',priority='normal',details='',recommendation='',
|
|
deadline=None,requires_authorization=False,task_id=None)
|
|
self.assertEqual(first,self.store.decision(explicit))
|
|
with self.assertRaises(Conflict):
|
|
self.store.decision(definition(question='Changed'))
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],0)
|
|
|
|
def test_persistence_response_dedup_and_ack(self):
|
|
self.store.decision(definition())
|
|
with patch('subprocess.run',side_effect=AssertionError('must not execute')):
|
|
first = self.store.decision_respond('choice',answer())
|
|
reopened = Store(self.path)
|
|
self.assertEqual(first,reopened.decision_respond('choice',answer()))
|
|
self.assertEqual(first,reopened.decision(definition()))
|
|
self.assertEqual(first['status'],'decided')
|
|
self.assertFalse(first['response']['authorization'])
|
|
self.assertEqual(first['response']['comment'],answer()['comment'])
|
|
for changes in [dict(comment='changed'),dict(idempotency_key='other'),dict(option_id=None)]:
|
|
with self.assertRaises(Conflict):
|
|
reopened.decision_respond('choice',answer(**changes))
|
|
queue = reopened.decision_notifications()
|
|
self.assertEqual(queue['pending_count'],1)
|
|
notice = queue['notifications'][0]
|
|
self.assertEqual(set(notice),{'notification_id','target','decision'})
|
|
self.assertEqual(notice['target'],'control')
|
|
self.assertEqual(notice['decision'],first)
|
|
ack = dict(notification_id=notice['notification_id'])
|
|
self.assertFalse(reopened.decision_ack(ack)['duplicate'])
|
|
self.assertTrue(Store(self.path).decision_ack(ack)['duplicate'])
|
|
self.assertEqual(reopened.decision_notifications()['pending_count'],0)
|
|
reopened.decision_respond('choice',answer())
|
|
self.assertEqual(reopened.decision_notifications()['pending_count'],0)
|
|
|
|
def test_authorization_is_only_pending_verification(self):
|
|
self.store.decision(definition(requires_authorization=True,options=[]))
|
|
value = self.store.decision_respond('choice',answer(option_id=None,comment='Approve sudo'))
|
|
self.assertEqual(value['status'],'pending_verification')
|
|
self.assertIs(value['response']['authorization'],False)
|
|
|
|
def test_invalid_definitions(self):
|
|
for changes in [dict(shell='x'),dict(title=''),dict(question=None),dict(impact=None),
|
|
dict(priority=[]),dict(priority='critical'),dict(requires_authorization=1),
|
|
dict(options={}),dict(options=[dict(id='a',label='a')]),
|
|
dict(options=[dict(id='a',label='a',description='')]*2),
|
|
dict(deadline='2026-02-30T01:00:00Z'),dict(deadline='tomorrow'),
|
|
dict(task_id='missing'),dict(details='x'*8001)]:
|
|
with self.subTest(changes=changes):
|
|
with self.assertRaises(Invalid): self.store.decision(definition(**changes))
|
|
self.assertEqual(self.store.state()['decisions'],[])
|
|
self.store.decision(definition(deadline='2026-10-19T13:00:00+08:00'))
|
|
|
|
def test_invalid_answers_and_ack(self):
|
|
self.store.decision(definition())
|
|
for changes in [dict(option_id=None),dict(option_id='bad'),dict(comment=None),
|
|
dict(comment='x'*8001),dict(idempotency_key=''),dict(authorization=True)]:
|
|
with self.subTest(changes=changes):
|
|
with self.assertRaises(Invalid): self.store.decision_respond('choice',answer(**changes))
|
|
with self.assertRaises(Invalid): self.store.decision_respond('missing',answer())
|
|
with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing'))
|
|
with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing',target='human'))
|
|
self.assertEqual(self.store.state()['decisions'][0]['status'],'open')
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],0)
|
|
self.store.decision(definition(id='free',options=[]))
|
|
for option,comment in [(None,' '),('a','text')]:
|
|
with self.assertRaises(Invalid):
|
|
self.store.decision_respond('free',answer(option_id=option,comment=comment))
|
|
|
|
def test_concurrent_same_key_has_one_response_and_notification(self):
|
|
self.store.decision(definition())
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
values = list(pool.map(lambda _:self.store.decision_respond('choice',answer()),range(16)))
|
|
self.assertTrue(all(v == values[0] for v in values))
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],1)
|
|
|
|
def test_concurrent_distinct_answers_have_one_winner(self):
|
|
self.store.decision(definition())
|
|
def respond(index):
|
|
try:
|
|
return self.store.decision_respond('choice',answer(idempotency_key=str(index)))
|
|
except Conflict:
|
|
return None
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
values = list(pool.map(respond,range(16)))
|
|
self.assertEqual(sum(v is not None for v in values),1)
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],1)
|
|
|
|
def test_atomic_rollback_on_outbox_failure(self):
|
|
self.store.decision(definition())
|
|
with self.store.db() as db:
|
|
db.execute("CREATE TRIGGER fail_notice BEFORE INSERT ON decision_notifications BEGIN SELECT RAISE(ABORT,'synthetic'); END")
|
|
with self.assertRaises(Exception): self.store.decision_respond('choice',answer())
|
|
self.assertEqual(self.store.state()['decisions'][0]['status'],'open')
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],0)
|
|
|
|
def test_pending_order_and_limit(self):
|
|
for index in range(103):
|
|
did = 'choice-'+str(index)
|
|
self.store.decision(definition(id=did))
|
|
self.store.decision_respond(did,answer())
|
|
queue = self.store.decision_notifications()
|
|
self.assertEqual(queue['pending_count'],103)
|
|
self.assertEqual(len(queue['notifications']),100)
|
|
self.assertEqual(queue['notifications'][0]['decision']['id'],'choice-0')
|
|
self.assertEqual(queue['notifications'][-1]['decision']['id'],'choice-99')
|
|
|
|
|
|
class DecisionHTTPTests(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_and_security(self):
|
|
create='/api/decisions'
|
|
respond='/api/decisions/choice/respond'
|
|
notifications='/api/decision-notifications'
|
|
ack=notifications+'/ack'
|
|
for path,body in [(create,definition()),(respond,answer()),(ack,dict(notification_id='missing'))]:
|
|
for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403),
|
|
({'Origin':'https://untrusted.example'},403)]:
|
|
self.assertEqual(self.request(path,'POST',body,headers)[0],status)
|
|
self.assertEqual(self.request(notifications,headers={'Authorization':''})[0],401)
|
|
self.assertEqual(self.request('/api/state')[1]['decisions'],[])
|
|
self.assertEqual(self.request(create,'POST',definition())[0],200)
|
|
self.assertEqual(self.request(create,'POST',definition(question='different'))[0],409)
|
|
self.assertEqual(self.request(respond,'POST',answer(option_id='bad'))[0],400)
|
|
code,value=self.request(respond,'POST',answer())
|
|
self.assertEqual(code,200)
|
|
self.assertEqual(self.request(respond,'POST',answer())[1],value)
|
|
self.assertEqual(self.request(respond,'POST',answer(idempotency_key='new'))[0],409)
|
|
notice=self.request(notifications)[1]['notifications'][0]
|
|
self.assertEqual(self.request(ack,'POST',dict(notification_id=notice['notification_id']))[0],200)
|
|
self.assertEqual(self.request(notifications)[1]['pending_count'],0)
|
|
self.assertEqual(self.request(notifications+'?target=human')[0],404)
|
|
self.assertEqual(self.request(respond+'?x=1','POST',answer())[0],404)
|
|
self.assertEqual(self.request('/api/decisions/choice','PATCH',definition())[0],404)
|
|
self.assertEqual(self.request('/api/state')[1]['decisions'][0],value)
|
|
|
|
def test_cross_site_only_public_top_level_navigation(self):
|
|
web = Path(self.tmp.name)/'web'
|
|
web.mkdir()
|
|
(web/'index.html').write_text('<html>Public login</html>')
|
|
self.server.web_root = web
|
|
navigation = {'Sec-Fetch-Site':'cross-site','Sec-Fetch-Mode':'navigate','Sec-Fetch-Dest':'document'}
|
|
conn = http.client.HTTPConnection(*self.server.server_address,timeout=3)
|
|
conn.request('GET','/',headers=navigation)
|
|
response = conn.getresponse()
|
|
self.assertEqual(response.status,200)
|
|
self.assertIn(b'Public login',response.read())
|
|
conn.close()
|
|
self.assertEqual(self.request('/api/state',headers=navigation)[0],403)
|
|
self.assertEqual(self.request('/api/decisions','POST',definition(),navigation)[0],403)
|
|
self.assertEqual(self.request('/','POST',definition(),navigation)[0],403)
|
|
for overrides in [{'Sec-Fetch-Mode':'no-cors','Sec-Fetch-Dest':'script'},
|
|
{'Sec-Fetch-Dest':'iframe'},{'Sec-Fetch-Mode':'cors'},
|
|
{'Origin':'https://untrusted.example'},{'Host':'untrusted.example'}]:
|
|
self.assertEqual(self.request('/',headers=dict(navigation,**overrides))[0],403)
|
|
|
|
def test_eta_http_security_ack_and_complete(self):
|
|
self.server.eta_clock.close()
|
|
with patch('agent_control.store.time.time',return_value=1000):
|
|
self.server.store.task(dict(id='http-eta',name='ETA',status='running',
|
|
eta_min_minutes=5,eta_max_minutes=10,eta_basis='Measured'))
|
|
self.server.store.eta_tick(1300)
|
|
self.assertEqual(self.request('/api/eta-notifications',headers={'Authorization':''})[0],401)
|
|
self.assertEqual(self.request('/api/eta-notifications',headers={'Sec-Fetch-Site':'cross-site'})[0],403)
|
|
self.assertEqual(self.request('/api/eta-notifications?x=1')[0],404)
|
|
notice=self.request('/api/eta-notifications')[1]['notifications'][0]
|
|
nid=notice['notification_id']
|
|
ack='/api/eta-notifications/ack'
|
|
complete='/api/eta-reviews/'+nid+'/complete'
|
|
body=dict(result='keep',reason='Evidence received')
|
|
self.assertEqual(self.request(ack,'POST',dict(notification_id=nid),{'X-CSRF-Token':''})[0],403)
|
|
self.assertEqual(self.request(complete,'POST',body,{'Origin':'https://untrusted.example'})[0],403)
|
|
self.assertEqual(self.request(ack,'POST',dict(notification_id=nid))[0],200)
|
|
self.assertEqual(self.request(complete,'POST',body)[0],200)
|
|
self.assertEqual(self.request(complete,'POST',dict(body,reason='Changed'))[0],409)
|
|
self.assertEqual(self.request('/api/eta-notifications')[1]['pending_count'],0)
|
|
self.assertEqual(self.server.eta_clock.interval,15)
|
|
self.assertFalse(self.server.eta_clock.thread.is_alive())
|
|
|
|
def test_server_starts_clock_without_http_polling(self):
|
|
self.assertIsInstance(self.server.eta_clock,EtaClock)
|
|
self.assertTrue(self.server.eta_clock.thread.is_alive())
|
|
self.assertTrue(self.server.eta_clock.thread.name=='eta-clock')
|