Implement a sample IDP for CSRF and PKCE tests

Author: ArtyomVancyan

main.py

@@ -0,0 +1,66 @@
+import urllib.parse
+
+from flask import Flask, request, jsonify, url_for, redirect
+from oauthlib.oauth2 import Server
+
+from validator import MyRequestValidator
+
+app = Flask(__name__)
+
+oauth2_server = Server(MyRequestValidator())
+
+
+@app.route('/auth', methods=['GET', 'POST'])
+def auth():
+    if request.method == 'GET':
+        try:
+            # Validate the client request for authorization
+            uri = request.url
+            http_method = request.method
+            headers = request.headers
+            body = request.get_data()
+
+            scopes, credentials = oauth2_server.validate_authorization_request(uri, http_method, body, headers)
+            del credentials['request']
+            action = url_for('auth') + "?" + urllib.parse.urlencode({"scopes": ','.join(scopes), **credentials})
+
+            # Assuming the user is authenticated and named 'user1'
+            # You can integrate real user authentication here
+            return f"""
+            Do you authorize the app to access your data?
+            <form action="{action}" method="POST">
+                <button type="submit">Yes</button>
+            </form>
+            """
+        except:
+            return "Invalid authorization request", 400
+
+    elif request.method == 'POST':
+        uri = request.url
+        http_method = request.method
+        headers = request.headers
+        body = request.get_data()
+
+        headers, body, status = oauth2_server.create_authorization_response(uri, http_method, body, headers)
+
+        if status == 302:
+            location = headers.get('Location', '')
+            return redirect(location)
+
+        return jsonify(body), status
+
+
+@app.route('/token', methods=['POST'])
+def token():
+    uri = request.url
+    http_method = request.method
+    headers = request.headers
+    body = request.get_data()
+
+    headers, body, status = oauth2_server.create_token_response(uri, http_method, body, headers, {})
+
+    return body, status
+
+
+if __name__ == "__main__":
+    app.run()

validator.py

@@ -0,0 +1,45 @@
+from oauthlib.oauth2 import Client
+from oauthlib.oauth2 import RequestValidator
+
+
+class MyRequestValidator(RequestValidator):
+
+    def validate_client_id(self, client_id, request, *args, **kwargs):
+        return True
+
+    def validate_redirect_uri(self, client_id, redirect_uri, request, *args, **kwargs):
+        return True
+
+    def get_default_redirect_uri(self, client_id, request, *args, **kwargs):
+        return ""
+
+    def get_default_scopes(self, client_id, request, *args, **kwargs):
+        return []
+
+    def authenticate_client(self, request, *args, **kwargs):
+        request.client = Client(client_id="my_client", access_token="my_token")
+        return True
+
+    def confirm_redirect_uri(self, client_id, code, redirect_uri, client, request, *args, **kwargs):
+        return True
+
+    def validate_code(self, client_id, code, client, request, *args, **kwargs):
+        return True
+
+    def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs):
+        return True
+
+    def save_authorization_code(self, client_id, code, request, *args, **kwargs):
+        return True
+
+    def validate_response_type(self, client_id, response_type, client, request, *args, **kwargs):
+        return True
+
+    def validate_grant_type(self, client_id, grant_type, client, request, *args, **kwargs):
+        return True
+
+    def save_bearer_token(self, token, request, *args, **kwargs):
+        return True
+
+    def invalidate_authorization_code(self, client_id, code, request, *args, **kwargs):
+        return True