Shopify: Creating a custom app

#!/usr/bin/env python3
"""
Gets an offline access token for your own Shopify app (Dev Dashboard) using the
OAuth authorization code grant. No extra packages needed, Python 3.8+ only.

How it works:
  1. Starts a small web server on http://localhost:<port>/callback
  2. Opens the Shopify approval page in your browser
  3. Verifies the redirect (state, shop, HMAC) and exchanges the code for the token

Required app settings (Dev Dashboard, "Create app manually"):
  - Allowed redirection URL(s): http://localhost:3456/callback   (same port as --port)
  - Use legacy install flow: leave unchecked
  - Scopes: the ones you need

Usage:
  python3 shopify_offline_token.py --shop my-shop.myshopify.com --client-id <ID> --scopes read_orders,write_products
  (The client secret is prompted for without echo or read from SHOPIFY_CLIENT_SECRET.)

By default you get a non-expiring offline token. With --expiring you get a token
valid for 1 hour plus a refresh token instead.
"""
import argparse
import getpass
import hashlib
import hmac
import html
import http.server
import json
import os
import re
import secrets
import sys
import threading
import urllib.error
import urllib.parse
import urllib.request
import webbrowser

SHOP_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9-]*\.myshopify\.com$")
# Shopify Admin API version for the test request (--check). If needed, enter the
# current version, see https://shopify.dev/docs/api/usage/versioning
API_VERSION = "2026-10"
TIMEOUT_SECONDS = 300


def normalize_shop(value: str) -> str:
    shop = value.strip().lower()
    shop = re.sub(r"^https?://", "", shop).split("/")[0]
    if "." not in shop:
        shop += ".myshopify.com"
    if not SHOP_RE.match(shop):
        raise ValueError(f"Invalid shop domain: {value!r} (expected: name.myshopify.com)")
    return shop


def valid_hmac(params: dict, secret: str) -> bool:
    """Shopify signs the redirect: all parameters except hmac, sorted, key=value joined with &."""
    received = params.get("hmac", "")
    message = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "hmac")
    expected = hmac.new(secret.encode(), message.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, received)


def post_form(url: str, data: dict) -> dict:
    body = urllib.parse.urlencode(data).encode()
    req = urllib.request.Request(url, data=body, method="POST", headers={
        "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json",
        "User-Agent": "shopify-offline-token-script"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)


def check_token(shop: str, token: str) -> str:
    req = urllib.request.Request(
        f"https://{shop}/admin/api/{API_VERSION}/graphql.json", method="POST",
        data=json.dumps({"query": "{ shop { name myshopifyDomain } }"}).encode(),
        headers={"Content-Type": "application/json", "X-Shopify-Access-Token": token,
                 "User-Agent": "shopify-offline-token-script"})
    with urllib.request.urlopen(req, timeout=30) as r:
        data = json.load(r)
    s = data.get("data", {}).get("shop") or {}
    return f'{s.get("name")} ({s.get("myshopifyDomain")})'


def run(shop, client_id, client_secret, scopes, port, expiring, out, open_browser=True, token_url=None):
    redirect_uri = f"http://localhost:{port}/callback"
    state = secrets.token_urlsafe(24)
    result = {}
    done = threading.Event()

    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a):  # no access logs in the terminal
            pass

        def reply(self, status, title, text):
            page = (f"<!doctype html><meta charset='utf-8'><title>{html.escape(title)}</title>"
                    f"<body style='font:16px/1.5 system-ui;margin:48px;max-width:40em'>"
                    f"<h1 style='font-size:24px'>{html.escape(title)}</h1><p>{html.escape(text)}</p>")
            self.send_response(status)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(page.encode())

        def do_GET(self):
            url = urllib.parse.urlsplit(self.path)
            if url.path != "/callback":
                return self.reply(404, "Not found", "This address is not used.")
            params = dict(urllib.parse.parse_qsl(url.query, keep_blank_values=True))
            if "error" in params:
                result["error"] = f'Shopify reports: {params.get("error")} {params.get("error_description", "")}'.strip()
            elif params.get("state") != state:
                result["error"] = "state does not match. Please restart the script."
            elif params.get("shop", "").lower() != shop:
                result["error"] = f'Redirect came from a different shop ({params.get("shop")}).'
            elif not valid_hmac(params, client_secret):
                result["error"] = "HMAC check failed. Is the client secret correct?"
            elif not params.get("code"):
                result["error"] = "No authorization code in the redirect."
            else:
                result["code"] = params["code"]
            if "error" in result:
                self.reply(400, "Failed", result["error"] + " See the terminal for details.")
            else:
                self.reply(200, "Approval received", "You can close this window and return to the terminal.")
            done.set()

    try:
        server = http.server.HTTPServer(("127.0.0.1", port), Handler)
    except OSError as e:
        raise SystemExit(f"Port {port} is in use ({e}). Choose another port with --port and add it to the app.")
    threading.Thread(target=server.serve_forever, daemon=True).start()

    auth_url = f"https://{shop}/admin/oauth/authorize?" + urllib.parse.urlencode({
        "client_id": client_id, "scope": scopes, "redirect_uri": redirect_uri, "state": state})
    print("\n1. Confirm the approval in your browser. If no browser opens, open this address:\n")
    print("   " + auth_url + "\n")
    if open_browser:
        webbrowser.open(auth_url)

    if not done.wait(TIMEOUT_SECONDS):
        server.shutdown()
        raise SystemExit(f"No redirect within {TIMEOUT_SECONDS // 60} minutes. Aborting.")
    server.shutdown()
    if "error" in result:
        raise SystemExit("Error: " + result["error"])

    print("2. Approval received, exchanging code for token …")
    data = {"client_id": client_id, "client_secret": client_secret, "code": result["code"],
            "expiring": "1" if expiring else "0"}
    try:
        tok = post_form(token_url or f"https://{shop}/admin/oauth/access_token", data)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"Token request failed: HTTP {e.code} {e.read().decode(errors='replace')[:300]}")

    granted = set((tok.get("scope") or "").split(","))
    missing = [s for s in scopes.split(",") if s and s not in granted
               and not (s.startswith("read_") and "write_" + s[5:] in granted)]

    with open(os.open(out, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f:
        json.dump({"shop": shop, **tok}, f, indent=2)

    print("\n3. Done.\n")
    print(f"   Shop:          {shop}")
    print(f"   Scopes:        {tok.get('scope')}")
    print(f"   Validity:      {'%s seconds (refresh token saved)' % tok.get('expires_in') if tok.get('expires_in') else 'does not expire (until the app is uninstalled or the secret is rotated)'}")
    print(f"   Saved:         {os.path.abspath(out)} (readable by your user only)")
    if missing:
        print(f"   WARNING: not granted: {', '.join(missing)}")
    return tok


def main():
    p = argparse.ArgumentParser(description="Get an offline access token for your own Shopify app.")
    p.add_argument("--shop", required=True, help="e.g. my-shop.myshopify.com")
    p.add_argument("--client-id", required=True, help="Client ID from the app settings in the Dev Dashboard")
    p.add_argument("--scopes", required=True, help="Comma-separated, e.g. read_orders,write_products")
    p.add_argument("--port", type=int, default=3456, help="Port for the redirect (default 3456)")
    p.add_argument("--expiring", action="store_true", help="Get an expiring token (1h) plus refresh token")
    p.add_argument("--out", default="shopify-token.json", help="File for the token (default shopify-token.json)")
    p.add_argument("--no-browser", action="store_true", help="Do not open the browser automatically")
    p.add_argument("--check", action="store_true", help="Verify the token with a test request afterwards")
    a = p.parse_args()

    try:
        shop = normalize_shop(a.shop)
    except ValueError as e:
        raise SystemExit(str(e))
    scopes = ",".join(s.strip() for s in a.scopes.split(",") if s.strip())
    secret = os.environ.get("SHOPIFY_CLIENT_SECRET") or getpass.getpass("Client secret (input stays hidden): ").strip()
    if not secret:
        raise SystemExit("No client secret provided.")

    print(f"\nRedirect URL that must be set in the app: http://localhost:{a.port}/callback")
    tok = run(shop, a.client_id.strip(), secret, scopes, a.port, a.expiring, a.out, not a.no_browser)
    if a.check:
        try:
            print(f"   Test request:  OK, {check_token(shop, tok['access_token'])}")
        except Exception as e:  # noqa: BLE001
            print(f"   Test request failed: {e}")


if __name__ == "__main__":
    main()