"""Common oauth2 client utilities."""
import json
import time
import random
import requests
from typing import Optional
from urllib.parse import urljoin
from datetime import datetime, timedelta
from flask import current_app as app
from pymonad.either import Left, Right, Either
from authlib.common.urls import url_decode
from authlib.jose.errors import BadSignatureError
from authlib.jose import KeySet, JsonWebKey, JsonWebToken
from authlib.integrations.requests_client import OAuth2Session
from gn2.wqflask.oauth2 import session
from gn2.wqflask.external_errors import ExternalRequestError
SCOPE = ("profile group role resource user masquerade introspect")
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 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 __make_token_validator__(keys: KeySet):
"""Make a token validator function."""
def __validator__(token: dict):
for key in keys.keys:
try:
# Fixes CVE-2016-10555. See
# https://docs.authlib.org/en/latest/jose/jwt.html
jwt = JsonWebToken(["RS256"])
jwt.decode(token["access_token"], key)
return Right(token)
except BadSignatureError:
pass
return Left("INVALID-TOKEN")
return __validator__
def auth_server_jwks() -> Optional[KeySet]:
"""Fetch the auth-server JSON Web Keys information."""
_jwks = session.session_info().get("auth_server_jwks")
if bool(_jwks):
return {
"last-updated": _jwks["last-updated"],
"jwks": KeySet([
JsonWebKey.import_key(key) for key in _jwks.get(
"auth_server_jwks", {}).get(
"jwks", {"keys": []})["keys"]])}
def __validate_token__(keys):
"""Validate that the token is really from the auth server."""
def __index__(_sess):
return _sess
return session.user_token().then(__make_token_validator__(keys)).then(
session.set_user_token).either(__index__, __index__)
def __update_auth_server_jwks__():
"""Updates the JWKs every 2 hours or so."""
jwks = auth_server_jwks()
if bool(jwks):
last_updated = jwks.get("last-updated")
now = datetime.now().timestamp()
if bool(last_updated) and (now - last_updated) < timedelta(hours=2).seconds:
return __validate_token__(jwks["jwks"])
jwksuri = urljoin(authserver_uri(), "auth/public-jwks")
jwks = KeySet([
JsonWebKey.import_key(key)
for key in requests.get(jwksuri).json()["jwks"]])
return __validate_token__(jwks)
def is_token_expired(token):
"""Check whether the token has expired."""
__update_auth_server_jwks__()
jwks = auth_server_jwks()
if bool(jwks):
for jwk in jwks["jwks"].keys:
try:
jwt = JsonWebToken(["RS256"]).decode(
token["access_token"], key=jwk)
return datetime.now().timestamp() > jwt["exp"]
except BadSignatureError as _bse:
pass
return False
def oauth2_client():
def __update_token__(token, refresh_token=None, access_token=None):
"""Update the token when refreshed."""
session.set_user_token(token)
return token
def __delay__():
"""Do a tiny delay."""
time.sleep(random.choice(tuple(i/1000.0 for i in range(0,100))))
def __refresh_token__(token):
"""Synchronise token refresh."""
if is_token_expired(token):
__delay__()
if session.is_token_refreshing():
while session.is_token_refreshing():
__delay__()
_token = session.user_token().either(None, lambda _tok: _tok)
return _token
session.toggle_token_refreshing()
_client = __client__(token)
_client.get(urljoin(authserver_uri(), "auth/user/"))
session.toggle_token_refreshing()
return _client.token
return token
def __json_auth__(client, method, uri, headers, body):
return (
uri,
{**headers, "Content-Type": "application/json"},
json.dumps({
**dict(url_decode(body)),
"client_id": oauth2_clientid(),
"client_secret": oauth2_clientsecret()
}))
def __client__(token) -> OAuth2Session:
client = OAuth2Session(
oauth2_clientid(),
oauth2_clientsecret(),
scope=token["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
__update_auth_server_jwks__()
return session.user_token().then(__refresh_token__).either(
lambda _notok: __client__(None),
lambda token: __client__(token))
def __no_token__(_err) -> Left:
"""Handle situation where request is attempted with no token."""
resp = requests.models.Response()
resp._content = json.dumps({
"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(
uri_path: str,
data: dict = {},
jsonify_p: bool = False,
headers: dict = {"Content-Type": "application/json"},
**kwargs
) -> Either:
def __get__(token) -> Either:
resp = oauth2_client().get(
urljoin(authserver_uri(), uri_path),
data=data,
headers=headers,
**kwargs)
if resp.status_code == 200:
if jsonify_p:
return Right(resp)
return Right(resp.json())
return Left(resp)
return session.user_token().either(__no_token__, __get__)
def oauth2_post(
uri_path: str,
data: Optional[dict] = None,
json: Optional[dict] = None,
headers: dict = {"Content-Type": "application/json"},
**kwargs
) -> Either:
def __post__(token) -> Either:
resp = oauth2_client().post(
urljoin(authserver_uri(), uri_path),
data=data,
json=json,
headers=headers,
**kwargs)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
return session.user_token().either(__no_token__, __post__)
def no_token_get(
uri_path: str,
headers: dict = {"Content-Type": "application/json"},
**kwargs
) -> Either:
uri = urljoin(authserver_uri(), uri_path)
try:
resp = requests.get(uri, headers=headers, **kwargs)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
except requests.exceptions.RequestException as exc:
raise ExternalRequestError(uri, exc) from exc
def no_token_post(uri_path: str, **kwargs) -> Either:
data = kwargs.get("data", {})
the_json = kwargs.get("json", {})
request_data = {
**data,
**the_json,
"client_id": oauth2_clientid(),
"client_secret": oauth2_clientsecret()
}
new_kwargs = {
**{
key: value for key, value in kwargs.items()
if key not in ("data", "json")
},
("data" if bool(data) else "json"): request_data
}
try:
resp = requests.post(urljoin(authserver_uri(), uri_path),
**new_kwargs)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
except requests.exceptions.RequestException as exc:
raise ExternalRequestError(uri_path, exc) from exc
def post(uri_path: str, **kwargs) -> Either:
"""
Generic function to do POST requests, that checks whether or not the user is
logged in and selects the appropriate function/method to run.
"""
if user_logged_in():
return oauth2_post(uri_path, **kwargs)
return no_token_post(uri_path, **kwargs)
def get(uri_path: str, **kwargs) -> Either:
"""
Generic function to do GET requests, that checks whether or not the user is
logged in and selects the appropriate function/method to run.
"""
if user_logged_in():
return oauth2_get(uri_path, **kwargs)
return no_token_get(uri_path, **kwargs)
