Last active
February 5, 2023 12:37
-
-
Save nitori/e139f5ad6d8674fd3bf3fc313220d299 to your computer and use it in GitHub Desktop.
Uses requests and flask to make a simple quick example for a webbased oauth2 user authentication. Note that this expects the webpart to actually be a proper server in any real scenario. For pure client-side applications currently only "Implicit grant flow" can be used (PKCE is currently not available afaik).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from multiprocessing import Process, Queue | |
| from urllib.parse import urlparse, urlunparse, urlencode, parse_qs | |
| import os | |
| import random | |
| import time | |
| class TwitchAuthError(Exception): | |
| pass | |
| class RequestDenied(TwitchAuthError): | |
| pass | |
| class InvalidState(TwitchAuthError): | |
| pass | |
| class InvalidCode(TwitchAuthError): | |
| pass | |
| def webapp(q, randstate): | |
| import requests | |
| from flask import Flask, request, abort | |
| redirect_uri = urlparse(os.environ['TWITCH_REDIRECT_URI']) | |
| app = Flask(__name__) | |
| @app.route(redirect_uri.path) | |
| def authpath(): | |
| state = request.args['state'] | |
| if state != randstate: | |
| msg = 'Invalid state provided.' | |
| q.put(InvalidState(msg)) | |
| abort(400, msg) | |
| return | |
| if 'error' in request.args: | |
| msg = f'Authorization server returned error: {request.args["error"]}\n' \ | |
| f'Description: {request.args["error_description"]}' | |
| q.put(RequestDenied(msg)) | |
| return msg | |
| code = request.args['code'] | |
| r = requests.post('https://id.twitch.tv/oauth2/token', params={ | |
| 'client_id': os.environ['TWITCH_CLIENT_ID'], | |
| 'client_secret': os.environ['TWITCH_CLIENT_SECRET'], | |
| 'code': code, | |
| 'grant_type': 'authorization_code', | |
| 'redirect_uri': os.environ['TWITCH_REDIRECT_URI'], | |
| }) | |
| if r.status_code != 200: | |
| msg = f'Something went wrong. Authorization server returned status code {r.status_code} with reason: {r.reason}' | |
| q.put(InvalidCode(msg)) | |
| abort(500, msg) | |
| return | |
| data = r.json() | |
| q.put(data) | |
| return 'Success! You can close this window now.' | |
| app.run('localhost', redirect_uri.port, debug=False, use_reloader=False, threaded=False) | |
| def startup_webapp(randstate): | |
| q = Queue() | |
| p = Process(target=webapp, args=(q, randstate)) | |
| p.start() | |
| try: | |
| result = q.get() | |
| if isinstance(result, BaseException): | |
| raise result | |
| finally: | |
| time.sleep(1) | |
| p.terminate() | |
| return result | |
| def main(): | |
| randstate = random.randbytes(16).hex() | |
| query_string = urlencode({ | |
| 'response_type': 'code', | |
| 'client_id': os.environ['TWITCH_CLIENT_ID'], | |
| 'redirect_uri': os.environ['TWITCH_REDIRECT_URI'], | |
| 'scope': 'channel:manage:polls channel:read:polls', | |
| 'state': randstate, | |
| }) | |
| url = f'https://id.twitch.tv/oauth2/authorize?{query_string}' | |
| print('*' * 80) | |
| print(url) | |
| print('*' * 80) | |
| try: | |
| result = startup_webapp(randstate) | |
| except RequestDenied: | |
| print('You denied the request. Good bye!') | |
| return | |
| print(result) | |
| if __name__ == '__main__': | |
| from dotenv import load_dotenv | |
| # required env variables: | |
| # TWITCH_CLIENT_ID | |
| # TWITCH_CLIENT_SECRET | |
| # TWITCH_REDIRECT_URI | |
| load_dotenv() | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment