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
|
# coding=utf-8
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
import time
import keyring
import ast
import base64
import hmac
import hashlib
import datetime
import requests
from requests.auth import AuthBase
from msrest.authentication import Authentication
from msrest import Serializer
from msrest.serialization import TZ_UTC
try:
from urlparse import urlparse, parse_qs
except ImportError:
from urllib.parse import urlparse, parse_qs
class SharedKeyAuth(AuthBase):
headers_to_sign = [
'content-encoding',
'content-language',
'content-length',
'content-md5',
'content-type',
'date',
'if-modified-since',
'if-match',
'if-none-match',
'if-unmodified-since',
'range']
def __init__(self, header, account_name, key):
self._header = header
self._account_name = account_name
self._key = key
def __call__(self, request):
if not request.headers.get('ocp-date'):
now = datetime.datetime.utcnow()
now = now.replace(tzinfo=TZ_UTC)
request.headers['ocp-date'] = Serializer.serialize_rfc(now)
url = urlparse(request.url)
uri_path = url.path
uri_path = uri_path.replace('%5C', '/')
uri_path = uri_path.replace('%2F', '/')
# method to sign
string_to_sign = request.method + '\n'
# get headers to sign
request_header_dict = {
key.lower(): val for key, val in request.headers.items() if val}
request_headers = [
str(request_header_dict.get(x, '')) for x in self.headers_to_sign]
string_to_sign += '\n'.join(request_headers) + '\n'
# get ocp- header to sign
ocp_headers = []
for name, value in request.headers.items():
if 'ocp-' in name and value:
ocp_headers.append((name.lower(), value))
for name, value in sorted(ocp_headers):
string_to_sign += "{}:{}\n".format(name, value)
# get account_name and uri path to sign
string_to_sign += "/{}{}".format(self._account_name, uri_path)
# get query string to sign if it is not table service
query_to_sign = parse_qs(url.query)
for name in sorted(query_to_sign.keys()):
value = query_to_sign[name][0]
if value:
string_to_sign += "\n{}:{}".format(name, value)
# sign the request
auth_string = "SharedKey {}:{}".format(
self._account_name, self._sign_string(string_to_sign))
request.headers[self._header] = auth_string
return request
def _sign_string(self, string_to_sign):
_key = self._key.encode('utf-8')
string_to_sign = string_to_sign.encode('utf-8')
try:
key = base64.b64decode(_key)
except TypeError:
raise ValueError("Invalid key value: {}".format(self._key))
signed_hmac_sha256 = hmac.HMAC(key, string_to_sign, hashlib.sha256)
digest = signed_hmac_sha256.digest()
return base64.b64encode(digest).decode('utf-8')
class SharedKeyCredentials(Authentication):
def __init__(self, account_name, key):
super(SharedKeyCredentials, self).__init__()
self.auth = SharedKeyAuth(self.header, account_name, key)
def signed_session(self):
session = super(SharedKeyCredentials, self).signed_session()
session.auth = self.auth
return session
|
