oauth_local_webhook_server.py

from http.server import BaseHTTPRequestHandler, HTTPServer
import urllib.parse
import json
import requests 

class RequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed_url = urllib.parse.urlparse(self.path)
        query_params = urllib.parse.parse_qs(parsed_url.query)

        if parsed_url.path == '/callback':  
            if 'code' in query_params and 'state' in query_params:
                code = query_params['code'][0]
                state = query_params['state'][0]

                
                if state != '<expected_state>': 
                    self.send_response(400) 
                    self.send_header('Content-type', 'text/html')
                    self.end_headers()
                    self.wfile.write(b'State mismatch. Possible CSRF attack.')
                    return

                print(f"Authorization Code: {code}")
                
                token_url = "https://www.linkedin.com/oauth/v2/accessToken"
                data = {
                    "grant_type": "authorization_code",
                    "code": code,
                    "redirect_uri": "http://localhost:3000/callback",
                    "client_id": "<client_id>", 
                    "client_secret": "<client_secret>",
                }

                try:
                    response = requests.post(token_url, data=data)
                    response.raise_for_status() 
                    token_data = response.json()
                    access_token = token_data.get('access_token')

                    if access_token:
                        print(f"Access Token: {access_token}")
                        self.send_response(200)
                        self.send_header('Content-type', 'text/html')
                        self.end_headers()
                        self.wfile.write(b'Authorization successful. Check the console for the access token.')

                    else:
                        print("Failed to retrieve access token:", token_data)
                        self.send_response(500)
                        self.send_header('Content-type', 'text/html')
                        self.end_headers()
                        self.wfile.write(b'Failed to retrieve access token.')

                except requests.exceptions.RequestException as e:
                    print(f"Error during token exchange: {e}")
                    self.send_response(500)
                    self.send_header('Content-type', 'text/html')
                    self.end_headers()
                    self.wfile.write(b'Error during token exchange.')

            else:
                self.send_response(400)
                self.send_header('Content-type', 'text/html')
                self.end_headers()
                self.wfile.write(b'Missing code or state parameter.')
        else:
            self.send_response(404) 
            self.send_header('Content-type', 'text/html')
            self.end_headers()
            self.wfile.write(b'Not found.')

def run(server_class=HTTPServer, handler_class=RequestHandler, port=3000):
    server_address = ('', port)
    httpd = server_class(server_address, handler_class)
    print(f'Starting server on port {port}...')
    httpd.serve_forever()

if __name__ == '__main__':
    run()