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
from rest_framework import authentication
from rest_framework import exceptions
from httpsig import HeaderVerifier, utils
"""
Reusing failure exceptions serves several purposes:
1. Lack of useful information regarding the failure inhibits attackers
from learning about valid keyIDs or other forms of information leakage.
Using the same actual object for any failure makes preventing such
leakage through mistakenly-distinct error messages less likely.
2. In an API scenario, the object is created once and raised many times
rather than generated on every failure, which could lead to higher loads
or memory usage in high-volume attack scenarios.
"""
FAILED = exceptions.AuthenticationFailed('Invalid signature.')
class SignatureAuthentication(authentication.BaseAuthentication):
"""
DRF authentication class for HTTP Signature support.
You must subclass this class in your own project and implement the
`fetch_user_data(self, keyId, algorithm)` method, returning a tuple of
the User object and a bytes object containing the user's secret. Note
that key_id and algorithm are DIRTY as they are supplied by the client
and so must be verified in your subclass!
You may set the following class properties in your subclass to configure
authentication for your particular use case:
:param www_authenticate_realm: Default: "api"
:param required_headers: Default: ["(request-target)", "date"]
"""
www_authenticate_realm = "api"
required_headers = ["(request-target)", "date"]
def fetch_user_data(self, key_id, algorithm=None):
"""Retuns a tuple (User, secret) or (None, None)."""
raise NotImplementedError()
def authenticate_header(self, request):
"""
DRF sends this for unauthenticated responses if we're the primary
authenticator.
"""
h = " ".join(self.required_headers)
return 'Signature realm="%s",headers="%s"' % (
self.www_authenticate_realm, h)
def authenticate(self, request):
"""
Perform the actual authentication.
Note that the exception raised is always the same. This is so that we
don't leak information about in/valid keyIds and other such useful
things.
"""
auth_header = authentication.get_authorization_header(request)
if not auth_header or len(auth_header) == 0:
return None
method, fields = utils.parse_authorization_header(auth_header)
# Ignore foreign Authorization headers.
if method.lower() != 'signature':
return None
# Verify basic header structure.
if len(fields) == 0:
raise FAILED
# Ensure all required fields were included.
if len({"keyid", "algorithm", "signature"} - set(fields.keys())) > 0:
raise FAILED
# Fetch the secret associated with the keyid
user, secret = self.fetch_user_data(
fields["keyid"],
algorithm=fields["algorithm"]
)
if not (user and secret):
raise FAILED
# Gather all request headers and translate them as stated in the Django docs:
# https://docs.djangoproject.com/en/1.6/ref/request-response/#django.http.HttpRequest.META
headers = {}
for key in request.META.keys():
if key.startswith("HTTP_") or \
key in ("CONTENT_TYPE", "CONTENT_LENGTH"):
header = key[5:].lower().replace('_', '-')
headers[header] = request.META[key]
# Verify headers
hs = HeaderVerifier(
headers,
secret,
required_headers=self.required_headers,
method=request.method.lower(),
path=request.get_full_path()
)
# All of that just to get to this.
if not hs.verify():
raise FAILED
return user, fields["keyid"]