112 lines
6.3 KiB
Python
112 lines
6.3 KiB
Python
import copy
|
|
import hashlib
|
|
import json
|
|
import types
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from scripts.tests import test_c01_network_v4 as v4
|
|
import c01_network_plan as plan
|
|
import c01_network_owner as owner
|
|
import c01_network_client as client
|
|
|
|
|
|
class MappedPrincipalTests(unittest.TestCase):
|
|
def fixture(self, principal=(100100,100101)):
|
|
policy,runtime=v4.V4Tests().fixture()
|
|
doc=policy['manifests'][0]
|
|
doc['producer']={'kind':'container-payload','container_name':'fixture','image_id':'b'*64,
|
|
'payload_principal':list(principal)}
|
|
policy['routes'][0]['target_uid']=principal[0]
|
|
self.rehash(policy)
|
|
return policy,runtime
|
|
|
|
def rehash(self, policy):
|
|
policy['routes'][0]['projection_sha256']=hashlib.sha256(json.dumps(policy['manifests'][0],sort_keys=True,separators=(',',':')).encode()).hexdigest()
|
|
|
|
def test_explicit_mapped_principal_is_accepted(self):
|
|
policy,runtime=self.fixture()
|
|
self.assertTrue(plan.validate(policy,runtime)['structural_validation'])
|
|
|
|
def test_route_cannot_substitute_runtime_or_other_uid(self):
|
|
for uid in (0,1000,100102):
|
|
policy,runtime=self.fixture();policy['routes'][0]['target_uid']=uid
|
|
with self.subTest(uid=uid),self.assertRaisesRegex(ValueError,'principal/hierarchy'):
|
|
plan.validate(policy,runtime)
|
|
|
|
def test_invalid_or_root_principal_rejected(self):
|
|
for principal in ([0,1000],[1000,0],[True,1000],[1000],[-1,1000],[2**32-1,1000],['1000',1000]):
|
|
policy,runtime=self.fixture()
|
|
policy['manifests'][0]['producer']['payload_principal']=principal;self.rehash(policy)
|
|
with self.subTest(principal=principal),self.assertRaisesRegex(ValueError,'mapped payload principal'):
|
|
plan.validate(policy,runtime)
|
|
|
|
def test_legacy_selector_does_not_gain_mapped_uid_authority(self):
|
|
policy,runtime=self.fixture();del policy['manifests'][0]['producer']['payload_principal'];self.rehash(policy)
|
|
with self.assertRaisesRegex(ValueError,'principal/hierarchy'):plan.validate(policy,runtime)
|
|
policy['routes'][0]['target_uid']=1000
|
|
self.assertTrue(plan.validate(policy,runtime)['structural_validation'])
|
|
|
|
def observation(self):
|
|
value,runtime=self.fixture();route=value['routes'][0];doc=value['manifests'][0]
|
|
policy=types.SimpleNamespace(manifests={route['projection_sha256']:doc},runtime=runtime)
|
|
actor={'pid':200,'group':route['target_cgroup']+'/libpod-payload-'+'1'*64,'uids':[100100]*4,'gids':[100101]*4}
|
|
initial={**actor,'pid':100}
|
|
config={'name':'fixture','rootfsImageID':'b'*64,'id':'1'*64}
|
|
return policy,route,actor,initial,config
|
|
|
|
def observe(self, values, namespaces=None):
|
|
policy,route,actor,initial,config=values
|
|
connection=mock.Mock()
|
|
connection.execute.return_value.fetchone.side_effect=[(json.dumps(config),),(json.dumps({'pid':100,'state':3}),)]
|
|
with mock.patch('sqlite3.connect',return_value=connection),mock.patch.object(owner,'process',return_value=initial),mock.patch.object(owner.os,'readlink',side_effect=namespaces or ['pid:[1]','pid:[1]','user:[2]','user:[2]']):
|
|
return owner.producer_group(policy,route,actor)
|
|
|
|
def test_live_binding_requires_matching_init_and_socket_principals(self):
|
|
values=self.observation()
|
|
self.assertEqual(self.observe(values),values[2]['group'])
|
|
for position in (2,3):
|
|
for field in ('uids','gids'):
|
|
changed=copy.deepcopy(values);changed[position][field]=[1000]*4
|
|
with self.subTest(position=position,field=field),self.assertRaisesRegex(ValueError,'mapped payload principal differs'):
|
|
self.observe(changed)
|
|
|
|
def test_live_binding_rejects_foreign_user_namespace(self):
|
|
with self.assertRaisesRegex(ValueError,'user namespace differs'):
|
|
self.observe(self.observation(),['pid:[1]','pid:[1]','user:[2]','user:[3]'])
|
|
|
|
def test_live_binding_still_rejects_foreign_image(self):
|
|
values=self.observation();values[4]['rootfsImageID']='c'*64
|
|
with self.assertRaisesRegex(ValueError,'configuration identity differs'):self.observe(values)
|
|
|
|
def test_lease_binds_runtime_account_instead_of_rejecting_mapped_payload(self):
|
|
value,runtime=self.fixture();plan.validate(value,runtime)
|
|
for uid in (1000,100100):
|
|
policy=types.SimpleNamespace(value=value,runtime=runtime,sha='a'*64,close=mock.Mock())
|
|
channel=mock.Mock();channel.connect.side_effect=OSError('fixture connection boundary')
|
|
with mock.patch.object(client,'Policy',return_value=policy),mock.patch.object(client.socket,'socket',return_value=channel):
|
|
if uid==1000:
|
|
with self.assertRaisesRegex(OSError,'fixture connection boundary'):client.NetworkLease(runtime_uid=uid)
|
|
channel.connect.assert_called_once()
|
|
else:
|
|
with self.assertRaisesRegex(ValueError,'network runtime appointment'):client.NetworkLease(runtime_uid=uid)
|
|
channel.connect.assert_not_called()
|
|
|
|
def test_management_only_legacy_policy_needs_no_runtime_appointment(self):
|
|
for version in (1,2,3,4):
|
|
policy=types.SimpleNamespace(value={'version':version,'routes':[{'target':'managed'}]},runtime=None,sha='a'*64,close=mock.Mock())
|
|
channel=mock.Mock();channel.connect.side_effect=OSError('fixture connection boundary')
|
|
with self.subTest(version=version),mock.patch.object(client,'Policy',return_value=policy),mock.patch.object(client.socket,'socket',return_value=channel):
|
|
with self.assertRaisesRegex(OSError,'fixture connection boundary'):client.NetworkLease(runtime_uid=1000)
|
|
channel.connect.assert_called_once()
|
|
|
|
def test_runtime_route_without_appointment_still_fails_closed(self):
|
|
policy=types.SimpleNamespace(value={'version':4,'routes':[{'target':'runtime'}]},runtime=None,sha='a'*64,close=mock.Mock())
|
|
channel=mock.Mock()
|
|
with mock.patch.object(client,'Policy',return_value=policy),mock.patch.object(client.socket,'socket',return_value=channel):
|
|
with self.assertRaisesRegex(ValueError,'network runtime appointment'):client.NetworkLease(runtime_uid=1000)
|
|
channel.connect.assert_not_called()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|