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()