Extended introspection to also deal with non-jws access tokens.

Author: rohe

src/oidcendpoint/oauth2/introspection.py

@@ -2,6 +2,7 @@
 import logging
 
 from cryptojwt import JWT
+from cryptojwt.jws.jws import factory
 from oidcmsg import oauth2
 from oidcmsg.time_util import utc_time_sans_frac
 
@@ -32,6 +33,38 @@ def get_client_id_from_token(self, endpoint_context, token, request=None):
         sinfo = endpoint_context.sdb[token]
         return sinfo["authn_req"]["client_id"]
 
+    def do_jws(self, token, default_response):
+        _jwt = JWT(key_jar=self.endpoint_context.keyjar)
+
+        try:
+            _jwt_info = _jwt.unpack(token)
+        except Exception:
+            return {"response_args": default_response}
+
+        return _jwt_info
+
+    def do_access_token(self, token):
+        _info = self.endpoint_context.sdb[token]
+        if _info:  # Now what can be returned ?
+            _ret = {
+                "sub": _info["sub"],
+                "client_id": _info["client_id"],
+                "token_type": "bearer",
+                "iss": self.endpoint_context.issuer
+            }
+            _authn = _info.get("authn_event")
+            if _authn:
+                _ret["authn_info"] = _authn["authn_info"]
+                _ret["authn_time"] = _authn["authn_time"]
+
+            _scope = _info.get("scope")
+            if not _scope:
+                _ret["scope"] = " ".join(_info["authn_req"]["scope"])
+
+            return _ret
+        else:
+            return _info
+
     def process_request(self, request=None, **kwargs):
         """
 
@@ -43,31 +76,33 @@ def process_request(self, request=None, **kwargs):
         if "error" in _introspect_request:
             return _introspect_request
 
+        _token = _introspect_request["token"]
         _resp = self.response_cls(active=False)
 
-        _jwt = JWT(key_jar=self.endpoint_context.keyjar)
+        if factory(_token):
+            _info = self.do_jws(_token, _resp)
+            # expired ?
+            if "exp" in _info:
+                now = utc_time_sans_frac()
+                if _info["exp"] < now:
+                    return {"response_args": _resp}
+        else:
+            # A non-jws access token
+            _info = self.do_access_token(_token)
 
-        try:
-            _jwt_info = _jwt.unpack(_introspect_request["token"])
-        except Exception:
+        if not _info:
             return {"response_args": _resp}
 
-        # expired ?
-        if "exp" in _jwt_info:
-            now = utc_time_sans_frac()
-            if _jwt_info["exp"] < now:
-                return {"response_args": _resp}
-
         if "release" in self.kwargs:
             if "username" in self.kwargs["release"]:
                 try:
-                    _jwt_info["username"] = self.endpoint_context.userinfo.search(
-                        sub=_jwt_info["sub"]
+                    _info["username"] = self.endpoint_context.userinfo.search(
+                        sub=_info["sub"]
                     )
                 except KeyError:
-                    return {"response_args": _resp}
+                    pass
 
-        _resp.update(_jwt_info)
+        _resp.update(_info)
         _resp.weed()
         _resp["active"] = True
 

tests/test_31_introspection.py

@@ -6,6 +6,10 @@
 from cryptojwt import JWT
 from cryptojwt import as_unicode
 from cryptojwt.utils import as_bytes
+from oidcmsg.oidc import AccessTokenRequest
+
+from oidcendpoint.oidc.token_coop import TokenCoop
+
 from oidcendpoint.client_authn import ClientSecretPost
 from oidcendpoint.client_authn import UnknownOrNoAuthnMethod
 from oidcendpoint.client_authn import WrongAuthnMethod
@@ -65,6 +69,16 @@
     response_type="code id_token",
 )
 
+TOKEN_REQ = AccessTokenRequest(
+    client_id="client_1",
+    redirect_uri="https://example.com/cb",
+    state="STATE",
+    grant_type="authorization_code",
+    client_secret="hemligt",
+)
+
+TOKEN_REQ_DICT = TOKEN_REQ.to_dict()
+
 BASEDIR = os.path.abspath(os.path.dirname(__file__))
 
 
@@ -98,6 +112,18 @@ def create_endpoint(self):
                         "client_authn_method": {"client_secret_post": ClientSecretPost},
                     },
                 },
+                "token": {
+                    "path": "token",
+                    "class": TokenCoop,
+                    "kwargs": {
+                        "client_authn_method": [
+                            "client_secret_basic",
+                            "client_secret_post",
+                            "client_secret_jwt",
+                            "private_key_jwt",
+                        ]
+                    },
+                },
             },
             "authentication": {
                 "anon": {
@@ -127,6 +153,7 @@ def create_endpoint(self):
             endpoint_context.issuer,
         )
         self.introspection_endpoint = endpoint_context.endpoint["introspection"]
+        self.token_endpoint = endpoint_context.endpoint["token"]
 
     def _create_jwt(self, uid, lifetime=0, with_jti=False):
         _jwt = JWT(
@@ -237,4 +264,33 @@ def test_do_response_no_token(self):
             }
         )
         _resp = self.introspection_endpoint.process_request(_req)
-        assert "error" in _resp
+        assert "error" in _resp
+
+    def test_access_token(self):
+        _context = self.introspection_endpoint.endpoint_context
+
+        session_id = setup_session(
+            _context,
+            AUTH_REQ,
+            uid="user",
+            acr=INTERNETPROTOCOLPASSWORD,
+        )
+        _token_request = TOKEN_REQ_DICT.copy()
+        _token_request["code"] = _context.sdb[session_id]["code"]
+        _context.sdb.update(session_id, user="diana")
+
+        _req = self.token_endpoint.parse_request(_token_request)
+        _resp = self.token_endpoint.process_request(request=_req)
+
+        _req = self.introspection_endpoint.parse_request(
+            {
+                "token": _resp["response_args"]["access_token"],
+                "client_id": "client_1",
+                "client_secret": _context.cdb["client_1"]["client_secret"],
+            }
+        )
+        _resp = self.introspection_endpoint.process_request(_req)
+        _resp_args = _resp["response_args"]
+        assert "sub" in _resp_args
+        assert _resp_args["active"]
+        assert _resp_args["scope"] == "openid"