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