1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
|
try:
from xml.etree.ElementTree import ParseError
except ImportError:
from xml.parsers.expat import ExpatError as ParseError
import warnings
from StringIO import StringIO
from twisted.internet.ssl import ClientContextFactory
from twisted.internet.protocol import Protocol
from twisted.internet.defer import Deferred, succeed, fail
from twisted.python import failure
from twisted.web import http
from twisted.web.iweb import UNKNOWN_LENGTH
from twisted.web.client import HTTPClientFactory
from twisted.web.client import Agent
from twisted.web.client import ResponseDone
from twisted.web.http import NO_CONTENT
from twisted.web.http_headers import Headers
from twisted.web.error import Error as TwistedWebError
try:
from twisted.web.client import FileBodyProducer
except ImportError:
from txaws.client._producers import FileBodyProducer
from txaws.util import parse
from txaws.credentials import AWSCredentials
from txaws.exception import AWSResponseParseError
from txaws.service import AWSServiceEndpoint
from txaws.client.ssl import VerifyingContextFactory
def error_wrapper(error, errorClass):
"""
We want to see all error messages from cloud services. Amazon's EC2 says
that their errors are accompanied either by a 400-series or 500-series HTTP
response code. As such, the first thing we want to do is check to see if
the error is in that range. If it is, we then need to see if the error
message is an EC2 one.
In the event that an error is not a Twisted web error nor an EC2 one, the
original exception is raised.
"""
http_status = 0
if error.check(TwistedWebError):
xml_payload = error.value.response
if error.value.status:
http_status = int(error.value.status)
else:
error.raiseException()
if http_status >= 400:
if not xml_payload:
error.raiseException()
try:
fallback_error = errorClass(
xml_payload, error.value.status, str(error.value),
error.value.response)
except (ParseError, AWSResponseParseError):
error_message = http.RESPONSES.get(http_status)
fallback_error = TwistedWebError(
http_status, error_message, error.value.response)
raise fallback_error
elif 200 <= http_status < 300:
return str(error.value)
else:
error.raiseException()
class BaseClient(object):
"""Create an AWS client.
@param creds: User authentication credentials to use.
@param endpoint: The service endpoint URI.
@param query_factory: The class or function that produces a query
object for making requests to the EC2 service.
@param parser: A parser object for parsing responses from the EC2 service.
@param receiver_factory: Factory for receiving responses from EC2 service.
"""
def __init__(self, creds=None, endpoint=None, query_factory=None,
parser=None, receiver_factory=None):
if creds is None:
creds = AWSCredentials()
if endpoint is None:
endpoint = AWSServiceEndpoint()
self.creds = creds
self.endpoint = endpoint
self.query_factory = query_factory
self.receiver_factory = receiver_factory
self.parser = parser
class StreamingError(Exception):
"""
Raised if more data or less data is received than expected.
"""
class StreamingBodyReceiver(Protocol):
"""
Streaming HTTP response body receiver.
TODO: perhaps there should be an interface specifying why
finished (Deferred) and content_length are necessary and
how to used them; eg. callback/errback finished on completion.
"""
finished = None
content_length = None
def __init__(self, fd=None, readback=True):
"""
@param fd: a file descriptor to write to
@param readback: if True read back data from fd to callback finished
with, otherwise we call back finish with fd itself
with
"""
if fd is None:
fd = StringIO()
self._fd = fd
self._received = 0
self._readback = readback
def dataReceived(self, bytes):
streaming = self.content_length is UNKNOWN_LENGTH
if not streaming and (self._received > self.content_length):
self.transport.loseConnection()
raise StreamingError(
"Buffer overflow - received more data than "
"Content-Length dictated: %d" % self.content_length)
# TODO should be some limit on how much we receive
self._fd.write(bytes)
self._received += len(bytes)
def connectionLost(self, reason):
reason.trap(ResponseDone)
d = self.finished
self.finished = None
streaming = self.content_length is UNKNOWN_LENGTH
if streaming or (self._received == self.content_length):
if self._readback:
self._fd.seek(0)
data = self._fd.read()
self._fd.close()
self._fd = None
d.callback(data)
else:
d.callback(self._fd)
else:
f = failure.Failure(StreamingError("Connection lost before "
"receiving all data"))
d.errback(f)
class WebClientContextFactory(ClientContextFactory):
def getContext(self, hostname, port):
return ClientContextFactory.getContext(self)
class WebVerifyingContextFactory(VerifyingContextFactory):
def getContext(self, hostname, port):
return VerifyingContextFactory.getContext(self)
class FakeClient(object):
"""
XXX
A fake client object for some degree of backwards compatability for
code using the client attibute on BaseQuery to check url, status
etc.
"""
url = None
status = None
class BaseQuery(object):
def __init__(self, action=None, creds=None, endpoint=None, reactor=None,
body_producer=None, receiver_factory=None):
if not action:
raise TypeError("The query requires an action parameter.")
self.action = action
self.creds = creds
self.endpoint = endpoint
if reactor is None:
from twisted.internet import reactor
self.reactor = reactor
self._client = None
self.request_headers = None
self.response_headers = None
self.body_producer = body_producer
self.receiver_factory = receiver_factory or StreamingBodyReceiver
@property
def client(self):
if self._client is None:
self._client_deprecation_warning()
self._client = FakeClient()
return self._client
@client.setter
def client(self, value):
self._client_deprecation_warning()
self._client = value
def _client_deprecation_warning(self):
warnings.warn('The client attribute on BaseQuery is deprecated and'
' will go away in future release.')
def get_page(self, url, *args, **kwds):
"""
Define our own get_page method so that we can easily override the
factory when we need to. This was copied from the following:
* twisted.web.client.getPage
* twisted.web.client._makeGetterFactory
"""
contextFactory = None
scheme, host, port, path = parse(url)
data = kwds.get('postdata', None)
self._method = method = kwds.get('method', 'GET')
self.request_headers = self._headers(kwds.get('headers', {}))
if (self.body_producer is None) and (data is not None):
self.body_producer = FileBodyProducer(StringIO(data))
if scheme == "https":
if self.endpoint.ssl_hostname_verification:
contextFactory = WebVerifyingContextFactory(host)
else:
contextFactory = WebClientContextFactory()
agent = Agent(self.reactor, contextFactory)
self.client.url = url
d = agent.request(method, url, self.request_headers,
self.body_producer)
else:
agent = Agent(self.reactor)
d = agent.request(method, url, self.request_headers,
self.body_producer)
d.addCallback(self._handle_response)
return d
def _headers(self, headers_dict):
"""
Convert dictionary of headers into twisted.web.client.Headers object.
"""
return Headers(dict((k,[v]) for (k,v) in headers_dict.items()))
def _unpack_headers(self, headers):
"""
Unpack twisted.web.client.Headers object to dict. This is to provide
backwards compatability.
"""
return dict((k,v[0]) for (k,v) in headers.getAllRawHeaders())
def get_request_headers(self, *args, **kwds):
"""
A convenience method for obtaining the headers that were sent to the
S3 server.
The AWS S3 API depends upon setting headers. This method is provided as
a convenience for debugging issues with the S3 communications.
"""
if self.request_headers:
return self._unpack_headers(self.request_headers)
def _handle_response(self, response):
"""
Handle the HTTP response by memoing the headers and then delivering
bytes.
"""
self.client.status = response.code
self.response_headers = headers = response.headers
# XXX This workaround (which needs to be improved at that) for possible
# bug in Twisted with new client:
# http://twistedmatrix.com/trac/ticket/5476
if self._method.upper() == 'HEAD' or response.code == NO_CONTENT:
return succeed('')
receiver = self.receiver_factory()
receiver.finished = d = Deferred()
receiver.content_length = response.length
response.deliverBody(receiver)
if response.code >= 400:
d.addCallback(self._fail_response, response)
return d
def _fail_response(self, data, response):
return fail(failure.Failure(
TwistedWebError(response.code, response=data)))
def get_response_headers(self, *args, **kwargs):
"""
A convenience method for obtaining the headers that were sent from the
S3 server.
The AWS S3 API depends upon setting headers. This method is used by the
head_object API call for getting a S3 object's metadata.
"""
if self.response_headers:
return self._unpack_headers(self.response_headers)
|