← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-boto/trusty

« back to all changes in this revision

Viewing changes to tests/cloudfront/test_signed_urls.py

  • Committer: Package Import Robot
  • Author(s): Eric Evans
  • Date: 2012-04-15 20:21:21 UTC
  • mfrom: (1.1.9)
  • Revision ID: package-import@ubuntu.com-20120415202121-3fpf6q355s0xqpyu
Tags: 2.3.0-1
http://bugs.debian.org/664478
http://bugs.debian.org/650480
* New upstream release (Closes: #664478)
* Update debian/watch for Boto's move to Github.  Thanks Scott
  Moser. (Closes: #650480)

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/cloudfront/test_signed_urls.py
 
1
 
 
2
import unittest
 
3
try:
 
4
    import simplejson as json
 
5
except ImportError:
 
6
    import json
 
7
from textwrap import dedent
 
8
from boto.cloudfront.distribution import Distribution
 
9
 
 
10
class CloudfrontSignedUrlsTest(unittest.TestCase):
 
11
    def setUp(self):
 
12
        self.pk_str = dedent("""
 
13
            -----BEGIN RSA PRIVATE KEY-----
 
14
            MIICXQIBAAKBgQDA7ki9gI/lRygIoOjV1yymgx6FYFlzJ+z1ATMaLo57nL57AavW
 
15
            hb68HYY8EA0GJU9xQdMVaHBogF3eiCWYXSUZCWM/+M5+ZcdQraRRScucmn6g4EvY
 
16
            2K4W2pxbqH8vmUikPxir41EeBPLjMOzKvbzzQy9e/zzIQVREKSp/7y1mywIDAQAB
 
17
            AoGABc7mp7XYHynuPZxChjWNJZIq+A73gm0ASDv6At7F8Vi9r0xUlQe/v0AQS3yc
 
18
            N8QlyR4XMbzMLYk3yjxFDXo4ZKQtOGzLGteCU2srANiLv26/imXA8FVidZftTAtL
 
19
            viWQZBVPTeYIA69ATUYPEq0a5u5wjGyUOij9OWyuy01mbPkCQQDluYoNpPOekQ0Z
 
20
            WrPgJ5rxc8f6zG37ZVoDBiexqtVShIF5W3xYuWhW5kYb0hliYfkq15cS7t9m95h3
 
21
            1QJf/xI/AkEA1v9l/WN1a1N3rOK4VGoCokx7kR2SyTMSbZgF9IWJNOugR/WZw7HT
 
22
            njipO3c9dy1Ms9pUKwUF46d7049ck8HwdQJARgrSKuLWXMyBH+/l1Dx/I4tXuAJI
 
23
            rlPyo+VmiOc7b5NzHptkSHEPfR9s1OK0VqjknclqCJ3Ig86OMEtEFBzjZQJBAKYz
 
24
            470hcPkaGk7tKYAgP48FvxRsnzeooptURW5E+M+PQ2W9iDPPOX9739+Xi02hGEWF
 
25
            B0IGbQoTRFdE4VVcPK0CQQCeS84lODlC0Y2BZv2JxW3Osv/WkUQ4dslfAQl1T303
 
26
            7uwwr7XTroMv8dIFQIPreoPhRKmd/SbJzbiKfS/4QDhU
 
27
            -----END RSA PRIVATE KEY-----
 
28
            """)
 
29
        self.pk_id = "PK123456789754"
 
30
        self.dist = Distribution()
 
31
        self.canned_policy = (
 
32
            '{"Statement":[{"Resource":'
 
33
            '"http://d604721fxaaqy9.cloudfront.net/horizon.jpg'
 
34
            '?large=yes&license=yes",'
 
35
            '"Condition":{"DateLessThan":{"AWS:EpochTime":1258237200}}}]}')
 
36
        self.custom_policy_1 = (
 
37
            '{ \n'
 
38
            '   "Statement": [{ \n'
 
39
            '      "Resource":"http://d604721fxaaqy9.cloudfront.net/training/*", \n'
 
40
            '      "Condition":{ \n'
 
41
            '         "IpAddress":{"AWS:SourceIp":"145.168.143.0/24"}, \n'
 
42
            '         "DateLessThan":{"AWS:EpochTime":1258237200}      \n'
 
43
            '      } \n'
 
44
            '   }] \n'
 
45
            '}\n')
 
46
        self.custom_policy_2 = (
 
47
            '{ \n'
 
48
            '   "Statement": [{ \n'
 
49
            '      "Resource":"http://*", \n'
 
50
            '      "Condition":{ \n'
 
51
            '         "IpAddress":{"AWS:SourceIp":"216.98.35.1/32"},\n'
 
52
            '         "DateGreaterThan":{"AWS:EpochTime":1241073790},\n'
 
53
            '         "DateLessThan":{"AWS:EpochTime":1255674716}\n'
 
54
            '      } \n'
 
55
            '   }] \n'
 
56
            '}\n')
 
57
 
 
58
    def test_encode_custom_policy_1(self):
 
59
        """
 
60
        Test base64 encoding custom policy 1 from Amazon's documentation.
 
61
        """
 
62
        expected = ("eyAKICAgIlN0YXRlbWVudCI6IFt7IAogICAgICAiUmVzb3VyY2Ui"
 
63
                    "OiJodHRwOi8vZDYwNDcyMWZ4YWFxeTkuY2xvdWRmcm9udC5uZXQv"
 
64
                    "dHJhaW5pbmcvKiIsIAogICAgICAiQ29uZGl0aW9uIjp7IAogICAg"
 
65
                    "ICAgICAiSXBBZGRyZXNzIjp7IkFXUzpTb3VyY2VJcCI6IjE0NS4x"
 
66
                    "NjguMTQzLjAvMjQifSwgCiAgICAgICAgICJEYXRlTGVzc1RoYW4i"
 
67
                    "OnsiQVdTOkVwb2NoVGltZSI6MTI1ODIzNzIwMH0gICAgICAKICAg"
 
68
                    "ICAgfSAKICAgfV0gCn0K")
 
69
        encoded = self.dist._url_base64_encode(self.custom_policy_1)
 
70
        self.assertEqual(expected, encoded)
 
71
 
 
72
    def test_encode_custom_policy_2(self):
 
73
        """
 
74
        Test base64 encoding custom policy 2 from Amazon's documentation.
 
75
        """
 
76
        expected = ("eyAKICAgIlN0YXRlbWVudCI6IFt7IAogICAgICAiUmVzb3VyY2Ui"
 
77
                    "OiJodHRwOi8vKiIsIAogICAgICAiQ29uZGl0aW9uIjp7IAogICAg"
 
78
                    "ICAgICAiSXBBZGRyZXNzIjp7IkFXUzpTb3VyY2VJcCI6IjIxNi45"
 
79
                    "OC4zNS4xLzMyIn0sCiAgICAgICAgICJEYXRlR3JlYXRlclRoYW4i"
 
80
                    "OnsiQVdTOkVwb2NoVGltZSI6MTI0MTA3Mzc5MH0sCiAgICAgICAg"
 
81
                    "ICJEYXRlTGVzc1RoYW4iOnsiQVdTOkVwb2NoVGltZSI6MTI1NTY3"
 
82
                    "NDcxNn0KICAgICAgfSAKICAgfV0gCn0K")
 
83
        encoded = self.dist._url_base64_encode(self.custom_policy_2)
 
84
        self.assertEqual(expected, encoded)
 
85
 
 
86
    def test_sign_canned_policy(self):
 
87
        """
 
88
        Test signing the canned policy from amazon's cloudfront documentation.
 
89
        """
 
90
        expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
 
91
                    "v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
 
92
                    "Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
 
93
                    "t9w2EOwi6sIIqrg_")
 
94
        sig = self.dist._sign_string(self.canned_policy, private_key_string=self.pk_str)
 
95
        encoded_sig = self.dist._url_base64_encode(sig)
 
96
        self.assertEqual(expected, encoded_sig)
 
97
 
 
98
    def test_sign_canned_policy_unicode(self):
 
99
        """
 
100
        Test signing the canned policy from amazon's cloudfront documentation.
 
101
        """
 
102
        expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
 
103
                    "v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
 
104
                    "Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
 
105
                    "t9w2EOwi6sIIqrg_")
 
106
        unicode_policy = unicode(self.canned_policy)
 
107
        sig = self.dist._sign_string(unicode_policy, private_key_string=self.pk_str)
 
108
        encoded_sig = self.dist._url_base64_encode(sig)
 
109
        self.assertEqual(expected, encoded_sig)
 
110
 
 
111
    def test_sign_custom_policy_1(self):
 
112
        """
 
113
        Test signing custom policy 1 from amazon's cloudfront documentation.
 
114
        """
 
115
        expected = ("cPFtRKvUfYNYmxek6ZNs6vgKEZP6G3Cb4cyVt~FjqbHOnMdxdT7e"
 
116
                    "T6pYmhHYzuDsFH4Jpsctke2Ux6PCXcKxUcTIm8SO4b29~1QvhMl-"
 
117
                    "CIojki3Hd3~Unxjw7Cpo1qRjtvrimW0DPZBZYHFZtiZXsaPt87yB"
 
118
                    "P9GWnTQoaVysMxQ_")
 
119
        sig = self.dist._sign_string(self.custom_policy_1, private_key_string=self.pk_str)
 
120
        encoded_sig = self.dist._url_base64_encode(sig)
 
121
        self.assertEqual(expected, encoded_sig)
 
122
 
 
123
    def test_sign_custom_policy_2(self):
 
124
        """
 
125
        Test signing custom policy 2 from amazon's cloudfront documentation.
 
126
        """
 
127
        expected = ("rc~5Qbbm8EJXjUTQ6Cn0LAxR72g1DOPrTmdtfbWVVgQNw0q~KHUA"
 
128
                    "mBa2Zv1Wjj8dDET4XSL~Myh44CLQdu4dOH~N9huH7QfPSR~O4tIO"
 
129
                    "S1WWcP~2JmtVPoQyLlEc8YHRCuN3nVNZJ0m4EZcXXNAS-0x6Zco2"
 
130
                    "SYx~hywTRxWR~5Q_")
 
131
        sig = self.dist._sign_string(self.custom_policy_2, private_key_string=self.pk_str)
 
132
        encoded_sig = self.dist._url_base64_encode(sig)
 
133
        self.assertEqual(expected, encoded_sig)
 
134
 
 
135
    def test_create_canned_policy(self):
 
136
        """
 
137
        Test that a canned policy is generated correctly.
 
138
        """
 
139
        url = "http://1234567.cloudfront.com/test_resource.mp3?dog=true"
 
140
        expires = 999999
 
141
        policy = self.dist._canned_policy(url, expires)
 
142
        policy = json.loads(policy)
 
143
 
 
144
        self.assertEqual(1, len(policy.keys()))
 
145
        statements = policy["Statement"]
 
146
        self.assertEqual(1, len(statements))
 
147
        statement = statements[0]
 
148
        resource = statement["Resource"]
 
149
        self.assertEqual(url, resource)
 
150
        condition = statement["Condition"]
 
151
        self.assertEqual(1, len(condition.keys()))
 
152
        date_less_than = condition["DateLessThan"]
 
153
        self.assertEqual(1, len(date_less_than.keys()))
 
154
        aws_epoch_time = date_less_than["AWS:EpochTime"]
 
155
        self.assertEqual(expires, aws_epoch_time)
 
156
        
 
157
    def test_custom_policy_expires_and_policy_url(self):
 
158
        """
 
159
        Test that a custom policy can be created with an expire time and an
 
160
        arbitrary URL.
 
161
        """
 
162
        url = "http://1234567.cloudfront.com/*"
 
163
        expires = 999999
 
164
        policy = self.dist._custom_policy(url, expires=expires)
 
165
        policy = json.loads(policy)
 
166
 
 
167
        self.assertEqual(1, len(policy.keys()))
 
168
        statements = policy["Statement"]
 
169
        self.assertEqual(1, len(statements))
 
170
        statement = statements[0]
 
171
        resource = statement["Resource"]
 
172
        self.assertEqual(url, resource)
 
173
        condition = statement["Condition"]
 
174
        self.assertEqual(1, len(condition.keys()))
 
175
        date_less_than = condition["DateLessThan"]
 
176
        self.assertEqual(1, len(date_less_than.keys()))
 
177
        aws_epoch_time = date_less_than["AWS:EpochTime"]
 
178
        self.assertEqual(expires, aws_epoch_time)
 
179
 
 
180
    def test_custom_policy_valid_after(self):
 
181
        """
 
182
        Test that a custom policy can be created with a valid-after time and
 
183
        an arbitrary URL.
 
184
        """
 
185
        url = "http://1234567.cloudfront.com/*"
 
186
        valid_after = 999999
 
187
        policy = self.dist._custom_policy(url, valid_after=valid_after)
 
188
        policy = json.loads(policy)
 
189
 
 
190
        self.assertEqual(1, len(policy.keys()))
 
191
        statements = policy["Statement"]
 
192
        self.assertEqual(1, len(statements))
 
193
        statement = statements[0]
 
194
        resource = statement["Resource"]
 
195
        self.assertEqual(url, resource)
 
196
        condition = statement["Condition"]
 
197
        self.assertEqual(1, len(condition.keys()))
 
198
        date_greater_than = condition["DateGreaterThan"]
 
199
        self.assertEqual(1, len(date_greater_than.keys()))
 
200
        aws_epoch_time = date_greater_than["AWS:EpochTime"]
 
201
        self.assertEqual(valid_after, aws_epoch_time)
 
202
 
 
203
    def test_custom_policy_ip_address(self):
 
204
        """
 
205
        Test that a custom policy can be created with an IP address and
 
206
        an arbitrary URL.
 
207
        """
 
208
        url = "http://1234567.cloudfront.com/*"
 
209
        ip_range = "192.168.0.1"
 
210
        policy = self.dist._custom_policy(url, ip_address=ip_range)
 
211
        policy = json.loads(policy)
 
212
 
 
213
        self.assertEqual(1, len(policy.keys()))
 
214
        statements = policy["Statement"]
 
215
        self.assertEqual(1, len(statements))
 
216
        statement = statements[0]
 
217
        resource = statement["Resource"]
 
218
        self.assertEqual(url, resource)
 
219
        condition = statement["Condition"]
 
220
        self.assertEqual(1, len(condition.keys()))
 
221
        ip_address = condition["IpAddress"]
 
222
        self.assertEqual(1, len(ip_address.keys()))
 
223
        source_ip = ip_address["AWS:SourceIp"]
 
224
        self.assertEqual("%s/32" % ip_range, source_ip)
 
225
 
 
226
    def test_custom_policy_ip_range(self):
 
227
        """
 
228
        Test that a custom policy can be created with an IP address and
 
229
        an arbitrary URL.
 
230
        """
 
231
        url = "http://1234567.cloudfront.com/*"
 
232
        ip_range = "192.168.0.0/24"
 
233
        policy = self.dist._custom_policy(url, ip_address=ip_range)
 
234
        policy = json.loads(policy)
 
235
 
 
236
        self.assertEqual(1, len(policy.keys()))
 
237
        statements = policy["Statement"]
 
238
        self.assertEqual(1, len(statements))
 
239
        statement = statements[0]
 
240
        resource = statement["Resource"]
 
241
        self.assertEqual(url, resource)
 
242
        condition = statement["Condition"]
 
243
        self.assertEqual(1, len(condition.keys()))
 
244
        ip_address = condition["IpAddress"]
 
245
        self.assertEqual(1, len(ip_address.keys()))
 
246
        source_ip = ip_address["AWS:SourceIp"]
 
247
        self.assertEqual(ip_range, source_ip)
 
248
 
 
249
    def test_custom_policy_all(self):
 
250
        """
 
251
        Test that a custom policy can be created with an IP address and
 
252
        an arbitrary URL.
 
253
        """
 
254
        url = "http://1234567.cloudfront.com/test.txt"
 
255
        expires = 999999
 
256
        valid_after = 111111
 
257
        ip_range = "192.168.0.0/24"
 
258
        policy = self.dist._custom_policy(url, expires=expires,
 
259
                                          valid_after=valid_after,
 
260
                                          ip_address=ip_range)
 
261
        policy = json.loads(policy)
 
262
 
 
263
        self.assertEqual(1, len(policy.keys()))
 
264
        statements = policy["Statement"]
 
265
        self.assertEqual(1, len(statements))
 
266
        statement = statements[0]
 
267
        resource = statement["Resource"]
 
268
        self.assertEqual(url, resource)
 
269
        condition = statement["Condition"]
 
270
        self.assertEqual(3, len(condition.keys()))
 
271
        #check expires condition
 
272
        date_less_than = condition["DateLessThan"]
 
273
        self.assertEqual(1, len(date_less_than.keys()))
 
274
        aws_epoch_time = date_less_than["AWS:EpochTime"]
 
275
        self.assertEqual(expires, aws_epoch_time)
 
276
        #check valid_after condition
 
277
        date_greater_than = condition["DateGreaterThan"]
 
278
        self.assertEqual(1, len(date_greater_than.keys()))
 
279
        aws_epoch_time = date_greater_than["AWS:EpochTime"]
 
280
        self.assertEqual(valid_after, aws_epoch_time)
 
281
        #check source ip address condition
 
282
        ip_address = condition["IpAddress"]
 
283
        self.assertEqual(1, len(ip_address.keys()))
 
284
        source_ip = ip_address["AWS:SourceIp"]
 
285
        self.assertEqual(ip_range, source_ip)
 
286
 
 
287
    def test_params_canned_policy(self):
 
288
        """
 
289
        Test the correct params are generated for a canned policy.
 
290
        """
 
291
        url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes"
 
292
        expire_time = 1258237200
 
293
        expected_sig = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyE"
 
294
                        "XPDNv0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4"
 
295
                        "kXAJK6tdNx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCM"
 
296
                        "IYHIaiOB6~5jt9w2EOwi6sIIqrg_")
 
297
        signed_url_params = self.dist._create_signing_params(url, self.pk_id, expire_time, private_key_string=self.pk_str)
 
298
        self.assertEqual(3, len(signed_url_params))
 
299
        self.assertEqual(signed_url_params["Expires"], "1258237200")
 
300
        self.assertEqual(signed_url_params["Signature"], expected_sig)
 
301
        self.assertEqual(signed_url_params["Key-Pair-Id"], "PK123456789754")
 
302
 
 
303
    def test_canned_policy(self):
 
304
        """
 
305
        Generate signed url from the Example Canned Policy in Amazon's
 
306
        documentation.
 
307
        """
 
308
        url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes"
 
309
        expire_time = 1258237200
 
310
        expected_url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes&Expires=1258237200&Signature=Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDNv0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6tdNx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5jt9w2EOwi6sIIqrg_&Key-Pair-Id=PK123456789754"
 
311
        signed_url = self.dist.create_signed_url(
 
312
            url, self.pk_id, expire_time, private_key_string=self.pk_str)
 
313
        self.assertEqual(expected_url, signed_url)
 
314