acapy/acapy_agent/admin/tests/test_auth.py at c8491839927849be1fbc65b2724de0cd18f382e6 · openwallet-foundation/acapy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
from unittest import IsolatedAsyncioTestCase

from aiohttp import web

from ...tests import mock
from ...utils.testing import create_test_profile
from ..decorators.auth import admin_authentication, tenant_authentication
from ..request_context import AdminRequestContext


class TestAdminAuthentication(IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        self.profile = await create_test_profile(
            settings={
                "admin.admin_api_key": "admin_api_key",
                "admin.admin_insecure_mode": False,
            }
        )
        self.context = AdminRequestContext.test_context({}, self.profile)
        self.request_dict = {
            "context": self.context,
        }
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k], headers={}, method="POST"
        )
        self.decorated_handler = mock.CoroutineMock()

    async def test_options_request(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k], headers={}, method="OPTIONS"
        )
        decor_func = admin_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_insecure_mode(self):
        self.profile.settings["admin.admin_insecure_mode"] = True
        decor_func = admin_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_invalid_api_key(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "wrong-key"},
            method="POST",
        )
        decor_func = admin_authentication(self.decorated_handler)
        with self.assertRaises(web.HTTPUnauthorized):
            await decor_func(self.request)

    async def test_valid_api_key(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key"},
            method="POST",
        )
        decor_func = admin_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)


class TestTenantAuthentication(IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        self.profile = await create_test_profile(
            settings={
                "admin.admin_api_key": "admin_api_key",
                "admin.admin_insecure_mode": False,
                "multitenant.enabled": True,
            }
        )
        self.context = AdminRequestContext.test_context({}, self.profile)
        self.request_dict = {
            "context": self.context,
        }
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k], headers={}, method="POST"
        )
        self.decorated_handler = mock.CoroutineMock()

    async def test_options_request(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k], headers={}, method="OPTIONS"
        )
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_insecure_mode(self):
        self.profile.settings["admin.admin_insecure_mode"] = True
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_single_tenant_invalid_api_key(self):
        self.profile.settings["multitenant.enabled"] = False
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "wrong-key"},
            method="POST",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        with self.assertRaises(web.HTTPUnauthorized):
            await decor_func(self.request)

    async def test_single_tenant_valid_api_key(self):
        self.profile.settings["multitenant.enabled"] = False
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key"},
            method="POST",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_multi_tenant_missing_auth_header(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "wrong-key"},
            method="POST",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        with self.assertRaises(web.HTTPUnauthorized):
            await decor_func(self.request)

    async def test_multi_tenant_valid_auth_header(self):
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key", "Authorization": "Bearer my-jwt"},
            method="POST",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_base_wallet_additional_route_allowed_string(self):
        self.profile.settings["multitenant.base_wallet_routes"] = (
            "/not-this-route /extra-route"
        )
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key"},
            method="POST",
            path="/extra-route",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_base_wallet_additional_route_allowed_list(self):
        self.profile.settings["multitenant.base_wallet_routes"] = [
            "/extra-route",
            "/not-this-route",
        ]
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key"},
            method="POST",
            path="/extra-route",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        await decor_func(self.request)
        self.decorated_handler.assert_called_once_with(self.request)

    async def test_base_wallet_additional_route_denied(self):
        self.profile.settings["multitenant.base_wallet_routes"] = "/extra-route"
        self.request = mock.MagicMock(
            __getitem__=lambda _, k: self.request_dict[k],
            headers={"x-api-key": "admin_api_key"},
            method="POST",
            path="/extra-route-wrong",
        )
        decor_func = tenant_authentication(self.decorated_handler)
        with self.assertRaises(web.HTTPUnauthorized):
            await decor_func(self.request)