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
191
|
"""OAuth2 client utilities."""
import json
from datetime import datetime, timedelta
from urllib.parse import urljoin, urlparse
import requests
from flask import request, current_app as app
from pymonad.either import Left, Right, Either
from authlib.common.urls import url_decode
from authlib.jose import KeySet, JsonWebKey
from authlib.integrations.requests_client import OAuth2Session
from uploader import session
import uploader.monadic_requests as mrequests
SCOPE = ("profile group role resource register-client user masquerade "
"introspect migrate-data")
def authserver_uri():
"""Return URI to authorisation server."""
return app.config["AUTH_SERVER_URL"]
def oauth2_clientid():
"""Return the client id."""
return app.config["OAUTH2_CLIENT_ID"]
def oauth2_clientsecret():
"""Return the client secret."""
return app.config["OAUTH2_CLIENT_SECRET"]
def __fetch_auth_server_jwks__() -> KeySet:
"""Fetch the JWKs from the auth server."""
return KeySet([
JsonWebKey.import_key(key)
for key in requests.get(
urljoin(authserver_uri(), "auth/public-jwks")
).json()["jwks"]])
def __update_auth_server_jwks__(jwks) -> KeySet:
"""Update the JWKs from the servers if necessary."""
last_updated = jwks["last-updated"]
now = datetime.now().timestamp()
# Maybe the `two_hours` variable below can be made into a configuration
# variable and passed in to this function
two_hours = timedelta(hours=2).seconds
if bool(last_updated) and (now - last_updated) < two_hours:
return jwks["jwks"]
return session.set_auth_server_jwks(__fetch_auth_server_jwks__())
def auth_server_jwks() -> KeySet:
"""Fetch the auth-server JSON Web Keys information."""
_jwks = session.session_info().get("auth_server_jwks")
if bool(_jwks):
return __update_auth_server_jwks__({
"last-updated": _jwks["last-updated"],
"jwks": KeySet([
JsonWebKey.import_key(key) for key in _jwks.get(
"jwks", {"keys": []})["keys"]])
})
return __update_auth_server_jwks__({
"last-updated": (datetime.now() - timedelta(hours=3)).timestamp()
})
def oauth2_client():
"""Build the OAuth2 client for use fetching data."""
def __update_token__(token, refresh_token=None, access_token=None):# pylint: disable=[unused-argument]
"""Update the token when refreshed."""
session.set_user_token(token)
def __json_auth__(client, _method, uri, headers, body):
return (
uri,
{**headers, "Content-Type": "application/json"},
json.dumps({
**dict(url_decode(body)),
"client_id": client.client_id,
"client_secret": client.client_secret
}))
def __client__(token) -> OAuth2Session:
client = OAuth2Session(
oauth2_clientid(),
oauth2_clientsecret(),
scope=SCOPE,
token_endpoint=urljoin(authserver_uri(), "/auth/token"),
token_endpoint_auth_method="client_secret_post",
token=token,
update_token=__update_token__)
client.register_client_auth_method(
("client_secret_post", __json_auth__))
return client
return session.user_token().either(
lambda _notok: __client__(None),
__client__)
def user_logged_in():
"""Check whether the user has logged in."""
suser = session.session_info()["user"]
return suser["logged_in"] and suser["token"].is_right()
def authserver_authorise_uri():
"""Build up the authorisation URI."""
req_baseurl = urlparse(request.base_url, scheme=request.scheme)
host_uri = f"{req_baseurl.scheme}://{req_baseurl.netloc}/"
return urljoin(
authserver_uri(),
"auth/authorise?response_type=code"
f"&client_id={oauth2_clientid()}"
f"&redirect_uri={urljoin(host_uri, 'oauth2/code')}")
def __no_token__(_err) -> Left:
"""Handle situation where request is attempted with no token."""
resp = requests.models.Response()
resp._content = json.dumps({#pylint: disable=[protected-access]
"error": "AuthenticationError",
"error-description": ("You need to authenticate to access requested "
"information.")}).encode("utf-8")
resp.status_code = 400
return Left(resp)
def oauth2_get(url, **kwargs) -> Either:
"""Do a get request to the authentication/authorisation server."""
def __get__(_token) -> Either:
_uri = urljoin(authserver_uri(), url)
try:
resp = oauth2_client().get(
_uri,
**{
**kwargs,
"headers": {
**kwargs.get("headers", {}),
"Content-Type": "application/json"
}
})
if resp.status_code in mrequests.SUCCESS_CODES:
return Right(resp.json())
return Left(resp)
except Exception as exc:#pylint: disable=[broad-except]
app.logger.error("Error retriving data from auth server: (GET %s)",
_uri,
exc_info=True)
return Left(exc)
return session.user_token().either(__no_token__, __get__)
def oauth2_post(url, data=None, json=None, **kwargs):#pylint: disable=[redefined-outer-name]
"""Do a POST request to the authentication/authorisation server."""
def __post__(_token) -> Either:
_uri = urljoin(authserver_uri(), url)
_headers = ({
**kwargs.get("headers", {}),
"Content-Type": "application/json"
}
if bool(json) else kwargs.get("headers", {}))
try:
request_data = {
**(data or {}),
**(json or {}),
"client_id": oauth2_clientid(),
"client_secret": oauth2_clientsecret()
}
resp = oauth2_client().post(
_uri,
data=(request_data if bool(data) else None),
json=(request_data if bool(json) else None),
**{**kwargs, "headers": _headers})
if resp.status_code in mrequests.SUCCESS_CODES:
return Right(resp.json())
return Left(resp)
except Exception as exc:#pylint: disable=[broad-except]
app.logger.error("Error retriving data from auth server: (POST %s)",
_uri,
exc_info=True)
return Left(exc)
return session.user_token().either(__no_token__, __post__)
|
