Merge pull request #115 from lionick/add_handle_errors

Handle errors related to refresh token and client redirect URI

src/idpyoidc/server/oauth2/authorization.py

@@ -125,11 +125,11 @@ def verify_uri(
     client_info = context.cdb.get(client_id)
     if not client_info:
         logger.error("No client info found")
-        raise KeyError("No client info found")
+        raise UnknownClient("No client info found")
 
     req_redirect_uri_quoted = request.get(uri_type)
     if req_redirect_uri_quoted is None:
-        raise ValueError(f"Wrong uri_type: {uri_type}")
+        raise URIError(f"Wrong uri_type: {uri_type}")
 
     req_redirect_uri = unquote(req_redirect_uri_quoted)
     req_redirect_uri_obj = urlparse(req_redirect_uri)
@@ -558,7 +558,7 @@ def _post_parse_request(self, request, client_id, context, **kwargs):
         # Get a verified redirect URI
         try:
             redirect_uri = get_uri(context, request, "redirect_uri", self.endpoint_type)
-        except (RedirectURIError, ParameterError) as err:
+        except (RedirectURIError, ParameterError, URIError, UnknownClient) as err:
             return self.authentication_error_response(
                 request,
                 error="invalid_request",

src/idpyoidc/server/oidc/token_helper/refresh_token.py

@@ -7,6 +7,8 @@
 from cryptojwt.jws.exception import NoSuitableSigningKeys
 from cryptojwt.jwt import utc_time_sans_frac
 
+from ...exception import InvalidBranchID
+from idpyoidc.exception import MissingRequiredAttribute
 from idpyoidc.message import Message
 from idpyoidc.message.oidc import RefreshAccessTokenRequest
 from idpyoidc.server.oauth2.token_helper import TokenEndpointHelper
@@ -140,16 +142,19 @@ def post_parse_request(
         request = RefreshAccessTokenRequest(**request.to_dict())
         _context = self.endpoint.upstream_get("context")
 
-        request.verify(
-            keyjar=self.endpoint.upstream_get("attribute", "keyjar"), opponent_id=client_id
-        )
+        try:
+            request.verify(
+                keyjar=self.endpoint.upstream_get("attribute", "keyjar"), opponent_id=client_id
+            )
+        except MissingRequiredAttribute as e:
+            return self.error_cls(error="invalid_grant", error_description=str(e))
 
         _mngr = _context.session_manager
         try:
             _session_info = _mngr.get_session_info_by_token(
                 request["refresh_token"], handler_key="refresh_token", grant=True
             )
-        except (KeyError, UnknownToken, BadSyntax):
+        except (KeyError, UnknownToken, BadSyntax, InvalidBranchID):
             logger.error("Refresh token invalid")
             return self.error_cls(error="invalid_grant", error_description="Invalid refresh token")
 

src/idpyoidc/server/session/manager.py

@@ -13,7 +13,9 @@
 from idpyoidc.server.exception import ConfigurationError
 from idpyoidc.server.session.grant_manager import GrantManager
 from idpyoidc.util import rndstr
+
 from .database import Database
+from ..exception import InvalidBranchID
 from .grant import Grant
 from .grant import SessionToken
 from .info import ClientSessionInfo
@@ -480,8 +482,13 @@ def get_session_info(
         :param authorization_request: Whether the authorization_request should part of the response
         :return: A dictionary with session information
         """
-        res = self.branch_info(session_id)
-
+        try:
+            res = self.branch_info(session_id)
+        except InvalidBranchID as e:
+            # Log the exception if needed
+            logging.error(f"InvalidBranchID error: {str(e)}")
+            raise
+
         if authentication_event:
             res["authentication_event"] = res["grant"].authentication_event
 

tests/test_server_24_oauth2_authorization_endpoint.py

@@ -348,7 +348,7 @@ def test_verify_uri_noregistered(self):
         _context = self.endpoint.upstream_get("context")
         request = {"redirect_uri": "https://rp.example.com/cb"}
 
-        with pytest.raises(KeyError):
+        with pytest.raises(UnknownClient):
             verify_uri(_context, request, "redirect_uri", "client_id")
 
     def test_verify_uri_unregistered(self):
@@ -553,7 +553,7 @@ def test_verify_uri_wrong_uri_type(self):
         _context.cdb["client_id"] = {"redirect_uris": [("https://rp.example.com/cb", {})]}
 
         request = {"redirect_uri": "https://rp.example.com/cb?foo=bob"}
-        with pytest.raises(ValueError):
+        with pytest.raises(URIError):
             verify_uri(_context, request, "post_logout_redirect_uri", "client_id")
 
     def test_verify_uri_none_registered(self):

tests/test_server_24_oidc_authorization_endpoint.py

@@ -447,7 +447,7 @@ def test_verify_uri_noregistered(self):
         _ec = self.endpoint.upstream_get("context")
         request = {"redirect_uri": "https://rp.example.com/cb"}
 
-        with pytest.raises(KeyError):
+        with pytest.raises(UnknownClient):
             verify_uri(_ec, request, "redirect_uri", "client_id")
 
     def test_verify_uri_unregistered(self):