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
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
|
import tempfile
import unittest
from boto.compat import StringIO, six, json
from textwrap import dedent
from boto.cloudfront.distribution import Distribution
class CloudfrontSignedUrlsTest(unittest.TestCase):
cloudfront = True
notdefault = True
def setUp(self):
self.pk_str = dedent("""
-----BEGIN RSA PRIVATE KEY-----
MIICXQIBAAKBgQDA7ki9gI/lRygIoOjV1yymgx6FYFlzJ+z1ATMaLo57nL57AavW
hb68HYY8EA0GJU9xQdMVaHBogF3eiCWYXSUZCWM/+M5+ZcdQraRRScucmn6g4EvY
2K4W2pxbqH8vmUikPxir41EeBPLjMOzKvbzzQy9e/zzIQVREKSp/7y1mywIDAQAB
AoGABc7mp7XYHynuPZxChjWNJZIq+A73gm0ASDv6At7F8Vi9r0xUlQe/v0AQS3yc
N8QlyR4XMbzMLYk3yjxFDXo4ZKQtOGzLGteCU2srANiLv26/imXA8FVidZftTAtL
viWQZBVPTeYIA69ATUYPEq0a5u5wjGyUOij9OWyuy01mbPkCQQDluYoNpPOekQ0Z
WrPgJ5rxc8f6zG37ZVoDBiexqtVShIF5W3xYuWhW5kYb0hliYfkq15cS7t9m95h3
1QJf/xI/AkEA1v9l/WN1a1N3rOK4VGoCokx7kR2SyTMSbZgF9IWJNOugR/WZw7HT
njipO3c9dy1Ms9pUKwUF46d7049ck8HwdQJARgrSKuLWXMyBH+/l1Dx/I4tXuAJI
rlPyo+VmiOc7b5NzHptkSHEPfR9s1OK0VqjknclqCJ3Ig86OMEtEFBzjZQJBAKYz
470hcPkaGk7tKYAgP48FvxRsnzeooptURW5E+M+PQ2W9iDPPOX9739+Xi02hGEWF
B0IGbQoTRFdE4VVcPK0CQQCeS84lODlC0Y2BZv2JxW3Osv/WkUQ4dslfAQl1T303
7uwwr7XTroMv8dIFQIPreoPhRKmd/SbJzbiKfS/4QDhU
-----END RSA PRIVATE KEY-----
""")
self.pk_id = "PK123456789754"
self.dist = Distribution()
self.canned_policy = (
'{"Statement":[{"Resource":'
'"http://d604721fxaaqy9.cloudfront.net/horizon.jpg'
'?large=yes&license=yes",'
'"Condition":{"DateLessThan":{"AWS:EpochTime":1258237200}}}]}')
self.custom_policy_1 = (
'{ \n'
' "Statement": [{ \n'
' "Resource":"http://d604721fxaaqy9.cloudfront.net/training/*", \n'
' "Condition":{ \n'
' "IpAddress":{"AWS:SourceIp":"145.168.143.0/24"}, \n'
' "DateLessThan":{"AWS:EpochTime":1258237200} \n'
' } \n'
' }] \n'
'}\n')
self.custom_policy_2 = (
'{ \n'
' "Statement": [{ \n'
' "Resource":"http://*", \n'
' "Condition":{ \n'
' "IpAddress":{"AWS:SourceIp":"216.98.35.1/32"},\n'
' "DateGreaterThan":{"AWS:EpochTime":1241073790},\n'
' "DateLessThan":{"AWS:EpochTime":1255674716}\n'
' } \n'
' }] \n'
'}\n')
def test_encode_custom_policy_1(self):
"""
Test base64 encoding custom policy 1 from Amazon's documentation.
"""
expected = ("eyAKICAgIlN0YXRlbWVudCI6IFt7IAogICAgICAiUmVzb3VyY2Ui"
"OiJodHRwOi8vZDYwNDcyMWZ4YWFxeTkuY2xvdWRmcm9udC5uZXQv"
"dHJhaW5pbmcvKiIsIAogICAgICAiQ29uZGl0aW9uIjp7IAogICAg"
"ICAgICAiSXBBZGRyZXNzIjp7IkFXUzpTb3VyY2VJcCI6IjE0NS4x"
"NjguMTQzLjAvMjQifSwgCiAgICAgICAgICJEYXRlTGVzc1RoYW4i"
"OnsiQVdTOkVwb2NoVGltZSI6MTI1ODIzNzIwMH0gICAgICAKICAg"
"ICAgfSAKICAgfV0gCn0K")
encoded = self.dist._url_base64_encode(self.custom_policy_1)
self.assertEqual(expected, encoded)
def test_encode_custom_policy_2(self):
"""
Test base64 encoding custom policy 2 from Amazon's documentation.
"""
expected = ("eyAKICAgIlN0YXRlbWVudCI6IFt7IAogICAgICAiUmVzb3VyY2Ui"
"OiJodHRwOi8vKiIsIAogICAgICAiQ29uZGl0aW9uIjp7IAogICAg"
"ICAgICAiSXBBZGRyZXNzIjp7IkFXUzpTb3VyY2VJcCI6IjIxNi45"
"OC4zNS4xLzMyIn0sCiAgICAgICAgICJEYXRlR3JlYXRlclRoYW4i"
"OnsiQVdTOkVwb2NoVGltZSI6MTI0MTA3Mzc5MH0sCiAgICAgICAg"
"ICJEYXRlTGVzc1RoYW4iOnsiQVdTOkVwb2NoVGltZSI6MTI1NTY3"
"NDcxNn0KICAgICAgfSAKICAgfV0gCn0K")
encoded = self.dist._url_base64_encode(self.custom_policy_2)
self.assertEqual(expected, encoded)
def test_sign_canned_policy(self):
"""
Test signing the canned policy from amazon's cloudfront documentation.
"""
expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
"v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
"Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
"t9w2EOwi6sIIqrg_")
sig = self.dist._sign_string(self.canned_policy, private_key_string=self.pk_str)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_canned_policy_pk_file(self):
"""
Test signing the canned policy from amazon's cloudfront documentation
with a file object.
"""
expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
"v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
"Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
"t9w2EOwi6sIIqrg_")
pk_file = tempfile.TemporaryFile()
pk_file.write(self.pk_str)
pk_file.seek(0)
sig = self.dist._sign_string(self.canned_policy, private_key_file=pk_file)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_canned_policy_pk_file_name(self):
"""
Test signing the canned policy from amazon's cloudfront documentation
with a file name.
"""
expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
"v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
"Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
"t9w2EOwi6sIIqrg_")
pk_file = tempfile.NamedTemporaryFile()
pk_file.write(self.pk_str)
pk_file.flush()
sig = self.dist._sign_string(self.canned_policy, private_key_file=pk_file.name)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_canned_policy_pk_file_like(self):
"""
Test signing the canned policy from amazon's cloudfront documentation
with a file-like object (not a subclass of 'file' type)
"""
expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
"v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
"Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
"t9w2EOwi6sIIqrg_")
pk_file = StringIO()
pk_file.write(self.pk_str)
pk_file.seek(0)
sig = self.dist._sign_string(self.canned_policy, private_key_file=pk_file)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_canned_policy_unicode(self):
"""
Test signing the canned policy from amazon's cloudfront documentation.
"""
expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
"v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
"Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
"t9w2EOwi6sIIqrg_")
unicode_policy = six.text_type(self.canned_policy)
sig = self.dist._sign_string(unicode_policy, private_key_string=self.pk_str)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_custom_policy_1(self):
"""
Test signing custom policy 1 from amazon's cloudfront documentation.
"""
expected = ("cPFtRKvUfYNYmxek6ZNs6vgKEZP6G3Cb4cyVt~FjqbHOnMdxdT7e"
"T6pYmhHYzuDsFH4Jpsctke2Ux6PCXcKxUcTIm8SO4b29~1QvhMl-"
"CIojki3Hd3~Unxjw7Cpo1qRjtvrimW0DPZBZYHFZtiZXsaPt87yB"
"P9GWnTQoaVysMxQ_")
sig = self.dist._sign_string(self.custom_policy_1, private_key_string=self.pk_str)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_sign_custom_policy_2(self):
"""
Test signing custom policy 2 from amazon's cloudfront documentation.
"""
expected = ("rc~5Qbbm8EJXjUTQ6Cn0LAxR72g1DOPrTmdtfbWVVgQNw0q~KHUA"
"mBa2Zv1Wjj8dDET4XSL~Myh44CLQdu4dOH~N9huH7QfPSR~O4tIO"
"S1WWcP~2JmtVPoQyLlEc8YHRCuN3nVNZJ0m4EZcXXNAS-0x6Zco2"
"SYx~hywTRxWR~5Q_")
sig = self.dist._sign_string(self.custom_policy_2, private_key_string=self.pk_str)
encoded_sig = self.dist._url_base64_encode(sig)
self.assertEqual(expected, encoded_sig)
def test_create_canned_policy(self):
"""
Test that a canned policy is generated correctly.
"""
url = "http://1234567.cloudfront.com/test_resource.mp3?dog=true"
expires = 999999
policy = self.dist._canned_policy(url, expires)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(1, len(condition.keys()))
date_less_than = condition["DateLessThan"]
self.assertEqual(1, len(date_less_than.keys()))
aws_epoch_time = date_less_than["AWS:EpochTime"]
self.assertEqual(expires, aws_epoch_time)
def test_custom_policy_expires_and_policy_url(self):
"""
Test that a custom policy can be created with an expire time and an
arbitrary URL.
"""
url = "http://1234567.cloudfront.com/*"
expires = 999999
policy = self.dist._custom_policy(url, expires=expires)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(1, len(condition.keys()))
date_less_than = condition["DateLessThan"]
self.assertEqual(1, len(date_less_than.keys()))
aws_epoch_time = date_less_than["AWS:EpochTime"]
self.assertEqual(expires, aws_epoch_time)
def test_custom_policy_valid_after(self):
"""
Test that a custom policy can be created with a valid-after time and
an arbitrary URL.
"""
url = "http://1234567.cloudfront.com/*"
valid_after = 999999
policy = self.dist._custom_policy(url, valid_after=valid_after)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(2, len(condition.keys()))
date_less_than = condition["DateLessThan"]
date_greater_than = condition["DateGreaterThan"]
self.assertEqual(1, len(date_greater_than.keys()))
aws_epoch_time = date_greater_than["AWS:EpochTime"]
self.assertEqual(valid_after, aws_epoch_time)
def test_custom_policy_ip_address(self):
"""
Test that a custom policy can be created with an IP address and
an arbitrary URL.
"""
url = "http://1234567.cloudfront.com/*"
ip_range = "192.168.0.1"
policy = self.dist._custom_policy(url, ip_address=ip_range)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(2, len(condition.keys()))
ip_address = condition["IpAddress"]
self.assertTrue("DateLessThan" in condition)
self.assertEqual(1, len(ip_address.keys()))
source_ip = ip_address["AWS:SourceIp"]
self.assertEqual("%s/32" % ip_range, source_ip)
def test_custom_policy_ip_range(self):
"""
Test that a custom policy can be created with an IP address and
an arbitrary URL.
"""
url = "http://1234567.cloudfront.com/*"
ip_range = "192.168.0.0/24"
policy = self.dist._custom_policy(url, ip_address=ip_range)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(2, len(condition.keys()))
self.assertTrue("DateLessThan" in condition)
ip_address = condition["IpAddress"]
self.assertEqual(1, len(ip_address.keys()))
source_ip = ip_address["AWS:SourceIp"]
self.assertEqual(ip_range, source_ip)
def test_custom_policy_all(self):
"""
Test that a custom policy can be created with an IP address and
an arbitrary URL.
"""
url = "http://1234567.cloudfront.com/test.txt"
expires = 999999
valid_after = 111111
ip_range = "192.168.0.0/24"
policy = self.dist._custom_policy(url, expires=expires,
valid_after=valid_after,
ip_address=ip_range)
policy = json.loads(policy)
self.assertEqual(1, len(policy.keys()))
statements = policy["Statement"]
self.assertEqual(1, len(statements))
statement = statements[0]
resource = statement["Resource"]
self.assertEqual(url, resource)
condition = statement["Condition"]
self.assertEqual(3, len(condition.keys()))
#check expires condition
date_less_than = condition["DateLessThan"]
self.assertEqual(1, len(date_less_than.keys()))
aws_epoch_time = date_less_than["AWS:EpochTime"]
self.assertEqual(expires, aws_epoch_time)
#check valid_after condition
date_greater_than = condition["DateGreaterThan"]
self.assertEqual(1, len(date_greater_than.keys()))
aws_epoch_time = date_greater_than["AWS:EpochTime"]
self.assertEqual(valid_after, aws_epoch_time)
#check source ip address condition
ip_address = condition["IpAddress"]
self.assertEqual(1, len(ip_address.keys()))
source_ip = ip_address["AWS:SourceIp"]
self.assertEqual(ip_range, source_ip)
def test_params_canned_policy(self):
"""
Test the correct params are generated for a canned policy.
"""
url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes"
expire_time = 1258237200
expected_sig = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyE"
"XPDNv0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4"
"kXAJK6tdNx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCM"
"IYHIaiOB6~5jt9w2EOwi6sIIqrg_")
signed_url_params = self.dist._create_signing_params(url, self.pk_id, expire_time, private_key_string=self.pk_str)
self.assertEqual(3, len(signed_url_params))
self.assertEqual(signed_url_params["Expires"], "1258237200")
self.assertEqual(signed_url_params["Signature"], expected_sig)
self.assertEqual(signed_url_params["Key-Pair-Id"], "PK123456789754")
def test_canned_policy(self):
"""
Generate signed url from the Example Canned Policy in Amazon's
documentation.
"""
url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes"
expire_time = 1258237200
expected_url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes&Expires=1258237200&Signature=Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDNv0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6tdNx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5jt9w2EOwi6sIIqrg_&Key-Pair-Id=PK123456789754"
signed_url = self.dist.create_signed_url(
url, self.pk_id, expire_time, private_key_string=self.pk_str)
self.assertEqual(expected_url, signed_url)
|
