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
|
"""Common oauth2 client utilities."""
from urllib.parse import urljoin
from flask import session, current_app as app
from pymonad.maybe import Just, Maybe, Nothing
from pymonad.either import Left, Right, Either
from authlib.integrations.requests_client import OAuth2Session
SCOPE = "profile group role resource register-client"
def oauth2_client():
config = app.config
return OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
scope=SCOPE, token_endpoint_auth_method="client_secret_post",
token=session.get("oauth2_token"))
def get_endpoint(uri_path: str) -> Maybe:
token = session.get("oauth2_token", False)
if token and not bool(session.get("user_details", False)):
config = app.config
client = OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
token=token)
resp = client.get(
urljoin(config["GN_SERVER_URL"], uri_path))
resp_json = resp.json()
if resp_json.get("error") == "invalid_token":
flash(resp_json["error_description"], "alert-danger")
flash("You are now logged out.", "alert-info")
session.pop("oauth2_token", None)
return Nothing
return Just(resp_json)
return Nothing
def oauth2_get(uri_path: str) -> Either:
token = session.get("oauth2_token")
config = app.config
client = OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
token=token, scope=SCOPE)
resp = client.get(
urljoin(config["GN_SERVER_URL"], uri_path))
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
def oauth2_post(uri_path: str, data: dict) -> Either:
token = session.get("oauth2_token")
config = app.config
client = OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
token=token, scope=SCOPE)
resp = client.post(urljoin(config["GN_SERVER_URL"], uri_path), data=data)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
|
