Improved token validity checking.

Author: rohe

src/oidcendpoint/oauth2/introspection.py

@@ -11,6 +11,18 @@
 LOGGER = logging.getLogger(__name__)
 
 
+def before(t1, t2, range):
+    if t1 < (t2 - range):
+        return True
+    return False
+
+
+def after(t1, t2, range):
+    if t1 > (t2 + range):
+        return True
+    return False
+
+
 class Introspection(Endpoint):
     """Implements RFC 7662"""
 
@@ -21,6 +33,10 @@ class Introspection(Endpoint):
     endpoint_name = "introspection_endpoint"
     name = "introspection"
 
+    def __init__(self, **kwargs):
+        Endpoint.__init__(self, **kwargs)
+        self.offset = kwargs.get("offset", 0)
+
     def get_client_id_from_token(self, endpoint_context, token, request=None):
         """
         Will try to match tokens against information in the session DB.
@@ -33,18 +49,30 @@ 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):
+    def do_jws(self, token):
         _jwt = JWT(key_jar=self.endpoint_context.keyjar)
 
         try:
             _jwt_info = _jwt.unpack(token)
-        except Exception:
-            return {"response_args": default_response}
+        except Exception as err:
+            return None
 
         return _jwt_info
 
     def do_access_token(self, token):
-        _info = self.endpoint_context.sdb[token]
+        try:
+            _info = self.endpoint_context.sdb[token]
+        except KeyError:
+            return None
+
+        _revoked = _info.get("revoked", False)
+        if _revoked:
+            return None
+
+        _eat = _info.get("expires_at")
+        if _eat < utc_time_sans_frac():
+            return None
+
         if _info:  # Now what can be returned ?
             _ret = {
                 "sub": _info["sub"],
@@ -80,15 +108,26 @@ def process_request(self, request=None, **kwargs):
         _resp = self.response_cls(active=False)
 
         if factory(_token):
-            _info = self.do_jws(_token, _resp)
-            # expired ?
+            _info = self.do_jws(_token)
+            if _info is None:
+                return {"response_args": _resp}
+            _now = utc_time_sans_frac()
+
+            # Time checks
             if "exp" in _info:
-                now = utc_time_sans_frac()
-                if _info["exp"] < now:
+                if after(_now, _info["exp"], self.offset):
+                    return {"response_args": _resp}
+            if 'iat' in _info:
+                if after(_info["iat"], _now, self.offset):
+                    return {"response_args": _resp}
+            if 'nbf' in _info:
+                if before(_now, _info["nbf"], self.offset):
                     return {"response_args": _resp}
         else:
             # A non-jws access token
             _info = self.do_access_token(_token)
+            if _info is None:
+                return {"response_args": _resp}
 
         if not _info:
             return {"response_args": _resp}

tests/test_31_introspection.py

@@ -5,23 +5,24 @@
 import pytest
 from cryptojwt import JWT
 from cryptojwt import as_unicode
+from cryptojwt.key_jar import build_keyjar
 from cryptojwt.utils import as_bytes
+from oidcmsg.oauth2 import TokenIntrospectionRequest
 from oidcmsg.oidc import AccessTokenRequest
-
-from oidcendpoint.oidc.token_coop import TokenCoop
+from oidcmsg.oidc import AuthorizationRequest
+from oidcmsg.time_util import utc_time_sans_frac
 
 from oidcendpoint.client_authn import ClientSecretPost
 from oidcendpoint.client_authn import UnknownOrNoAuthnMethod
 from oidcendpoint.client_authn import WrongAuthnMethod
 from oidcendpoint.client_authn import verify_client
 from oidcendpoint.endpoint_context import EndpointContext
-from oidcendpoint.oauth2.introspection import Introspection
 from oidcendpoint.oauth2.authorization import Authorization
+from oidcendpoint.oauth2.introspection import Introspection
+from oidcendpoint.oidc.token_coop import TokenCoop
 from oidcendpoint.session import setup_session
 from oidcendpoint.user_authn.authn_context import INTERNETPROTOCOLPASSWORD
 from oidcendpoint.user_info import UserInfo
-from oidcmsg.oauth2 import TokenIntrospectionRequest
-from oidcmsg.oidc import AuthorizationRequest
 
 KEYDEFS = [
     {"type": "RSA", "key": "", "use": ["sig"]},
@@ -252,7 +253,7 @@ def test_do_response(self):
             "iss",
             "jti",
         }
-        assert _payload["active"] == True
+        assert _payload["active"] is True
 
     def test_do_response_no_token(self):
         _context = self.introspection_endpoint.endpoint_context
@@ -294,3 +295,89 @@ def test_access_token(self):
         assert "sub" in _resp_args
         assert _resp_args["active"]
         assert _resp_args["scope"] == "openid"
+
+    def test_jwt_unknown_key(self):
+        _keyjar = build_keyjar(KEYDEFS)
+
+        _jwt = JWT(
+            _keyjar,
+            iss=self.introspection_endpoint.endpoint_context.issuer,
+            lifetime=3600
+        )
+
+        _jwt.with_jti = True
+
+        _payload = {"sub": "subject_id"}
+        _token = _jwt.pack(_payload, aud="client_1")
+        _context = self.introspection_endpoint.endpoint_context
+
+        _req = self.introspection_endpoint.parse_request(
+            {
+                "token": _token,
+                "client_id": "client_1",
+                "client_secret": _context.cdb["client_1"]["client_secret"],
+            }
+        )
+
+        _req = self.introspection_endpoint.parse_request(_req)
+        _resp = self.introspection_endpoint.process_request(_req)
+        assert _resp["response_args"]["active"] is False
+
+    def test_expired_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)
+
+        _info = self.token_endpoint.endpoint_context.sdb[_resp["response_args"]["access_token"]]
+        _info['expires_at'] = utc_time_sans_frac() - 1000
+        self.token_endpoint.endpoint_context.sdb[_resp["response_args"]["access_token"]] = _info
+
+        _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)
+        assert _resp["response_args"]["active"] is False
+
+    def test_revoked_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)
+
+        self.token_endpoint.endpoint_context.sdb.revoke_session(
+            token=_resp["response_args"]["access_token"])
+
+        _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)
+        assert _resp["response_args"]["active"] is False