Merge pull request #23 from sbrunner/allows-pkce

Author: lilioid

README.md

@@ -82,4 +82,6 @@ The list of [OpenID specifications](https://openid.net/developers/specs/) can be
 
 - ✔️ Full [OpenID Connect Back-Channel Logout 1.0](https://openid.net/specs/openid-connect-backchannel-1_0.html)
 
+- ✔️ Full [Proof Key for Code Exchange (PKCE)](https://www.rfc-editor.org/rfc/rfc7636)
+
 - ✔️ Full [OAuth 2.0 Token Introspection](https://www.rfc-editor.org/rfc/rfc7662)

docs/usage.rst

@@ -103,3 +103,39 @@ It should be interpreted as pseudocode without any specific web or application f
         token_response = client.authorization_code_flow.handle_authentication_result(current_url)
         # token_response now contains access and id tokens
         ...
+
+Proof Key for Code Exchange (PKCE)
+----------------------------------
+
+The Proof Key for Code Exchange (PKCE) is a security extension to OAuth2.0 that is used to prevent authorization code injection attacks.
+
+The PKCE extension is used in the authorization code flow.
+
+Example usage of PKCE::
+
+    from simple_openid_connect import pkce
+
+    client = OpenidClient.from_issuer_url(
+        url="https://provider.example.com/openid",
+        authentication_redirect_uri="https://myapp.example.com/login-callback",
+        client_id="client-id",
+    )
+    code_verifier, code_challenge = pkce.generate_pkce_pair()
+
+    def on_login():
+        # this method should be called when the user wants to log in
+        # it returns an HTTP redirect to the Openid provider
+        return HttpRedirect(to=client.authorization_code_flow.start_authentication(
+            code_challenge=code_challenge,
+            code_challenge_method="S256"
+        ))
+
+    def on_login_callback(current_url):
+        token_response = client.authorization_code_flow.handle_authentication_result(
+            current_url,
+            code_verifier=code_verifier,
+            code_challenge=code_challenge,
+            code_challenge_method="S256",
+        )
+        # token_response now contains access and id tokens
+        ...

src/simple_openid_connect/data.py

@@ -440,6 +440,12 @@ class AuthenticationRequest(OpenidBaseModel):
     acr_values: Optional[List[str]] = None
     "OPTIONAL. Requested Authentication Context Class Reference values Space-separated string that specifies the acr values that the Authorization Server is being requested to use for processing this Authentication Request, with the values appearing in order of preference The Authentication Context Class satisfied by the authentication performed is returned as the acr Claim Value, as specified in Section 2 The acr Claim is requested as a Voluntary Claim by this parameter."
 
+    code_challenge: Optional[str] = None
+    "OPTIONAL. Code Challenge. This parameter is intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], to be used with code_challenge_method."
+
+    code_challenge_method: Optional[str] = None
+    "OPTIONAL. Code Challenge Method. This parameter is intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], to be used with code_challenge."
+
 
 class AuthenticationSuccessResponse(OpenidBaseModel):
     """
@@ -575,6 +581,15 @@ class TokenRequest(OpenidBaseModel):
     scope: Optional[str] = None
     "REQUIRED, if grant type is 'password'. The scope requested by the application"
 
+    code_verifier: Optional[str] = None
+    "OPTIONAL. Code Verifier. This parameter is intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], to be used with code_challenge and code_challenge_method."
+
+    code_challenge: Optional[str] = None
+    "OPTIONAL. Code Challenge. This parameter is intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], to be used with code_verifier, code_challenge_method."
+
+    code_challenge_method: Optional[str] = None
+    "OPTIONAL. Code Challenge Method. This parameter is intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], to be used with code_verifier, code_challenge."
+
     @model_validator(mode="before")
     @classmethod
     def _validate_required_based_on_grant_type(

src/simple_openid_connect/flows/authorization_code_flow/__init__.py

@@ -7,7 +7,7 @@
 """
 import copy
 import logging
-from typing import Literal, Union
+from typing import Literal, Optional, Union
 
 import requests
 from furl import furl
@@ -27,7 +27,12 @@
 
 
 def start_authentication(
-    authorization_endpoint: str, scope: str, client_id: str, redirect_uri: str
+    authorization_endpoint: str,
+    scope: str,
+    client_id: str,
+    redirect_uri: str,
+    code_challenge: Optional[str] = None,
+    code_challenge_method: Optional[str] = None,
 ) -> str:
     """
     Start the authentication process by constructing an appropriate :class:`AuthenticationRequest`, serializing it and
@@ -40,6 +45,8 @@ def start_authentication(
         client_id=client_id,
         redirect_uri=redirect_uri,
         response_type="code",
+        code_challenge=code_challenge,
+        code_challenge_method=code_challenge_method,
     )
     return request.encode_url(authorization_endpoint)
 
@@ -49,6 +56,9 @@ def handle_authentication_result(
     token_endpoint: str,
     client_authentication: ClientAuthenticationMethod,
     redirect_uri: Union[Literal["auto"], str] = "auto",
+    code_verifier: Optional[str] = None,
+    code_challenge: Optional[str] = None,
+    code_challenge_method: Optional[str] = None,
 ) -> Union[TokenSuccessResponse, TokenErrorResponse]:
     """
     Handle an authentication result that is communicated to the RP in form of the user agents current url after having started an authentication process via :func:`start_authentication`.
@@ -87,6 +97,9 @@ def handle_authentication_result(
         authentication_response=auth_response_msg,
         redirect_uri=redirect_uri,
         client_authentication=client_authentication,
+        code_verifier=code_verifier,
+        code_challenge=code_challenge,
+        code_challenge_method=code_challenge_method,
     )
 
 
@@ -95,6 +108,9 @@ def exchange_code_for_tokens(
     authentication_response: AuthenticationSuccessResponse,
     redirect_uri: str,
     client_authentication: ClientAuthenticationMethod,
+    code_verifier: Optional[str] = None,
+    code_challenge: Optional[str] = None,
+    code_challenge_method: Optional[str] = None,
 ) -> Union[TokenSuccessResponse, TokenErrorResponse]:
     """
     Exchange a received code for access, refresh and id tokens.
@@ -116,6 +132,9 @@ def exchange_code_for_tokens(
         redirect_uri=redirect_uri,
         client_id=client_authentication.client_id,
         grant_type="authorization_code",
+        code_verifier=code_verifier,
+        code_challenge=code_challenge,
+        code_challenge_method=code_challenge_method,
     )
     response = requests.post(
         token_endpoint,

src/simple_openid_connect/flows/authorization_code_flow/client.py

@@ -27,11 +27,18 @@ class AuthorizationCodeFlowClient:
     def __init__(self, base_client: "OpenidClient"):
         self._base_client = base_client
 
-    def start_authentication(self) -> str:
+    def start_authentication(
+        self,
+        code_challenge: Optional[str] = None,
+        code_challenge_method: Optional[str] = None,
+    ) -> str:
         """
         Start the authentication process by constructing an appropriate :class:`AuthenticationRequest`, serializing it and
         returning a which the end user now needs to visit.
 
+        :param code_challenge: The code challenge intended for use with Proof Key for Code Exchange (PKCE) [RFC7636].
+        :param code_challenge_method: The code challenge method intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], typically "S256" or "plain".
+
         :raises ImpossibleOperationError: If the client has no redirect_uri configured and therefore cannot perform this operation.
 
         :returns: A URL to which the user agent should be redirected
@@ -48,12 +55,17 @@ def start_authentication(self) -> str:
             self._base_client.scope,
             self._base_client.client_auth.client_id,
             redirect_uri.tostr(),
+            code_challenge=code_challenge,
+            code_challenge_method=code_challenge_method,
         )
 
     def handle_authentication_result(
         self,
         current_url: str,
         additional_redirect_args: Optional[Mapping[str, str]] = None,
+        code_verifier: Optional[str] = None,
+        code_challenge: Optional[str] = None,
+        code_challenge_method: Optional[str] = None,
     ) -> Union[TokenSuccessResponse, TokenErrorResponse]:
         """
         Handle an authentication result that is communicated to the RP in form of the user agents current url after having started an authentication process via :func:`start_authentication`.
@@ -62,6 +74,9 @@ def handle_authentication_result(
             The authentication result should be encoded into this url by the authorization server.
         :param additional_redirect_args: Additional URL parameters that were added to the redirect uri.
             They are probably still present in `current_url` but since they could be of any shape, no attempt is made here to automatically reconstruct them.
+        :param code_verifier: The code verifier intended for use with Proof Key for Code Exchange (PKCE) [RFC7636].
+        :param code_challenge: The code challenge intended for use with Proof Key for Code Exchange (PKCE) [RFC7636].
+        :param code_challenge_method: The code challenge method intended for use with Proof Key for Code Exchange (PKCE) [RFC7636], typically "S256" or "plain".
 
         :raises AuthenticationFailedError: If the current url indicates an authentication failure that prevents an access token from being retrieved.
         :raises UnsupportedByProviderError: If the provider only supports implicit flow and has no token endpoint.
@@ -88,6 +103,9 @@ def handle_authentication_result(
             token_endpoint=self._base_client.provider_config.token_endpoint,
             client_authentication=self._base_client.client_auth,
             redirect_uri=redirect_uri.tostr(),
+            code_verifier=code_verifier,
+            code_challenge=code_challenge,
+            code_challenge_method=code_challenge_method,
         )
 
     def exchange_code_for_tokens(

src/simple_openid_connect/pkce.py

@@ -0,0 +1,102 @@
+"""
+Simple module to generate PKCE code verifier and code challenge.
+
+Original code from @MartinThoma in this file:
+https://github.com/RomeoDespres/pkce/blob/master/pkce/__init__.py
+
+
+Examples
+--------
+>>> from simple_openid_connect import pkce
+>>> code_verifier, code_challenge = pkce.generate_pkce_pair()
+
+>>> from simple_openid_connect import pkce
+>>> code_verifier = pkce.generate_code_verifier(length=128)
+>>> code_challenge = pkce.get_code_challenge(code_verifier)
+"""
+
+import base64
+import hashlib
+import secrets
+from typing import Tuple
+
+
+def generate_code_verifier(length: int = 128) -> str:
+    """Return a random PKCE-compliant code verifier.
+
+    Parameters
+    ----------
+    length : int
+        Code verifier length. Must verify `43 <= length <= 128`.
+
+    Returns
+    -------
+    code_verifier : str
+        Code verifier.
+
+    Raises
+    ------
+    ValueError
+        When `43 <= length <= 128` is not verified.
+    """
+    if not 43 <= length <= 128:
+        msg = "Parameter `length` must verify `43 <= length <= 128`."
+        raise ValueError(msg)
+    code_verifier = secrets.token_urlsafe(96)[:length]
+    return code_verifier
+
+
+def generate_pkce_pair(code_verifier_length: int = 128) -> Tuple[str, str]:
+    """Return random PKCE-compliant code verifier and code challenge.
+
+    Parameters
+    ----------
+    code_verifier_length : int
+        Code verifier length. Must verify
+        `43 <= code_verifier_length <= 128`.
+
+    Returns
+    -------
+    code_verifier : str
+    code_challenge : str
+
+    Raises
+    ------
+    ValueError
+        When `43 <= code_verifier_length <= 128` is not verified.
+    """
+    if not 43 <= code_verifier_length <= 128:
+        msg = "Parameter `code_verifier_length` must verify "
+        msg += "`43 <= code_verifier_length <= 128`."
+        raise ValueError(msg)
+    code_verifier = generate_code_verifier(code_verifier_length)
+    code_challenge = get_code_challenge(code_verifier)
+    return code_verifier, code_challenge
+
+
+def get_code_challenge(code_verifier: str) -> str:
+    """Return the PKCE-compliant code challenge for a given verifier.
+
+    Parameters
+    ----------
+    code_verifier : str
+        Code verifier. Must verify `43 <= len(code_verifier) <= 128`.
+
+    Returns
+    -------
+    code_challenge : str
+        Code challenge that corresponds to the input code verifier.
+
+    Raises
+    ------
+    ValueError
+        When `43 <= len(code_verifier) <= 128` is not verified.
+    """
+    if not 43 <= len(code_verifier) <= 128:
+        msg = "Parameter `code_verifier` must verify "
+        msg += "`43 <= len(code_verifier) <= 128`."
+        raise ValueError(msg)
+    hashed = hashlib.sha256(code_verifier.encode("ascii")).digest()
+    encoded = base64.urlsafe_b64encode(hashed)
+    code_challenge = encoded.decode("ascii")[:-1]
+    return code_challenge