~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/storage/tests/test_client.py

Merged 424018-add-keypair-support [r=jkakar] [f=424018]

This change implements three keypair methods in the EC2 client:
 1. describe_keypairs
 2. create_keypair
 3. delete_keypair

Show diffs side-by-side

added added

removed removed

Lines of Context:
16
16
 
17
17
    def get_page(self, url, method, postdata, headers):
18
18
        self.getPageArgs = (url, method, postdata, headers)
19
 
        return succeed('')
 
19
        return succeed("")
20
20
 
21
21
 
22
22
class RequestTestCase(TXAWSTestCase):
23
23
 
24
 
    creds = AWSCredentials(access_key='fookeyid', secret_key='barsecretkey')
 
24
    creds = AWSCredentials(access_key="fookeyid", secret_key="barsecretkey")
25
25
    endpoint = AWSServiceEndpoint("https://s3.amazonaws.com/")
26
26
 
27
27
    def test_get_uri_with_endpoint(self):
28
28
        endpoint = AWSServiceEndpoint("http://localhost/")
29
 
        request = S3Request('PUT', endpoint=endpoint)
 
29
        request = S3Request("PUT", endpoint=endpoint)
30
30
        self.assertEquals(request.endpoint.get_uri(), "http://localhost/")
31
31
        self.assertEquals(request.get_uri(), "http://localhost/")
32
32
 
33
33
    def test_get_uri_with_endpoint_bucket_and_object(self):
34
34
        endpoint = AWSServiceEndpoint("http://localhost/")
35
 
        request = S3Request('PUT', bucket="mybucket", object_name="myobject",
 
35
        request = S3Request("PUT", bucket="mybucket", object_name="myobject",
36
36
                            endpoint=endpoint)
37
37
        self.assertEquals(
38
38
            request.get_uri(),
39
39
            "http://localhost/mybucket/myobject")
40
40
 
41
41
    def test_get_uri_with_no_endpoint(self):
42
 
        request = S3Request('PUT')
 
42
        request = S3Request("PUT")
43
43
        self.assertEquals(request.endpoint, None)
44
44
        self.assertEquals(request.get_uri(), "http:///")
45
45
 
46
46
    def test_get_path_with_bucket_and_object(self):
47
 
        request = S3Request('PUT', bucket="mybucket", object_name="myobject")
 
47
        request = S3Request("PUT", bucket="mybucket", object_name="myobject")
48
48
        self.assertEquals(request.get_path(), "/mybucket/myobject")
49
49
 
50
50
    def test_get_path_with_no_bucket_or_object(self):
51
 
        request = S3Request('PUT')
 
51
        request = S3Request("PUT")
52
52
        self.assertEquals(request.get_path(), "/")
53
53
 
54
54
    def test_objectRequest(self):
55
55
        """
56
56
        Test that a request addressing an object is created correctly.
57
57
        """
58
 
        DATA = 'objectData'
59
 
        DIGEST = 'zhdB6gwvocWv/ourYUWMxA=='
 
58
        DATA = "objectData"
 
59
        DIGEST = "zhdB6gwvocWv/ourYUWMxA=="
60
60
 
61
 
        request = S3Request('PUT', 'somebucket', 'object/name/here', DATA,
62
 
                            content_type='text/plain', metadata={'foo': 'bar'},
 
61
        request = S3Request("PUT", "somebucket", "object/name/here", DATA,
 
62
                            content_type="text/plain", metadata={"foo": "bar"},
63
63
                            creds=self.creds, endpoint=self.endpoint)
64
64
        request.get_signature = lambda headers: "TESTINGSIG="
65
 
        self.assertEqual(request.verb, 'PUT')
 
65
        self.assertEqual(request.verb, "PUT")
66
66
        self.assertEqual(
67
67
            request.get_uri(),
68
 
            'https://s3.amazonaws.com/somebucket/object/name/here')
 
68
            "https://s3.amazonaws.com/somebucket/object/name/here")
69
69
        headers = request.get_headers()
70
 
        self.assertNotEqual(headers.pop('Date'), '')
 
70
        self.assertNotEqual(headers.pop("Date"), "")
71
71
        self.assertEqual(
72
72
            headers, {
73
 
                'Authorization': 'AWS fookeyid:TESTINGSIG=',
74
 
                'Content-Type': 'text/plain',
75
 
                'Content-Length': len(DATA),
76
 
                'Content-MD5': DIGEST,
77
 
                'x-amz-meta-foo': 'bar'})
78
 
        self.assertEqual(request.data, 'objectData')
 
73
                "Authorization": "AWS fookeyid:TESTINGSIG=",
 
74
                "Content-Type": "text/plain",
 
75
                "Content-Length": len(DATA),
 
76
                "Content-MD5": DIGEST,
 
77
                "x-amz-meta-foo": "bar"})
 
78
        self.assertEqual(request.data, "objectData")
79
79
 
80
80
    def test_bucketRequest(self):
81
81
        """
82
82
        Test that a request addressing a bucket is created correctly.
83
83
        """
84
 
        DIGEST = '1B2M2Y8AsgTpgAmY7PhCfg=='
 
84
        DIGEST = "1B2M2Y8AsgTpgAmY7PhCfg=="
85
85
 
86
 
        request = S3Request('GET', 'somebucket', creds=self.creds,
 
86
        request = S3Request("GET", "somebucket", creds=self.creds,
87
87
                            endpoint=self.endpoint)
88
88
        request.get_signature = lambda headers: "TESTINGSIG="
89
 
        self.assertEqual(request.verb, 'GET')
 
89
        self.assertEqual(request.verb, "GET")
90
90
        self.assertEqual(
91
 
            request.get_uri(), 'https://s3.amazonaws.com/somebucket')
 
91
            request.get_uri(), "https://s3.amazonaws.com/somebucket")
92
92
        headers = request.get_headers()
93
 
        self.assertNotEqual(headers.pop('Date'), '')
 
93
        self.assertNotEqual(headers.pop("Date"), "")
94
94
        self.assertEqual(
95
95
            headers, {
96
 
            'Authorization': 'AWS fookeyid:TESTINGSIG=',
97
 
            'Content-Length': 0,
98
 
            'Content-MD5': DIGEST})
99
 
        self.assertEqual(request.data, '')
 
96
            "Authorization": "AWS fookeyid:TESTINGSIG=",
 
97
            "Content-Length": 0,
 
98
            "Content-MD5": DIGEST})
 
99
        self.assertEqual(request.data, "")
100
100
 
101
101
    def test_submit(self):
102
102
        """
103
103
        Submitting the request should invoke getPage correctly.
104
104
        """
105
 
        request = StubbedS3Request('GET', 'somebucket', creds=self.creds,
 
105
        request = StubbedS3Request("GET", "somebucket", creds=self.creds,
106
106
                                   endpoint=self.endpoint)
107
107
 
108
108
        def _postCheck(result):
109
 
            self.assertEqual(result, '')
 
109
            self.assertEqual(result, "")
110
110
 
111
111
            url, method, postdata, headers = request.getPageArgs
112
112
            self.assertEqual(url, request.get_uri())
117
117
        return request.submit().addCallback(_postCheck)
118
118
 
119
119
    def test_authenticationTestCases(self):
120
 
        request = S3Request('GET', creds=self.creds, endpoint=self.endpoint)
 
120
        request = S3Request("GET", creds=self.creds, endpoint=self.endpoint)
121
121
        request.get_signature = lambda headers: "TESTINGSIG="
122
 
        request.date = 'Wed, 28 Mar 2007 01:29:59 +0000'
 
122
        request.date = "Wed, 28 Mar 2007 01:29:59 +0000"
123
123
 
124
124
        headers = request.get_headers()
125
125
        self.assertEqual(
126
 
            headers['Authorization'], 
127
 
            'AWS fookeyid:TESTINGSIG=')
 
126
            headers["Authorization"], 
 
127
            "AWS fookeyid:TESTINGSIG=")
128
128
 
129
129
 
130
130
class InertRequest(S3Request):
136
136
    submitted = False
137
137
 
138
138
    def __init__(self, *a, **kw):
139
 
        self.response = kw.pop('response')
 
139
        self.response = kw.pop("response")
140
140
        super(InertRequest, self).__init__(*a, **kw)
141
141
 
142
142
    def submit(self):
164
164
 
165
165
 
166
166
samples = {
167
 
    'ListAllMyBucketsResult':
 
167
    "ListAllMyBucketsResult":
168
168
    """<?xml version="1.0" encoding="UTF-8"?>
169
169
<ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
170
170
  <Owner>
190
190
    def setUp(self):
191
191
        TXAWSTestCase.setUp(self)
192
192
        self.creds = AWSCredentials(
193
 
            access_key='accessKey', secret_key='secretKey')
 
193
            access_key="accessKey", secret_key="secretKey")
194
194
        self.endpoint = AWSServiceEndpoint()
195
195
        self.s3 = TestableS3(creds=self.creds, endpoint=self.endpoint)
196
196
 
201
201
        marker = object()
202
202
 
203
203
        def _cb(*a, **kw):
204
 
            self.assertEqual(kw['creds'], self.creds)
205
 
            self.assertEqual(kw['endpoint'], self.endpoint)
 
204
            self.assertEqual(kw["creds"], self.creds)
 
205
            self.assertEqual(kw["endpoint"], self.endpoint)
206
206
            return marker
207
207
 
208
208
        self.s3.request_factory = _cb
209
 
        self.assertIdentical(self.s3.make_request('GET'), marker)
 
209
        self.assertIdentical(self.s3.make_request("GET"), marker)
210
210
 
211
211
    def test_list_buckets(self):
212
 
        self.s3.response = samples['ListAllMyBucketsResult']
 
212
        self.s3.response = samples["ListAllMyBucketsResult"]
213
213
        d = self.s3.list_buckets()
214
214
 
215
215
        req = self.s3._lastRequest
216
216
        self.assertTrue(req.submitted)
217
 
        self.assertEqual(req.verb, 'GET')
 
217
        self.assertEqual(req.verb, "GET")
218
218
        self.assertEqual(req.bucket, None)
219
219
        self.assertEqual(req.object_name, None)
220
220
 
221
221
        def _check_result(buckets):
222
222
            self.assertEqual(
223
223
                list(buckets),
224
 
                [{'name': u'quotes',
225
 
                  'created': Time.fromDatetime(
 
224
                [{"name": u"quotes",
 
225
                  "created": Time.fromDatetime(
226
226
                    datetime(2006, 2, 3, 16, 45, 9))},
227
 
                 {'name': u'samples',
228
 
                  'created': Time.fromDatetime(
 
227
                 {"name": u"samples",
 
228
                  "created": Time.fromDatetime(
229
229
                    datetime(2006, 2, 3, 16, 41, 58))}])
230
230
        return d.addCallback(_check_result)
231
231
 
232
232
    def test_create_bucket(self):
233
 
        self.s3.create_bucket('foo')
 
233
        self.s3.create_bucket("foo")
234
234
        req = self.s3._lastRequest
235
235
        self.assertTrue(req.submitted)
236
 
        self.assertEqual(req.verb, 'PUT')
237
 
        self.assertEqual(req.bucket, 'foo')
 
236
        self.assertEqual(req.verb, "PUT")
 
237
        self.assertEqual(req.bucket, "foo")
238
238
        self.assertEqual(req.object_name, None)
239
239
 
240
240
    def test_delete_bucket(self):
241
 
        self.s3.delete_bucket('foo')
 
241
        self.s3.delete_bucket("foo")
242
242
        req = self.s3._lastRequest
243
243
        self.assertTrue(req.submitted)
244
 
        self.assertEqual(req.verb, 'DELETE')
245
 
        self.assertEqual(req.bucket, 'foo')
 
244
        self.assertEqual(req.verb, "DELETE")
 
245
        self.assertEqual(req.bucket, "foo")
246
246
        self.assertEqual(req.object_name, None)
247
247
 
248
248
    def test_put_object(self):
249
249
        self.s3.put_object(
250
 
            'foobucket', 'foo', 'data', 'text/plain', {'foo': 'bar'})
 
250
            "foobucket", "foo", "data", "text/plain", {"foo": "bar"})
251
251
        req = self.s3._lastRequest
252
252
        self.assertTrue(req.submitted)
253
 
        self.assertEqual(req.verb, 'PUT')
254
 
        self.assertEqual(req.bucket, 'foobucket')
255
 
        self.assertEqual(req.object_name, 'foo')
256
 
        self.assertEqual(req.data, 'data')
257
 
        self.assertEqual(req.content_type, 'text/plain')
258
 
        self.assertEqual(req.metadata, {'foo': 'bar'})
 
253
        self.assertEqual(req.verb, "PUT")
 
254
        self.assertEqual(req.bucket, "foobucket")
 
255
        self.assertEqual(req.object_name, "foo")
 
256
        self.assertEqual(req.data, "data")
 
257
        self.assertEqual(req.content_type, "text/plain")
 
258
        self.assertEqual(req.metadata, {"foo": "bar"})
259
259
 
260
260
    def test_get_object(self):
261
 
        self.s3.get_object('foobucket', 'foo')
 
261
        self.s3.get_object("foobucket", "foo")
262
262
        req = self.s3._lastRequest
263
263
        self.assertTrue(req.submitted)
264
 
        self.assertEqual(req.verb, 'GET')
265
 
        self.assertEqual(req.bucket, 'foobucket')
266
 
        self.assertEqual(req.object_name, 'foo')
 
264
        self.assertEqual(req.verb, "GET")
 
265
        self.assertEqual(req.bucket, "foobucket")
 
266
        self.assertEqual(req.object_name, "foo")
267
267
 
268
268
    def test_head_object(self):
269
 
        self.s3.head_object('foobucket', 'foo')
 
269
        self.s3.head_object("foobucket", "foo")
270
270
        req = self.s3._lastRequest
271
271
        self.assertTrue(req.submitted)
272
 
        self.assertEqual(req.verb, 'HEAD')
273
 
        self.assertEqual(req.bucket, 'foobucket')
274
 
        self.assertEqual(req.object_name, 'foo')
 
272
        self.assertEqual(req.verb, "HEAD")
 
273
        self.assertEqual(req.bucket, "foobucket")
 
274
        self.assertEqual(req.object_name, "foo")
275
275
 
276
276
    def test_delete_object(self):
277
 
        self.s3.delete_object('foobucket', 'foo')
 
277
        self.s3.delete_object("foobucket", "foo")
278
278
        req = self.s3._lastRequest
279
279
        self.assertTrue(req.submitted)
280
 
        self.assertEqual(req.verb, 'DELETE')
281
 
        self.assertEqual(req.bucket, 'foobucket')
282
 
        self.assertEqual(req.object_name, 'foo')
 
280
        self.assertEqual(req.verb, "DELETE")
 
281
        self.assertEqual(req.bucket, "foobucket")
 
282
        self.assertEqual(req.object_name, "foo")
283
283
 
284
284
 
285
285
class MiscellaneousTests(TXAWSTestCase):
286
286
 
287
287
    def test_contentMD5(self):
288
 
        self.assertEqual(calculate_md5('somedata'), 'rvr3UC1SmUw7AZV2NqPN0g==')
 
288
        self.assertEqual(calculate_md5("somedata"), "rvr3UC1SmUw7AZV2NqPN0g==")