62 lines
3.5 KiB
Python
62 lines
3.5 KiB
Python
"""Migration approval parsing; no live services, keys or financial operations."""
|
|
import copy
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
import tempfile
|
|
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
from c01_migration import validate, refuse_existing_legacy_anchor
|
|
|
|
|
|
class MigrationApprovalTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.launch={'binding':{'incarnation':[3]*32,'local_identity':[1]*32,'reader_schema':2,'writer_schema':2},
|
|
'manager_backing':'/private/manager','runtime_uid':1000,'runtime_gid':1000}
|
|
self.value={'version':1,'operation':'import-legacy-requests-v1',
|
|
'authorization':'one-qualified-guarded-framework-migration',
|
|
'launch_sha256':'a'*64,'access_sha256':'b'*64,'binding':copy.deepcopy(self.launch['binding']),
|
|
'source_public_sha256':'c'*64,'approved_watermark':1793680000,
|
|
'source':{'path':'/private/manager/federation/pending_requests.json','device':3,'inode':19,
|
|
'uid':1000,'gid':1000,'bytes':1024,'sha256':'d'*64}}
|
|
|
|
def check(self,value):validate(value,self.launch,'a'*64,'b'*64)
|
|
|
|
def test_explicit_bound_legacy_appointment(self):
|
|
before=copy.deepcopy(self.value);self.check(self.value);self.assertEqual(before,self.value)
|
|
|
|
def test_no_default_floor_or_foreign_scope(self):
|
|
for field,value in (('approved_watermark',0),('approved_watermark',True),
|
|
('approved_watermark',2**64),('authorization','yaya-migration'),
|
|
('operation','inspect'),('launch_sha256','e'*64),('access_sha256','e'*64)):
|
|
with self.subTest(field=field),self.assertRaises(ValueError):self.check(dict(self.value,**{field:value}))
|
|
changed=copy.deepcopy(self.value);changed['binding']['incarnation']=[4]*32
|
|
with self.assertRaises(ValueError):self.check(changed)
|
|
|
|
def test_cannot_select_other_data_or_silently_accept_anchor_reset_fields(self):
|
|
for field,value in (('path','/var/lib/archipelago/wallet'),('uid',0),('bytes',0),
|
|
('bytes',4*1024*1024+1),('inode',True),('watermark',0),('incarnation',[0]*32)):
|
|
changed=copy.deepcopy(self.value);changed['source'][field]=value
|
|
with self.subTest(field=field),self.assertRaises(ValueError):self.check(changed)
|
|
|
|
def test_stale_plaintext_never_supersedes_existing_or_aliased_protected_history(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root=Path(temporary)
|
|
launch={'manager_backing':str(root/'manager'),'runtime_data_dir':str(root/'runtime')}
|
|
refuse_existing_legacy_anchor(launch)
|
|
anchor=root/'manager/federation/protected-pending'
|
|
anchor.mkdir(parents=True)
|
|
control=anchor/'control';control.write_bytes(b'retained control sentinel')
|
|
before=control.stat()
|
|
with self.assertRaisesRegex(ValueError,'retained-key recovery'):
|
|
refuse_existing_legacy_anchor(launch)
|
|
self.assertEqual(control.stat().st_ino,before.st_ino)
|
|
self.assertEqual(control.read_bytes(),b'retained control sentinel')
|
|
other=root/'runtime/federation';other.mkdir(parents=True)
|
|
(other/'protected-pending').symlink_to(root/'absent')
|
|
with self.assertRaises(ValueError):
|
|
refuse_existing_legacy_anchor({'manager_backing':str(root/'unused'),'runtime_data_dir':str(root/'runtime')})
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|