46 lines
2.6 KiB
Python
46 lines
2.6 KiB
Python
"""Finite kiosk delegation validation and generation-aware re-admission."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
import unittest
|
|
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
import c01_kiosk_endpoints as kiosk
|
|
|
|
|
|
class DelegationTests(unittest.TestCase):
|
|
def test_live_producer_is_never_replaced(self):
|
|
name='app/archipelago-kiosk/tcp/9222/managed'
|
|
policy=SimpleNamespace(sha='a'*64,check=lambda:None,value={'version':3})
|
|
value={'routes':[name]}
|
|
response={'policy_sha256':policy.sha,'name':name,'generation':'b'*64,'producer':{'pid':123}}
|
|
with patch.object(kiosk.network,'protected_read',return_value=b'x'),patch.object(kiosk,'request',return_value=response) as request,patch.object(kiosk,'discover') as discover:
|
|
kiosk.tick(policy,value,kiosk.network.digest(b'x'),{})
|
|
self.assertEqual(request.call_count,1)
|
|
discover.assert_not_called()
|
|
|
|
def test_retired_generation_token_is_explicitly_reused(self):
|
|
name='app/archipelago-kiosk/tcp/9222/managed';policy=SimpleNamespace(sha='a'*64,check=lambda:None,value={'version':3})
|
|
responses=[{'policy_sha256':policy.sha,'name':name,'generation':'b'*64,'producer':None},
|
|
{'policy_sha256':policy.sha,'registered':name,'generation':'c'*64}]
|
|
with patch.object(kiosk.network,'protected_read',return_value=b'x'),patch.object(kiosk,'request',side_effect=responses) as request,patch.object(kiosk,'discover',return_value=(123,45)):
|
|
kiosk.tick(policy,{'routes':[name]},kiosk.network.digest(b'x'),{})
|
|
self.assertEqual(request.call_args.args[1],{'method':'register','name':name,'pid':123,'fd':45,'previous':'b'*64})
|
|
|
|
def test_policy_drift_refuses_before_discovery_or_registration(self):
|
|
policy=SimpleNamespace(sha='a'*64,check=lambda:None)
|
|
with patch.object(kiosk.network,'protected_read',return_value=b'changed'),patch.object(kiosk,'request') as request:
|
|
with self.assertRaisesRegex(ValueError,'delegation changed'):kiosk.tick(policy,{'routes':[]},'old',{})
|
|
request.assert_not_called()
|
|
|
|
def test_arbitrary_route_cannot_be_delegated(self):
|
|
value=dict(version=1,policy_sha256='a'*64,executables={},routes=['management'])
|
|
policy=SimpleNamespace(sha='a'*64)
|
|
with patch.object(kiosk.network,'protected_read',return_value=json.dumps(value).encode()):
|
|
with self.assertRaisesRegex(ValueError,'finite kiosk routes'):kiosk.load(policy)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|