/usr/share/pyshared/cloudservers/client.py is in python-rackspace-cloudservers 1.0~a5-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | import httplib2
try:
import json
except ImportError:
import simplejson as json
import cloudservers
from . import exceptions
class CloudServersClient(httplib2.Http):
AUTH_URL = 'https://auth.api.rackspacecloud.com/v1.0'
USER_AGENT = 'python-cloudservers/%s' % cloudservers.__version__
def __init__(self, user, apikey):
super(CloudServersClient, self).__init__()
self.user = user
self.apikey = apikey
self.management_url = None
self.auth_token = None
# httplib2 overrides
self.force_exception_to_status_code = True
def request(self, *args, **kwargs):
kwargs.setdefault('headers', {})
kwargs['headers']['User-Agent'] = self.USER_AGENT
if 'body' in kwargs:
kwargs['headers']['Content-Type'] = 'application/json'
kwargs['body'] = json.dumps(kwargs['body'])
resp, body = super(CloudServersClient, self).request(*args, **kwargs)
body = json.loads(body) if body else None
if resp.status in (400, 401, 403, 404, 413, 500):
raise exceptions.from_response(resp, body)
return resp, body
def _cs_request(self, url, method, **kwargs):
if not self.management_url:
self.authenticate()
# Perform the request once. If we get a 401 back then it
# might be because the auth token expired, so try to
# re-authenticate and try again. If it still fails, bail.
try:
kwargs.setdefault('headers', {})['X-Auth-Token'] = self.auth_token
resp, body = self.request(self.management_url + url, method, **kwargs)
return resp, body
except exceptions.Unauthorized, ex:
try:
self.authenticate()
resp, body = self.request(self.management_url + url, method, **kwargs)
return resp, body
except exceptions.Unauthorized:
raise ex
def get(self, url, **kwargs):
# The Rackspace API returns cached results by default.
# We like our responses nice and fresh, though, so we
# stick a fake GET parameter on the URL.
if not '?' in url:
url += '?fresh'
return self._cs_request(url, 'GET', **kwargs)
def post(self, url, **kwargs):
return self._cs_request(url, 'POST', **kwargs)
def put(self, url, **kwargs):
return self._cs_request(url, 'PUT', **kwargs)
def delete(self, url, **kwargs):
return self._cs_request(url, 'DELETE', **kwargs)
def authenticate(self):
headers = {'X-Auth-User': self.user, 'X-Auth-Key': self.apikey}
resp, body = self.request(self.AUTH_URL, 'GET', headers=headers)
self.management_url = resp['x-server-management-url']
self.auth_token = resp['x-auth-token']
|