Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(831)

Issues Search
    My Issues | Starred     Open | Closed | All

Unified Diff: tests/cloudfront/test_signed_urls.py

Issue 8386013: Merging in latest boto. (Closed) Base URL: svn://svn.chromium.org/boto
Patch Set: Redoing vendor drop by deleting and then merging. Created 9 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « tests/cloudfront/__init__.py ('k') | tests/db/test_password.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: tests/cloudfront/test_signed_urls.py
diff --git a/tests/cloudfront/test_signed_urls.py b/tests/cloudfront/test_signed_urls.py
new file mode 100644
index 0000000000000000000000000000000000000000..118117ca47549c60bec6f117b717a1a47810f94d
--- /dev/null
+++ b/tests/cloudfront/test_signed_urls.py
@@ -0,0 +1,311 @@
+
+import unittest
+import json
+from textwrap import dedent
+from boto.cloudfront.distribution import Distribution
+
+class CloudfrontSignedUrlsTest(unittest.TestCase):
+ 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_unicode(self):
+ """
+ Test signing the canned policy from amazon's cloudfront documentation.
+ """
+ expected = ("Nql641NHEUkUaXQHZINK1FZ~SYeUSoBJMxjdgqrzIdzV2gyEXPDN"
+ "v0pYdWJkflDKJ3xIu7lbwRpSkG98NBlgPi4ZJpRRnVX4kXAJK6td"
+ "Nx6FucDB7OVqzcxkxHsGFd8VCG1BkC-Afh9~lOCMIYHIaiOB6~5j"
+ "t9w2EOwi6sIIqrg_")
+ unicode_policy = unicode(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(1, len(condition.keys()))
+ 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(1, len(condition.keys()))
+ ip_address = condition["IpAddress"]
+ 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(1, len(condition.keys()))
+ 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)
+
« no previous file with comments | « tests/cloudfront/__init__.py ('k') | tests/db/test_password.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698