<s> import copy <EOL> import os <EOL> import time <EOL> import base64 <EOL> import hmac <EOL> from hashlib import sha256 <EOL> from libcloud . utils . py3 import httplib <EOL> from libcloud . utils . py3 import b <EOL> from libcloud . utils . xml import fixxpath <EOL> try : <EOL> from lxml import etree as ET <EOL> except ImportError : <EOL> from xml . etree import ElementTree as ET <EOL> from libcloud . common . types import InvalidCredsError <EOL> from libcloud . common . types import LibcloudError , MalformedResponseError <EOL> from libcloud . common . base import ConnectionUserAndKey , RawResponse <EOL> from libcloud . common . base import CertificateConnection <EOL> from libcloud . common . base import XmlResponse <EOL> API_VERSION = '<STR_LIT>' <EOL> AZURE_TIME_FORMAT = '<STR_LIT>' <EOL> class AzureRedirectException ( Exception ) : <EOL> def __init__ ( self , response ) : <EOL> self . location = response . headers [ '<STR_LIT:location>' ] <EOL> class AzureResponse ( XmlResponse ) : <EOL> valid_response_codes = [ <EOL> httplib . NOT_FOUND , <EOL> httplib . CONFLICT , <EOL> httplib . BAD_REQUEST , <EOL> httplib . TEMPORARY_REDIRECT <EOL> ] <EOL> def success ( self ) : <EOL> i = int ( self . status ) <EOL> return <NUM_LIT:200> <= i <= <NUM_LIT> or i in self . valid_response_codes <EOL> def parse_error ( self , msg = None ) : <EOL> error_msg = '<STR_LIT>' <EOL> try : <EOL> body = self . parse_body ( ) <EOL> if type ( body ) == ET . Element : <EOL> code = body . findtext ( fixxpath ( xpath = '<STR_LIT>' ) ) <EOL> message = body . findtext ( fixxpath ( xpath = '<STR_LIT>' ) ) <EOL> message = message . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] <EOL> error_msg = '<STR_LIT>' % ( code , message ) <EOL> except MalformedResponseError : <EOL> pass <EOL> if msg : <EOL> error_msg = '<STR_LIT>' % ( msg , error_msg ) <EOL> if self . status in [ httplib . UNAUTHORIZED , httplib . FORBIDDEN ] : <EOL> raise InvalidCredsError ( error_msg ) <EOL> raise LibcloudError ( <EOL> '<STR_LIT>' % ( error_msg , self . status ) , <EOL> driver = self <EOL> ) <EOL> def parse_body ( self ) : <EOL> is_redirect = int ( self . status ) == httplib . TEMPORARY_REDIRECT <EOL> if is_redirect and self . connection . driver . follow_redirects : <EOL> raise AzureRedirectException ( self ) <EOL> else : <EOL> return super ( AzureResponse , self ) . parse_body ( ) <EOL> class AzureRawResponse ( RawResponse ) : <EOL> pass <EOL> class AzureConnection ( ConnectionUserAndKey ) : <EOL> """<STR_LIT>""" <EOL> responseCls = AzureResponse <EOL> rawResponseCls = AzureRawResponse <EOL> def add_default_params ( self , params ) : <EOL> return params <EOL> def pre_connect_hook ( self , params , headers ) : <EOL> headers = copy . deepcopy ( headers ) <EOL> headers [ '<STR_LIT>' ] = time . strftime ( AZURE_TIME_FORMAT , time . gmtime ( ) ) <EOL> headers [ '<STR_LIT>' ] = API_VERSION <EOL> headers [ '<STR_LIT>' ] = self . _get_azure_auth_signature ( <EOL> method = self . method , <EOL> headers = headers , <EOL> params = params , <EOL> account = self . user_id , <EOL> secret_key = self . key , <EOL> path = self . action <EOL> ) <EOL> headers . pop ( '<STR_LIT>' , None ) <EOL> return params , headers <EOL> def _get_azure_auth_signature ( self , <EOL> method , <EOL> headers , <EOL> params , <EOL> account , <EOL> secret_key , <EOL> path = '<STR_LIT:/>' ) : <EOL> """<STR_LIT>""" <EOL> special_header_values = [ ] <EOL> xms_header_values = [ ] <EOL> param_list = [ ] <EOL> special_header_keys = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:date>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> headers_copy = { } <EOL> for header , value in headers . items ( ) : <EOL> header = header . lower ( ) <EOL> value = str ( value ) . strip ( ) <EOL> if header . startswith ( '<STR_LIT>' ) : <EOL> xms_header_values . append ( ( header , value ) ) <EOL> else : <EOL> headers_copy [ header ] = value <EOL> for header in special_header_keys : <EOL> header = header . lower ( ) <EOL> if header in headers_copy : <EOL> special_header_values . append ( headers_copy [ header ] ) <EOL> else : <EOL> special_header_values . append ( '<STR_LIT>' ) <EOL> values_to_sign = [ method ] + special_header_values <EOL> xms_header_values . sort ( ) <EOL> for header , value in xms_header_values : <EOL> values_to_sign . append ( '<STR_LIT>' % ( header , value ) ) <EOL> values_to_sign . append ( '<STR_LIT>' % ( account , path ) ) <EOL> for key , value in params . items ( ) : <EOL> param_list . append ( ( key . lower ( ) , str ( value ) . strip ( ) ) ) <EOL> param_list . sort ( ) <EOL> for key , value in param_list : <EOL> values_to_sign . append ( '<STR_LIT>' % ( key , value ) ) <EOL> string_to_sign = b ( '<STR_LIT:\n>' . join ( values_to_sign ) ) <EOL> secret_key = b ( secret_key ) <EOL> b64_hmac = base64 . b64encode ( <EOL> hmac . new ( secret_key , string_to_sign , digestmod = sha256 ) . digest ( ) <EOL> ) <EOL> return '<STR_LIT>' % ( self . user_id , b64_hmac . decode ( '<STR_LIT:utf-8>' ) ) <EOL> class AzureBaseDriver ( object ) : <EOL> name = "<STR_LIT>" <EOL> class AzureServiceManagementConnection ( CertificateConnection ) : <EOL> """<STR_LIT>""" <EOL> driver = AzureBaseDriver <EOL> responseCls = AzureResponse <EOL> rawResponseCls = AzureRawResponse <EOL> name = '<STR_LIT>' <EOL> host = '<STR_LIT>' <EOL> keyfile = "<STR_LIT>" <EOL> def __init__ ( self , subscription_id , key_file , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> super ( AzureServiceManagementConnection , self ) . __init__ ( <EOL> key_file , <EOL> * args , <EOL> ** kwargs <EOL> ) <EOL> self . subscription_id = subscription_id <EOL> keypath = os . path . expanduser ( key_file ) <EOL> self . keyfile = keypath <EOL> is_file_path = os . path . exists ( keypath ) and os . path . isfile ( keypath ) <EOL> if not is_file_path : <EOL> raise InvalidCredsError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> self . key_file = key_file <EOL> def add_default_headers ( self , headers ) : <EOL> """<STR_LIT>""" <EOL> headers [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> headers [ '<STR_LIT>' ] = time . strftime ( AZURE_TIME_FORMAT , time . gmtime ( ) ) <EOL> return headers </s>
