~tribaal/txaws/xss-hardening

« back to all changes in this revision

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

  • Committer: Tristan Seligmann
  • Date: 2008-07-06 22:51:54 UTC
  • Revision ID: mithrandi@mithrandi.net-20080706225154-35ok5ivzh7h3ve59
Initial import of S3 code from EdgeVerse.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
from datetime import datetime
 
2
 
 
3
from epsilon.extime import Time
 
4
 
 
5
from twisted.trial.unittest import TestCase
 
6
from twisted.internet.defer import succeed
 
7
 
 
8
from aws.s3.client import S3, S3Request, calculateMD5, hmac_sha1
 
9
 
 
10
class StubbedS3Request(S3Request):
 
11
    def getPage(self, url, method, postdata, headers):
 
12
        self.getPageArgs = (url, method, postdata, headers)
 
13
        return succeed('')
 
14
 
 
15
 
 
16
class RequestTests(TestCase):
 
17
    accessKey = '0PN5J17HBGZHT7JJ3X82'
 
18
    secretKey = 'uV3F3YluFJax1cknvbcGwgjvx4QpvB+leU8dUj2o'
 
19
 
 
20
    def test_objectRequest(self):
 
21
        """
 
22
        Test that a request addressing an object is created correctly.
 
23
        """
 
24
        DATA = 'objectData'
 
25
        DIGEST = 'zhdB6gwvocWv/ourYUWMxA=='
 
26
 
 
27
        request = S3Request('PUT', 'somebucket', 'object/name/here', DATA, contentType='text/plain', metadata={'foo': 'bar'})
 
28
        self.assertEqual(request.verb, 'PUT')
 
29
        self.assertEqual(request.getURI(), 'https://s3.amazonaws.com/somebucket/object/name/here')
 
30
        headers = request.getHeaders()
 
31
        self.assertNotEqual(headers.pop('Date'), '')
 
32
        self.assertEqual(headers,
 
33
                         {'Content-Type': 'text/plain',
 
34
                          'Content-Length': len(DATA),
 
35
                          'Content-MD5': DIGEST,
 
36
                          'x-amz-meta-foo': 'bar'})
 
37
        self.assertEqual(request.data, 'objectData')
 
38
 
 
39
    def test_bucketRequest(self):
 
40
        """
 
41
        Test that a request addressing a bucket is created correctly.
 
42
        """
 
43
        DIGEST = '1B2M2Y8AsgTpgAmY7PhCfg=='
 
44
 
 
45
        request = S3Request('GET', 'somebucket')
 
46
        self.assertEqual(request.verb, 'GET')
 
47
        self.assertEqual(request.getURI(), 'https://s3.amazonaws.com/somebucket')
 
48
        headers = request.getHeaders()
 
49
        self.assertNotEqual(headers.pop('Date'), '')
 
50
        self.assertEqual(headers,
 
51
                         {'Content-Length': 0,
 
52
                          'Content-MD5': DIGEST})
 
53
        self.assertEqual(request.data, '')
 
54
 
 
55
    def test_submit(self):
 
56
        """
 
57
        Submitting the request should invoke getPage correctly.
 
58
        """
 
59
        request = StubbedS3Request('GET', 'somebucket')
 
60
 
 
61
        def _postCheck(result):
 
62
            self.assertEqual(result, '')
 
63
 
 
64
            url, method, postdata, headers = request.getPageArgs
 
65
            self.assertEqual(url, request.getURI())
 
66
            self.assertEqual(method, request.verb)
 
67
            self.assertEqual(postdata, request.data)
 
68
            self.assertEqual(headers, request.getHeaders())
 
69
 
 
70
        return request.submit().addCallback(_postCheck)
 
71
 
 
72
    def test_invalidAuthenticatedRequest(self):
 
73
        """
 
74
        An authenticated request must be supplied both private and public keys.
 
75
        """
 
76
        self.assertRaises(ValueError, S3Request, 'GET', accessKey='foo')
 
77
        self.assertRaises(ValueError, S3Request, 'GET', secretKey='foo')
 
78
 
 
79
    def test_authenticationTestCases(self):
 
80
        req = S3Request('GET', accessKey=self.accessKey, secretKey=self.secretKey)
 
81
        req.date = 'Wed, 28 Mar 2007 01:29:59 +0000'
 
82
 
 
83
        headers = req.getHeaders()
 
84
        self.assertEqual(headers['Authorization'], 'AWS 0PN5J17HBGZHT7JJ3X82:jF7L3z/FTV47vagZzhKupJ9oNig=')
 
85
 
 
86
 
 
87
class InertRequest(S3Request):
 
88
    """
 
89
    Inert version of S3Request.
 
90
 
 
91
    The submission action is stubbed out to return the provided response.
 
92
    """
 
93
    def __init__(self, *a, **kw):
 
94
        self.response = kw.pop('response')
 
95
        super(InertRequest, self).__init__(*a, **kw)
 
96
 
 
97
    def submit(self):
 
98
        """
 
99
        Return the canned result instead of performing a network operation.
 
100
        """
 
101
        return succeed(self.response)
 
102
 
 
103
 
 
104
class TestableS3(S3):
 
105
    """
 
106
    Testable version of S3.
 
107
 
 
108
    This subclass stubs requestFactory to use InertRequest, making it easy to
 
109
    assert things about the requests that are created in response to various
 
110
    operations.
 
111
    """
 
112
    response = None
 
113
 
 
114
    def requestFactory(self, *a, **kw):
 
115
        req = InertRequest(response=self.response, *a, **kw)
 
116
        self._lastRequest = req
 
117
        return req
 
118
 
 
119
 
 
120
samples = {
 
121
    'ListAllMyBucketsResult':
 
122
    """<?xml version="1.0" encoding="UTF-8"?>
 
123
<ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
 
124
  <Owner>
 
125
    <ID>bcaf1ffd86f41caff1a493dc2ad8c2c281e37522a640e161ca5fb16fd081034f</ID>
 
126
    <DisplayName>webfile</DisplayName>
 
127
  </Owner>
 
128
  <Buckets>
 
129
    <Bucket>
 
130
      <Name>quotes</Name>
 
131
      <CreationDate>2006-02-03T16:45:09.000Z</CreationDate>
 
132
    </Bucket>
 
133
    <Bucket>
 
134
      <Name>samples</Name>
 
135
      <CreationDate>2006-02-03T16:41:58.000Z</CreationDate>
 
136
    </Bucket>
 
137
  </Buckets>
 
138
</ListAllMyBucketsResult>""",
 
139
    }
 
140
 
 
141
 
 
142
class WrapperTests(TestCase):
 
143
    def setUp(self):
 
144
        self.s3 = TestableS3(accessKey='accessKey', secretKey='secretKey')
 
145
 
 
146
    def test_makeRequest(self):
 
147
        """
 
148
        Test that makeRequest passes in the service credentials.
 
149
        """
 
150
        marker = object()
 
151
 
 
152
        def _cb(*a, **kw):
 
153
            self.assertEqual(kw['accessKey'], 'accessKey')
 
154
            self.assertEqual(kw['secretKey'], 'secretKey')
 
155
            return marker
 
156
 
 
157
        self.s3.requestFactory = _cb
 
158
        self.assertIdentical(self.s3.makeRequest('GET'), marker)
 
159
 
 
160
    def test_listBuckets(self):
 
161
        self.s3.response = samples['ListAllMyBucketsResult']
 
162
        d = self.s3.listBuckets()
 
163
 
 
164
        req = self.s3._lastRequest
 
165
        self.assertEqual(req.verb, 'GET')
 
166
        self.assertEqual(req.bucket, None)
 
167
        self.assertEqual(req.objectName, None)
 
168
 
 
169
        def _checkResult(buckets):
 
170
            self.assertEqual(list(buckets),
 
171
                             [{'name': u'quotes',
 
172
                               'created': Time.fromDatetime(datetime(2006, 2, 3, 16, 45, 9))},
 
173
                              {'name': u'samples',
 
174
                               'created': Time.fromDatetime(datetime(2006, 2, 3, 16, 41, 58))}])
 
175
        return d.addCallback(_checkResult)
 
176
 
 
177
    def test_createBucket(self):
 
178
        self.s3.createBucket('foo')
 
179
        req = self.s3._lastRequest
 
180
        self.assertEqual(req.verb, 'PUT')
 
181
        self.assertEqual(req.bucket, 'foo')
 
182
        self.assertEqual(req.objectName, None)
 
183
 
 
184
    def test_deleteBucket(self):
 
185
        self.s3.deleteBucket('foo')
 
186
        req = self.s3._lastRequest
 
187
        self.assertEqual(req.verb, 'DELETE')
 
188
        self.assertEqual(req.bucket, 'foo')
 
189
        self.assertEqual(req.objectName, None)
 
190
 
 
191
    def test_putObject(self):
 
192
        self.s3.putObject('foobucket', 'foo', 'data', 'text/plain', {'foo': 'bar'})
 
193
        req = self.s3._lastRequest
 
194
        self.assertEqual(req.verb, 'PUT')
 
195
        self.assertEqual(req.bucket, 'foobucket')
 
196
        self.assertEqual(req.objectName, 'foo')
 
197
        self.assertEqual(req.data, 'data')
 
198
        self.assertEqual(req.contentType, 'text/plain')
 
199
        self.assertEqual(req.metadata, {'foo': 'bar'})
 
200
 
 
201
    def test_getObject(self):
 
202
        self.s3.getObject('foobucket', 'foo')
 
203
        req = self.s3._lastRequest
 
204
        self.assertEqual(req.verb, 'GET')
 
205
        self.assertEqual(req.bucket, 'foobucket')
 
206
        self.assertEqual(req.objectName, 'foo')
 
207
 
 
208
    def test_headObject(self):
 
209
        self.s3.headObject('foobucket', 'foo')
 
210
        req = self.s3._lastRequest
 
211
        self.assertEqual(req.verb, 'HEAD')
 
212
        self.assertEqual(req.bucket, 'foobucket')
 
213
        self.assertEqual(req.objectName, 'foo')
 
214
 
 
215
    def test_deleteObject(self):
 
216
        self.s3.deleteObject('foobucket', 'foo')
 
217
        req = self.s3._lastRequest
 
218
        self.assertEqual(req.verb, 'DELETE')
 
219
        self.assertEqual(req.bucket, 'foobucket')
 
220
        self.assertEqual(req.objectName, 'foo')
 
221
 
 
222
 
 
223
class MiscellaneousTests(TestCase):
 
224
    def test_contentMD5(self):
 
225
        self.assertEqual(calculateMD5('somedata'), 'rvr3UC1SmUw7AZV2NqPN0g==')
 
226
 
 
227
    def test_hmac_sha1(self):
 
228
        cases = [
 
229
            ('0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b'.decode('hex'), 'Hi There', 'thcxhlUFcmTii8C2+zeMjvFGvgA='),
 
230
            ('Jefe', 'what do ya want for nothing?', '7/zfauXrL6LSdBbV8YTfnCWafHk='),
 
231
            ('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'.decode('hex'), '\xdd' * 50, 'El1zQrmsEc2Ro5r0iqF7T2PxddM='),
 
232
            ]
 
233
 
 
234
        for key, data, expected in cases:
 
235
            self.assertEqual(hmac_sha1(key, data), expected)