forked from miguelgrinberg/Flask-HTTPAuth
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathflask_httpauth.py
190 lines (156 loc) · 6.17 KB
/
flask_httpauth.py
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
"""
flask_httpauth
==================
This module provides Basic and Digest HTTP authentication for Flask routes.
:copyright: (C) 2014 by Miguel Grinberg.
:license: BSD, see LICENSE for more details.
"""
from functools import wraps
from hashlib import md5
from random import Random, SystemRandom
from flask import request, make_response, session
class HTTPAuth(object):
def __init__(self):
def default_get_password(username):
return None
def default_auth_error():
return "Unauthorized Access"
self.realm = "Authentication Required"
self.get_password(default_get_password)
self.error_handler(default_auth_error)
def get_password(self, f):
self.get_password_callback = f
return f
def error_handler(self, f):
@wraps(f)
def decorated(*args, **kwargs):
res = f(*args, **kwargs)
if type(res) == str:
res = make_response(res)
res.status_code = 401
if 'WWW-Authenticate' not in res.headers.keys():
res.headers['WWW-Authenticate'] = self.authenticate_header()
return res
self.auth_error_callback = decorated
return decorated
def login_required(self, f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
# We need to ignore authentication headers for OPTIONS to avoid
# unwanted interactions with CORS.
# Chrome and Firefox issue a preflight OPTIONS request to check
# Access-Control-* headers, and will fail if it returns 401.
if request.method != 'OPTIONS':
if auth:
password = self.get_password_callback(auth.username)
else:
password = None
if not self.authenticate(auth, password):
return self.auth_error_callback()
return f(*args, **kwargs)
return decorated
def username(self):
if not request.authorization:
return ""
return request.authorization.username
class HTTPBasicAuth(HTTPAuth):
def __init__(self):
super(HTTPBasicAuth, self).__init__()
self.hash_password(None)
self.verify_password(None)
def hash_password(self, f):
self.hash_password_callback = f
return f
def verify_password(self, f):
self.verify_password_callback = f
return f
def authenticate_header(self):
return 'Basic realm="{0}"'.format(self.realm)
def authenticate(self, auth, stored_password):
if auth:
username = auth.username
client_password = auth.password
else:
username = ""
client_password = ""
if self.verify_password_callback:
return self.verify_password_callback(username, client_password)
if not auth:
return False
if self.hash_password_callback:
try:
client_password = self.hash_password_callback(client_password)
except TypeError:
client_password = self.hash_password_callback(username,
client_password)
return client_password == stored_password
class HTTPDigestAuth(HTTPAuth):
def __init__(self, use_ha1_pw = False):
super(HTTPDigestAuth, self).__init__()
self.use_ha1_pw = use_ha1_pw
self.random = SystemRandom()
try:
self.random.random()
except NotImplementedError:
self.random = Random()
def _generate_random():
return md5(str(self.random.random()).encode('utf-8')).hexdigest()
def default_generate_nonce():
session["auth_nonce"] = _generate_random()
return session["auth_nonce"]
def default_verify_nonce(nonce):
return nonce == session.get("auth_nonce")
def default_generate_opaque():
session["auth_opaque"] = _generate_random()
return session["auth_opaque"]
def default_verify_opaque(opaque):
return opaque == session.get("auth_opaque")
self.generate_nonce(default_generate_nonce)
self.generate_opaque(default_generate_opaque)
self.verify_nonce(default_verify_nonce)
self.verify_opaque(default_verify_opaque)
def generate_nonce(self, f):
self.generate_nonce_callback = f
return f
def verify_nonce(self, f):
self.verify_nonce_callback = f
return f
def generate_opaque(self, f):
self.generate_opaque_callback = f
return f
def verify_opaque(self, f):
self.verify_opaque_callback = f
return f
def get_nonce(self):
return self.generate_nonce_callback()
def get_opaque(self):
return self.generate_opaque_callback()
def generate_ha1(self, username, password):
a1 = username + ":" + self.realm + ":" + password
a1 = a1.encode('utf-8')
return md5(a1).hexdigest()
def authenticate_header(self):
session["auth_nonce"] = self.get_nonce()
session["auth_opaque"] = self.get_opaque()
return 'Digest realm="{0}",nonce="{1}",opaque="{2}"'.format(
self.realm, session["auth_nonce"], session["auth_opaque"])
def authenticate(self, auth, stored_password_or_ha1):
if not auth or not auth.username or not auth.realm or not auth.uri \
or not auth.nonce or not auth.response \
or not stored_password_or_ha1:
return False
if not(self.verify_nonce_callback(auth.nonce)) or \
not(self.verify_opaque_callback(auth.opaque)):
return False
if self.use_ha1_pw:
ha1 = stored_password_or_ha1
else:
a1 = auth.username + ":" + auth.realm + ":" + \
stored_password_or_ha1
ha1 = md5(a1.encode('utf-8')).hexdigest()
a2 = request.method + ":" + auth.uri
ha2 = md5(a2.encode('utf-8')).hexdigest()
a3 = ha1 + ":" + auth.nonce + ":" + ha2
response = md5(a3.encode('utf-8')).hexdigest()
return response == auth.response