Implement verification for the PKCE flow

Author: ArtyomVancyan

tests/idp/validator.py

@@ -1,8 +1,13 @@
+import base64
+import hashlib
+
 from oauthlib.oauth2 import Client
 from oauthlib.oauth2 import RequestValidator
 
 
 class TestValidator(RequestValidator):
+    pkce_codes = {}
+
     def validate_client_id(self, client_id, request, *args, **kwargs):
         return True
 
@@ -23,12 +28,30 @@ def confirm_redirect_uri(self, client_id, code, redirect_uri, client, request, *
         return True
 
     def validate_code(self, client_id, code, client, request, *args, **kwargs):
-        return True
+        stored_challenge = self.pkce_codes.get(code)
+        if not stored_challenge:
+            return False
+
+        code_verifier = request.code_verifier
+        code_challenge = stored_challenge.get("code_challenge")
+        code_challenge_method = stored_challenge.get("code_challenge_method")
+
+        computed_challenge = code_verifier
+        if code_challenge_method == "S256":
+            sha256 = hashlib.sha256()
+            sha256.update(code_verifier.encode("utf-8"))
+            computed_challenge = base64.urlsafe_b64encode(sha256.digest()).decode("utf-8").replace("=", "")
+
+        return computed_challenge == code_challenge
 
     def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs):
         return True
 
     def save_authorization_code(self, client_id, code, request, *args, **kwargs):
+        self.pkce_codes[code.get("code")] = dict(
+            code_challenge=request.code_challenge,
+            code_challenge_method=request.code_challenge_method,
+        )
         return True
 
     def validate_response_type(self, client_id, response_type, client, request, *args, **kwargs):