diff options
Diffstat (limited to 'gn_auth/auth/authentication/oauth2/endpoints')
4 files changed, 100 insertions, 0 deletions
diff --git a/gn_auth/auth/authentication/oauth2/endpoints/__init__.py b/gn_auth/auth/authentication/oauth2/endpoints/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/gn_auth/auth/authentication/oauth2/endpoints/__init__.py diff --git a/gn_auth/auth/authentication/oauth2/endpoints/introspection.py b/gn_auth/auth/authentication/oauth2/endpoints/introspection.py new file mode 100644 index 0000000..a567363 --- /dev/null +++ b/gn_auth/auth/authentication/oauth2/endpoints/introspection.py @@ -0,0 +1,48 @@ +"""Handle introspection of tokens.""" +import datetime +from urllib.parse import urlparse + +from flask import request as flask_request +from authlib.oauth2.rfc7662 import ( + IntrospectionEndpoint as _IntrospectionEndpoint) + +from gn3.auth.authentication.oauth2.models.oauth2token import OAuth2Token + +from .utilities import query_token as _query_token + +def get_token_user_sub(token: OAuth2Token) -> str:# pylint: disable=[unused-argument] + """ + Return the token's subject as defined in + https://datatracker.ietf.org/doc/html/rfc7519#section-4.1.2 + """ + ## For now a dummy return to prevent issues. + return "sub" + +class IntrospectionEndpoint(_IntrospectionEndpoint): + """Introspect token.""" + def query_token(self, token_string: str, token_type_hint: str): + """Query the token.""" + return _query_token(self, token_string, token_type_hint) + + def introspect_token(self, token: OAuth2Token) -> dict:# pylint: disable=[no-self-use] + """Return the introspection information.""" + url = urlparse(flask_request.url) + return { + "active": True, + "scope": token.get_scope(), + "client_id": token.client.client_id, + "username": token.user.name, + "token_type": token.token_type, + "exp": int(token.expires_at.timestamp()), + "iat": int(token.issued_at.timestamp()), + "nbf": int( + (token.issued_at - datetime.timedelta(seconds=120)).timestamp()), + # "sub": get_token_user_sub(token), + "aud": token.client.client_id, + "iss": f"{url.scheme}://{url.netloc}", + "jti": token.token_id + } + + def check_permission(self, token, client, request):# pylint: disable=[unused-argument, no-self-use] + """Check that the client has permission to introspect token.""" + return client.client_type == "internal" diff --git a/gn_auth/auth/authentication/oauth2/endpoints/revocation.py b/gn_auth/auth/authentication/oauth2/endpoints/revocation.py new file mode 100644 index 0000000..b8517b6 --- /dev/null +++ b/gn_auth/auth/authentication/oauth2/endpoints/revocation.py @@ -0,0 +1,22 @@ +"""Handle token revocation.""" + +from flask import current_app +from authlib.oauth2.rfc7009 import RevocationEndpoint as _RevocationEndpoint + +from gn3.auth import db +from gn3.auth.authentication.oauth2.models.oauth2token import ( + save_token, OAuth2Token, revoke_token) + +from .utilities import query_token as _query_token + +class RevocationEndpoint(_RevocationEndpoint): + """Revoke the tokens""" + ENDPOINT_NAME = "revoke" + def query_token(self, token_string: str, token_type_hint: str): + """Query the token.""" + return _query_token(self, token_string, token_type_hint) + + def revoke_token(self, token: OAuth2Token, request): + """Revoke token `token`.""" + with db.connection(current_app.config["AUTH_DB"]) as conn: + save_token(conn, revoke_token(token)) diff --git a/gn_auth/auth/authentication/oauth2/endpoints/utilities.py b/gn_auth/auth/authentication/oauth2/endpoints/utilities.py new file mode 100644 index 0000000..299f151 --- /dev/null +++ b/gn_auth/auth/authentication/oauth2/endpoints/utilities.py @@ -0,0 +1,30 @@ +"""endpoint utilities""" +from typing import Any, Optional + +from flask import current_app +from pymonad.maybe import Nothing + +from gn3.auth import db +from gn3.auth.authentication.oauth2.models.oauth2token import ( + OAuth2Token, token_by_access_token, token_by_refresh_token) + +def query_token(# pylint: disable=[unused-argument] + endpoint_object: Any, token_str: str, token_type_hint) -> Optional[ + OAuth2Token]: + """Retrieve the token from the database.""" + __identity__ = lambda val: val + token = Nothing + with db.connection(current_app.config["AUTH_DB"]) as conn: + if token_type_hint == "access_token": + token = token_by_access_token(conn, token_str) + if token_type_hint == "access_token": + token = token_by_refresh_token(conn, token_str) + + return token.maybe( + token_by_access_token(conn, token_str).maybe( + token_by_refresh_token(conn, token_str).maybe( + None, __identity__), + __identity__), + __identity__) + + return None |