throwback/do_oauth.py at main · netdevjamie/throwback · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
"""
One-time OAuth flow. Spins up a local web server on http://localhost:8000,
opens browser to Strava's authorize URL, captures the ?code= callback,
exchanges it for tokens, and writes them back to .env.

Run once after creating the Strava app:
    .venv/bin/python do_oauth.py
"""

import os
import secrets
import sys
import threading
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import urlparse, parse_qs

from dotenv import load_dotenv, set_key
from stravalib import Client

ENV_PATH = Path(__file__).parent / ".env"
load_dotenv(ENV_PATH)

CLIENT_ID = os.getenv("STRAVA_CLIENT_ID")
CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET")

if not CLIENT_ID or not CLIENT_SECRET:
    sys.exit("Missing STRAVA_CLIENT_ID or STRAVA_CLIENT_SECRET in .env")

REDIRECT_URI = "http://localhost:8000/callback"
SCOPES = ["activity:write", "activity:read_all"]
CALLBACK_TIMEOUT_SECONDS = 600
EXPECTED_STATE = secrets.token_urlsafe(16)

captured_code = {"value": None}
captured_event = threading.Event()


class CallbackHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        qs = parse_qs(urlparse(self.path).query)
        code = qs.get("code", [None])[0]
        state = qs.get("state", [None])[0]
        error = qs.get("error", [None])[0]
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        if error:
            self.wfile.write(f"<h1>Authorization failed: {error}</h1>".encode())
        elif state != EXPECTED_STATE:
            self.wfile.write(b"<h1>State mismatch. Request rejected.</h1>")
        elif code:
            captured_code["value"] = code
            captured_event.set()
            self.wfile.write(b"<h1>Authorized. You can close this tab.</h1>")
        else:
            self.wfile.write(b"<h1>No code received.</h1>")

    def log_message(self, *args):
        pass


def main():
    client = Client()
    auth_url = client.authorization_url(
        client_id=int(CLIENT_ID),
        redirect_uri=REDIRECT_URI,
        scope=SCOPES,
        state=EXPECTED_STATE,
    )

    server = HTTPServer(("localhost", 8000), CallbackHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()

    print(f"Opening browser. If it doesn't open, paste this URL:\n  {auth_url}\n")
    webbrowser.open(auth_url)

    print("Waiting for authorization callback on http://localhost:8000/callback ...")
    if not captured_event.wait(timeout=CALLBACK_TIMEOUT_SECONDS):
        server.shutdown()
        sys.exit(f"Timed out after {CALLBACK_TIMEOUT_SECONDS}s waiting for browser authorization.")
    server.shutdown()

    print("Got authorization code. Exchanging for tokens...")
    token_response = client.exchange_code_for_token(
        client_id=int(CLIENT_ID),
        client_secret=CLIENT_SECRET,
        code=captured_code["value"],
    )

    set_key(str(ENV_PATH), "STRAVA_ACCESS_TOKEN", token_response["access_token"])
    set_key(str(ENV_PATH), "STRAVA_REFRESH_TOKEN", token_response["refresh_token"])

    print("Tokens saved to .env. OAuth complete.")


if __name__ == "__main__":
    main()