-
Notifications
You must be signed in to change notification settings - Fork 166
Expand file tree
/
Copy pathtest_common.py
More file actions
266 lines (211 loc) · 10.4 KB
/
Copy pathtest_common.py
File metadata and controls
266 lines (211 loc) · 10.4 KB
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
from unittest.mock import MagicMock, patch
import pytest
from meraki.common import (
check_python_version,
validate_user_agent,
reject_v0_base_url,
validate_base_url,
iterator_for_get_pages_bool,
use_iterator_for_get_pages_setter,
)
from meraki.exceptions import PythonVersionError, SessionInputError
class TestCheckPythonVersion:
@patch("platform.python_version_tuple", return_value=("3", "11", "0"))
def test_valid_version(self, mock_ver):
check_python_version()
@patch("platform.python_version_tuple", return_value=("3", "9", "0"))
def test_too_old_raises(self, mock_ver):
with pytest.raises(PythonVersionError):
check_python_version()
@patch("platform.python_version_tuple", return_value=("2", "7", "0"))
def test_python2_raises(self, mock_ver):
with pytest.raises(PythonVersionError):
check_python_version()
def test_check_python_version_valid_does_not_raise(self):
"""check_python_version should not raise on current interpreter (>=3.10)."""
check_python_version()
def test_check_python_version_rejects_39(self):
"""check_python_version raises PythonVersionError for 3.9."""
with patch("platform.python_version_tuple", return_value=("3", "9", "0")):
with pytest.raises(PythonVersionError):
check_python_version()
def test_check_python_version_accepts_310(self):
"""check_python_version accepts exactly 3.10.0 (minimum)."""
with patch("platform.python_version_tuple", return_value=("3", "10", "0")):
check_python_version()
class TestValidateUserAgent:
def test_valid_caller(self):
result = validate_user_agent("", "TestApp TestVendor")
assert "TestApp TestVendor" in result
assert result.startswith("Caller/(")
def test_valid_caller_with_version(self):
result = validate_user_agent("", "MyApp/1.0 MyVendor")
assert "MyApp/1.0 MyVendor" in result
def test_invalid_caller_raises(self):
with pytest.raises(SessionInputError):
validate_user_agent("", "invalid format!!!")
def test_be_geo_id_fallback(self):
result = validate_user_agent("GeoApp GeoVendor", "")
assert "GeoApp GeoVendor" in result
def test_invalid_be_geo_id_raises(self):
with pytest.raises(SessionInputError):
validate_user_agent("bad format!!!", "")
def test_unidentified_when_both_empty(self):
result = validate_user_agent("", "")
assert "unidentified" in result
class TestRejectV0BaseUrl:
def test_v0_exits(self):
session = MagicMock()
session._base_url = "https://api.meraki.com/api/v0"
with pytest.raises(SystemExit):
reject_v0_base_url(session)
def test_v1_ok(self):
session = MagicMock()
session._base_url = "https://api.meraki.com/api/v1"
reject_v0_base_url(session)
def test_trailing_slash_stripped(self):
session = MagicMock()
session._base_url = "https://api.meraki.com/api/v1/"
reject_v0_base_url(session)
assert session._base_url == "https://api.meraki.com/api/v1"
def test_reject_v0_strips_trailing_slash(self):
"""reject_v0_base_url strips trailing slash from valid URLs."""
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1/"
reject_v0_base_url(obj)
assert obj._base_url == "https://api.meraki.com/api/v1"
def test_reject_v0_leaves_valid_url_unchanged(self):
"""reject_v0_base_url leaves clean v1 URL untouched."""
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
reject_v0_base_url(obj)
assert obj._base_url == "https://api.meraki.com/api/v1"
class TestValidateBaseUrl:
def test_absolute_meraki_url_passthrough(self):
session = MagicMock()
session._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(session, "https://n123.meraki.com/api/v1/orgs")
assert result == "https://n123.meraki.com/api/v1/orgs"
def test_relative_path_prepends_base(self):
session = MagicMock()
session._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(session, "/organizations")
assert result == "https://api.meraki.com/api/v1/organizations"
def test_china_domain(self):
session = MagicMock()
session._base_url = "https://api.meraki.cn/api/v1"
result = validate_base_url(session, "https://n1.meraki.cn/api/v1/x")
assert result == "https://n1.meraki.cn/api/v1/x"
def test_canada_domain(self):
session = MagicMock()
session._base_url = "https://api.meraki.ca/api/v1"
result = validate_base_url(session, "https://n1.meraki.ca/api/v1/x")
assert result == "https://n1.meraki.ca/api/v1/x"
def test_india_domain(self):
session = MagicMock()
session._base_url = "https://api.meraki.in/api/v1"
result = validate_base_url(session, "https://n1.meraki.in/api/v1/x")
assert result == "https://n1.meraki.in/api/v1/x"
def test_gov_domain(self):
session = MagicMock()
session._base_url = "https://api.gov-meraki.com/api/v1"
result = validate_base_url(session, "https://n1.gov-meraki.com/api/v1/x")
assert result == "https://n1.gov-meraki.com/api/v1/x"
def test_meraki_com_absolute(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "https://n1.meraki.com/api/v1/networks")
assert result == "https://n1.meraki.com/api/v1/networks"
def test_relative_url_prepends_base(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "/organizations")
assert result == "https://api.meraki.com/api/v1/organizations"
def test_canada_domain_treated_as_absolute(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.ca/api/v1"
result = validate_base_url(obj, "https://api.meraki.ca/api/v1/networks")
assert result == "https://api.meraki.ca/api/v1/networks"
def test_china_domain_treated_as_absolute(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.cn/api/v1"
result = validate_base_url(obj, "https://api.meraki.cn/api/v1/networks")
assert result == "https://api.meraki.cn/api/v1/networks"
def test_india_domain_treated_as_absolute(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.in/api/v1"
result = validate_base_url(obj, "https://api.meraki.in/api/v1/networks")
assert result == "https://api.meraki.in/api/v1/networks"
def test_gov_meraki_domain_treated_as_absolute(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.gov-meraki.com/api/v1"
result = validate_base_url(obj, "https://api.gov-meraki.com/api/v1/networks")
assert result == "https://api.gov-meraki.com/api/v1/networks"
def test_unknown_domain_treated_as_relative(self):
from meraki.common import validate_base_url
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "https://evil.com/steal")
assert result == "https://api.meraki.com/api/v1https://evil.com/steal"
# --- Fix #14: host-boundary trusted-domain check (SSRF / key-exfil) ---
def test_legit_api_host_trusted(self):
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "https://api.meraki.com/api/v1/orgs")
assert result == "https://api.meraki.com/api/v1/orgs"
def test_legit_shard_subdomain_trusted(self):
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "https://n123.meraki.com/api/v1/orgs")
assert result == "https://n123.meraki.com/api/v1/orgs"
def test_legit_cn_host_trusted(self):
obj = MagicMock()
obj._base_url = "https://api.meraki.cn/api/v1"
result = validate_base_url(obj, "https://api.meraki.cn/api/v1/orgs")
assert result == "https://api.meraki.cn/api/v1/orgs"
def test_lookalike_suffix_rejected(self):
# "meraki.com" appears as a substring but not on a host boundary.
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
evil = "https://api.meraki.com.attacker.net/steal"
result = validate_base_url(obj, evil)
assert result == "https://api.meraki.com/api/v1" + evil
def test_lookalike_evil_example_rejected(self):
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
evil = "https://meraki.com.evil.example/steal"
result = validate_base_url(obj, evil)
assert result == "https://api.meraki.com/api/v1" + evil
def test_lookalike_prefix_rejected(self):
# "evil-meraki.com" embeds the domain but is a different host.
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
evil = "https://evil-meraki.com/steal"
result = validate_base_url(obj, evil)
assert result == "https://api.meraki.com/api/v1" + evil
def test_trusted_host_with_port(self):
obj = MagicMock()
obj._base_url = "https://api.meraki.com/api/v1"
result = validate_base_url(obj, "https://api.meraki.com:443/api/v1/orgs")
assert result == "https://api.meraki.com:443/api/v1/orgs"
class TestIteratorForGetPages:
def test_getter(self):
session = MagicMock()
session._use_iterator_for_get_pages = True
assert iterator_for_get_pages_bool(session) is True
def test_setter_true_assigns_iterator(self):
session = MagicMock()
use_iterator_for_get_pages_setter(session, True)
assert session.get_pages == session._get_pages_iterator
assert session._use_iterator_for_get_pages is True
def test_setter_false_assigns_legacy(self):
session = MagicMock()
use_iterator_for_get_pages_setter(session, False)
assert session.get_pages == session._get_pages_legacy
assert session._use_iterator_for_get_pages is False