1# Copyright (c) 2010 Mitch Garnaat http://garnaat.org/ 2# Copyright (c) 2010, Eucalyptus Systems, Inc. 3# All rights reserved. 4# 5# Permission is hereby granted, free of charge, to any person obtaining a 6# copy of this software and associated documentation files (the 7# "Software"), to deal in the Software without restriction, including 8# without limitation the rights to use, copy, modify, merge, publish, dis- 9# tribute, sublicense, and/or sell copies of the Software, and to permit 10# persons to whom the Software is furnished to do so, subject to the fol- 11# lowing conditions: 12# 13# The above copyright notice and this permission notice shall be included 14# in all copies or substantial portions of the Software. 15# 16# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS 17# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- 18# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT 19# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 20# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 21# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS 22# IN THE SOFTWARE. 23 24""" 25Some unit tests for the S3 Encryption 26""" 27import unittest 28import time 29from boto.s3.connection import S3Connection 30from boto.exception import S3ResponseError 31 32json_policy = """{ 33 "Version":"2008-10-17", 34 "Id":"PutObjPolicy", 35 "Statement":[{ 36 "Sid":"DenyUnEncryptedObjectUploads", 37 "Effect":"Deny", 38 "Principal":{ 39 "AWS":"*" 40 }, 41 "Action":"s3:PutObject", 42 "Resource":"arn:aws:s3:::%s/*", 43 "Condition":{ 44 "StringNotEquals":{ 45 "s3:x-amz-server-side-encryption":"AES256" 46 } 47 } 48 } 49 ] 50}""" 51 52class S3EncryptionTest (unittest.TestCase): 53 s3 = True 54 55 def test_1_versions(self): 56 print('--- running S3Encryption tests ---') 57 c = S3Connection() 58 # create a new, empty bucket 59 bucket_name = 'encryption-%d' % int(time.time()) 60 bucket = c.create_bucket(bucket_name) 61 62 # now try a get_bucket call and see if it's really there 63 bucket = c.get_bucket(bucket_name) 64 65 # create an unencrypted key 66 k = bucket.new_key('foobar') 67 s1 = 'This is unencrypted data' 68 s2 = 'This is encrypted data' 69 k.set_contents_from_string(s1) 70 time.sleep(5) 71 72 # now get the contents from s3 73 o = k.get_contents_as_string().decode('utf-8') 74 75 # check to make sure content read from s3 is identical to original 76 assert o == s1 77 78 # now overwrite that same key with encrypted data 79 k.set_contents_from_string(s2, encrypt_key=True) 80 time.sleep(5) 81 82 # now retrieve the contents as a string and compare 83 o = k.get_contents_as_string().decode('utf-8') 84 assert o == s2 85 86 # now set bucket policy to require encrypted objects 87 bucket.set_policy(json_policy % bucket.name) 88 time.sleep(5) 89 90 # now try to write unencrypted key 91 write_failed = False 92 try: 93 k.set_contents_from_string(s1) 94 except S3ResponseError: 95 write_failed = True 96 97 assert write_failed 98 99 # now try to write unencrypted key 100 write_failed = False 101 try: 102 k.set_contents_from_string(s1, encrypt_key=True) 103 except S3ResponseError: 104 write_failed = True 105 106 assert not write_failed 107 108 # Now do regular delete 109 k.delete() 110 time.sleep(5) 111 112 # now delete bucket 113 bucket.delete() 114 print('--- tests completed ---') 115