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