How To Talk To OAuth2 Server With Authorization Code Flow From The Console

OAuth2 is the most common authentication protocol for HTTP APIs. It is a beast. How to talk to it when all you have is a console CLI?

Let me first note that using Authorization Code Flow should be your last option. It is intended for web servers and you should use those flows instead, in the order of preference.

Best and Easiest: PKCE

If your API supports Proof Key for Code Exchange, you are lucky and enjoy it. It’s not as straightforward under the hood, but there are plenty of libraries out there and PKCE is intended exactly for CLIs and Single Page Applications.

PKCE allows you to get away with only having a Client ID. The key difference from the code exchange of grant flow is that client has to generate a cryptographically strong code_verifier and derive a code_challenge. That is not a problem in CLI but was a problem in the browser until recently.

Auth0 PKCE page describes the dance well. If you insist on manual verification, Stefaan has a neat step-by-step guide for Python.

Simplest and Deemed Insecure: Implicit Grant

Implicit Grant is a less secure predecessor of PKCE. As for PKCE, you only get a Client ID, but the flow looks more like Authorization Code below. The key difference is that when server redirects user back to your app, an Access Token is included directly in the URL. This makes it vulnerable for sniffing, being retained in logs, et cetera. It’s up to you whether that risk is worth it for your application.

That said, the flow is deprecated in the favor of PKCE and you may not find APIs that support it.

Convoluted and Working: Authorization Code Flow

You may end up working with an API that doesn’t support any of the flow above. For me, it was Strava when I worked on generating my weeknotes. Time to get our hands dirty. I’ll use Python for the examples, but this should be easy enough in your favorite language.

First, register your application (for Strava, that would be their API page in settings). For the callback domain, use localhost. If your server asks for a full callback URL, use http://localhost:9999/oauth2callback.

Storing Client Secret Properly

When you launch your CLI, ask for Client ID and Client Secret. Remember that secret, is, well, secret: do not display it and do not store it in clear text anywhere and not on the disk. Use respective system storage using a helpful keyring library (pip install keyring).

For demonstration purposes, I am storing client ID on a config file, but it can be in the keyring as well.

from getpass import getpass
import json
from pathlib import Path

from keyring import get_keyring, get_password


APPLICATION_KEYRING_NAME = "blog.almad.oauth2.example"
# Mac OS; Use respective config folders
PRIVATE_FOLDER = Path(os.path.expanduser("~")) / "Library/Application Support/OAuth2 Example"
CONFIG_FILE = PRIVATE_FOLDER / "config.json"

client_id = input("Enter client id: ").strip()
client_secret = getpass("Enter client secret: ").strip()

keyring.set_password(APPLICATION_KEYRING_NAME, client_id, client_secret)

if not os.path.exists(PRIVATE_FOLDER):
    os.makedirs(PRIVATE_FOLDER)

CONFIG_FILE.write_text(json.dumps(dict(client_id=client_id)))

Get Server Ready And Ask User To Log In

As said, Authorization Code Flow is intended for the server. Let’s launch one locally. Once we’ll get a request, we’ll use the code query parameter to exchange it for access and refresh token. Both are secrets as well, hence we’ll store them on the keychain again.

The access token will expire and it’s our responsibility to keep track of it. Note that Strava returns expiration as a UNIX timestamp, but you may encounter other time formats as well.

Note that I am using the storage as a form of intra-thread communication.

For simplicity, I’ve hardcoded the port server is launching on. It would be better to have free port detection.

from http.server import SimpleHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

SERVER_PORT=9999


def get_access_token_from_strava(client_id, client_secret):
    class ResponseStoringServer(SimpleHTTPRequestHandler):
        def do_GET(self):
            # Retrieve OAuth2 "code" parameter
            url_params = parse_qs(urlparse(self.path).query)
            code = url_params["code"][0]

            self.exchange_code(code)

            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(
                "<html><body>All OK, go back to console. <script>window.close('','_parent','');</script></body></html>".encode(
                    "utf-8"
                )
            )

        def exchange_code(self, oauth_code):
            # Exchange Code for Access and Refresh token
            response = requests.post(
                url="https://www.strava.com/oauth/token",
                data={
                    "client_id": client_id,
                    "client_secret": client_secret,
                    "code": oauth_code,
                    "grant_type": "authorization_code",
                },
            )
            response.raise_for_status()
            # And store them to secrets storage
            store_tokens_from_response(client_id, response.json())

    with HTTPServer(("", SERVER_PORT), ResponseStoringServer) as httpd:
        thread = Thread(target=httpd.serve_forever, daemon=True)
        thread.start()
            
        # `open` opens the given URL in the user’s default browser on Mac OS X
        # On Linux, use `xdg-open`. On Windows, `explore` may do the trick, 
        # but every time I invoke console command on win, a computer catches fire
        run(
            [
                "open",
                f"https://www.strava.com/oauth/authorize?client_id={client_id}&response_type=code&redirect_uri=http://localhost:{SERVER_PORT}/exchange_token&approval_prompt=force&scope=activity:read",
            ]
        )

        while not get_password(APPLICATION_KEYRING_NAME, client_id + ".access_token"):
            sleep(1)

        httpd.shutdown()
        thread.join(timeout=5)

def store_tokens_from_response(client_id, response_json):
    keyring = get_keyring()
    keyring.set_password(
        APPLICATION_KEYRING_NAME,
        client_id + ".refresh_token",
        response_json["refresh_token"],
    )

    keyring.set_password(
        APPLICATION_KEYRING_NAME,
        client_id + ".access_token",
        json.dumps(
            {
                "token": response_json["access_token"],
                "expires_at": (
                    datetime.now() + timedelta(seconds=response_json["expires_in"])
                ).isoformat(),
            }
        ),
    )

Make a Request

You have Access Token now! Make a request:

import requests
    response = requests.get(
    f"https://www.strava.com/api/v3/activities",
    headers={"Authorization": f"Bearer {access_token}"},
    params=dict(page=1, per_page=10),
)
response.raise_for_status()
print(response.json())

That’s it! Just kidding. But almost.

Use a Refresh Token

Access tokens are intentionally short-lived. Before using them, you need to check if they are still valid and upon expiration, get a new one using refresh token.

from datetime import datetime
import json
def get_refreshed_access_token(client_id, client_secret, refresh_token):
    response = requests.post(
        url="https://www.strava.com/oauth/token",
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "refresh_token": refresh_token,
            "grant_type": "refresh_token",
        },
    )
    response.raise_for_status()
    response_json = response.json()
    store_tokens_from_response(client_id, response_json)

    return response_json["access_token"]


stored_access_token = get_password(
    APPLICATION_KEYRING_NAME, config["client_id"] + ".access_token"
)

if stored_access_token:
    stored_access_token = json.loads(stored_access_token)
    expire_at = datetime.fromisoformat(stored_access_token["expires_at"])
    if expire_at > datetime.now():
        return stored_access_token["token"]

stored_refresh_token = get_password(
    APPLICATION_KEYRING_NAME, config["client_id"] + ".refresh_token"
)

if stored_refresh_token:
    return get_refreshed_access_token(
        config["client_id"], config["client_secret"], stored_refresh_token
    )

But this is really it!

Conclusion

Of course, this is a hack. But since I doubt I’ll be able to pursue Strava using tweets, I just have to deal with the cards I have. From my experience dealing with internal APIs in certain larger companies, this is not the only API with a problem. Hope you’ll manage to survive it too.

I am using the code above to generate my weekly notes. You can find the library on GitHub.

Published in Notes and tagged •

All texts written by . I'd love to hear your feedback. If you've liked this, you may want to subscribe for my monthly newsletter, RSS , or Mastodon. You can always return to home page or read about the site and its privacy handling.