Token introspection

tests/conftest.py

@@ -91,7 +91,7 @@ def app(slapd_server):
             "JWT": {
                 "KEY": "secret-key",
                 "ALG": "HS256",
-                "ISS": "http://mydomain.tld",
+                "ISS": "https://mydomain.tld",
                 "EXP": 3600,
                 "MAPPING": {
                     "SUB": "uid",
@@ -155,6 +155,22 @@ def user(app, slapd_connection):
     return u
 
 
+@pytest.fixture
+def token(slapd_connection, client, user):
+    t = Token(
+        oauthAccessToken=gen_salt(48),
+        oauthClientID=client.oauthClientID,
+        oauthSubject=user.dn,
+        oauthTokenType=None,
+        oauthRefreshToken=gen_salt(48),
+        oauthScope="openid profile",
+        oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%M%SZ"),
+        oauthTokenLifetime=str(3600),
+    )
+    t.save(slapd_connection)
+    return t
+
+
 @pytest.fixture
 def logged_user(user, testclient):
     with testclient.session_transaction() as sess:

tests/test_authorization_code_flow.py

@@ -141,7 +141,6 @@ def test_refresh_token(testclient, slapd_connection, logged_user, client):
     token = Token.get(access_token, conn=slapd_connection)
     assert token is not None
 
-    print("------------------------------------")
     res = testclient.post(
         "/oauth/token",
         params=dict(

web/models.py

@@ -165,6 +165,10 @@ class AuthorizationCode(LDAPObjectHelper, AuthorizationCodeMixin):
         )
         return (auth_time - datetime.datetime(1970, 1, 1)).total_seconds()
 
+    @property
+    def code_challenge(self):
+        return self.oauthCodeChallenge
+
 
 class Token(LDAPObjectHelper, TokenMixin):
     objectClass = ["oauthToken"]
@@ -180,6 +184,10 @@ class Token(LDAPObjectHelper, TokenMixin):
     def get_expires_in(self):
         return int(self.oauthTokenLifetime)
 
+    def get_issued_at(self):
+        issue_date = datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
+        return (issue_date - datetime.datetime(1970, 1, 1)).total_seconds()
+
     def get_expires_at(self):
         issue_date = datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
         issue_timestamp = (issue_date - datetime.datetime(1970, 1, 1)).total_seconds()
@@ -193,3 +201,7 @@ class Token(LDAPObjectHelper, TokenMixin):
             + datetime.timedelta(seconds=int(self.oauthTokenLifetime))
             >= datetime.datetime.now()
         )
+
+    @property
+    def revoked(self):
+        return False

web/oauth.py

@@ -3,7 +3,7 @@ from flask import Blueprint, request, session, redirect
 from flask import render_template, jsonify, flash
 from flask_babel import gettext
 from .models import User, Client
-from .oauth2utils import authorization
+from .oauth2utils import authorization, IntrospectionEndpoint
 from .forms import LoginForm
 from .flaskutils import current_user
 
@@ -56,3 +56,8 @@ def authorize():
 @bp.route("/token", methods=["POST"])
 def issue_token():
     return authorization.create_token_response()
+
+
+@bp.route("/introspect", methods=["POST"])
+def introspect_token():
+    return authorization.create_endpoint_response(IntrospectionEndpoint.ENDPOINT_NAME)

web/oauth2utils.py

@@ -8,6 +8,8 @@ from authlib.oauth2.rfc6749.grants import (
     ClientCredentialsGrant,
 )
 from authlib.oauth2.rfc6750 import BearerTokenValidator as _BearerTokenValidator
+from authlib.oauth2.rfc7636 import CodeChallenge
+from authlib.oauth2.rfc7662 import IntrospectionEndpoint as _IntrospectionEndpoint
 from authlib.oidc.core.grants import (
     OpenIDCode as _OpenIDCode,
     OpenIDImplicitGrant as _OpenIDImplicitGrant,
@@ -81,6 +83,8 @@ def save_authorization_code(code, request):
         oauthNonce=nonce,
         oauthAuthorizationDate=now.strftime("%Y%m%d%H%M%SZ"),
         oauthAuthorizationLifetime=str(84000),
+        oauthCodeChallenge=request.data.get("code_challenge"),
+        oauthCodeChallengeMethod=request.data.get("code_challenge_method"),
     )
     code.save()
     return code.oauthCode
@@ -178,9 +182,8 @@ def save_token(token, request):
         oauthTokenLifetime=str(token["expires_in"]),
         oauthScope=token["scope"],
         oauthClientID=request.client.oauthClientID,
+        oauthRefreshToken=token.get("refresh_token"),
     )
-    if "refresh_token" in token:
-        t.oauthRefreshToken = token["refresh_token"]
     t.save()
 
 
@@ -195,6 +198,38 @@ class BearerTokenValidator(_BearerTokenValidator):
         return False
 
 
+class IntrospectionEndpoint(_IntrospectionEndpoint):
+    def query_token(self, token, token_type_hint, client):
+        if token_type_hint == "access_token":
+            tok = Token.filter(oauthAccessToken=token)
+        elif token_type_hint == "refresh_token":
+            tok = Token.filter(oauthRefreshToken=token)
+        else:
+            tok = Token.filter(oauthAccessToken=token)
+            if not tok:
+                tok = Token.filter(oauthRefreshToken=token)
+        if tok:
+            tok = tok[0]
+            if tok.oauthClientID == client.oauthClientID:
+                return tok
+            # if has_introspect_permission(client):
+            #    return tok
+
+    def introspect_token(self, token):
+        return {
+            "active": True,
+            "client_id": token.oauthClientID,
+            "token_type": token.oauthTokenType,
+            "username": User.get(token.oauthSubject).name,
+            "scope": token.get_scope(),
+            "sub": token.oauthSubject,
+            "aud": token.oauthClientID,
+            "iss": current_app.config["JWT"]["ISS"],
+            "exp": token.get_expires_at(),
+            "iat": token.get_issued_at(),
+        }
+
+
 authorization = AuthorizationServer()
 require_oauth = ResourceProtector()
 
@@ -208,9 +243,12 @@ def config_oauth(app):
     authorization.register_grant(ClientCredentialsGrant)
 
     authorization.register_grant(
-        AuthorizationCodeGrant, [OpenIDCode(require_nonce=True)]
+        AuthorizationCodeGrant,
+        [OpenIDCode(require_nonce=True), CodeChallenge(required=False)],
     )
     authorization.register_grant(OpenIDImplicitGrant)
     authorization.register_grant(OpenIDHybridGrant)
 
     require_oauth.register_token_validator(BearerTokenValidator())
+
+    authorization.register_endpoint(IntrospectionEndpoint)