aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authentication/oauth2/endpoints
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/auth/authentication/oauth2/endpoints')
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/__init__.py0
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/introspection.py48
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/revocation.py21
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/utilities.py30
4 files changed, 99 insertions, 0 deletions
diff --git a/gn3/auth/authentication/oauth2/endpoints/__init__.py b/gn3/auth/authentication/oauth2/endpoints/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/gn3/auth/authentication/oauth2/endpoints/__init__.py
diff --git a/gn3/auth/authentication/oauth2/endpoints/introspection.py b/gn3/auth/authentication/oauth2/endpoints/introspection.py
new file mode 100644
index 0000000..a567363
--- /dev/null
+++ b/gn3/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/gn3/auth/authentication/oauth2/endpoints/revocation.py b/gn3/auth/authentication/oauth2/endpoints/revocation.py
new file mode 100644
index 0000000..0693c2d
--- /dev/null
+++ b/gn3/auth/authentication/oauth2/endpoints/revocation.py
@@ -0,0 +1,21 @@
+"""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"""
+ 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/gn3/auth/authentication/oauth2/endpoints/utilities.py b/gn3/auth/authentication/oauth2/endpoints/utilities.py
new file mode 100644
index 0000000..299f151
--- /dev/null
+++ b/gn3/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