Shopify: Custom App anlegen

#!/usr/bin/env python3
"""
Holt einen Offline Access Token für eine eigene Shopify-App (Dev Dashboard) über den
OAuth Authorization Code Grant. Ohne zusätzliche Pakete, nur Python 3.8+.

Ablauf:
  1. Startet einen kleinen Webserver auf http://localhost:<port>/callback
  2. Öffnet die Shopify-Freigabeseite im Browser
  3. Prüft die Rückleitung (state, shop, HMAC) und tauscht den Code gegen den Token

Voraussetzung in der App (Dev Dashboard, "Create app manually"):
  - Allowed redirection URL(s): http://localhost:3456/callback   (Port wie --port)
  - Use legacy install flow: nicht anhaken
  - Scopes eingetragen, die du brauchst

Aufruf:
  python3 shopify_offline_token.py --shop mein-shop.myshopify.com --client-id <ID> --scopes read_orders,write_products
  (Das Client Secret wird verdeckt abgefragt oder aus SHOPIFY_CLIENT_SECRET gelesen.)

Standard ist ein nicht ablaufender Offline Token. Mit --expiring gibt es stattdessen
einen Token für 1 Stunde plus Refresh Token.
"""
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$")
# Version der Shopify Admin API für die Testabfrage (--check). Bei Bedarf die aktuelle
# Version eintragen, siehe 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"Ungültige Shop-Domain: {value!r} (erwartet: name.myshopify.com)")
    return shop


def valid_hmac(params: dict, secret: str) -> bool:
    """Shopify signiert die Rückleitung: alle Parameter außer hmac, sortiert, key=value mit & verbunden."""
    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):  # keine Zugriffslogs im 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, "Nicht gefunden", "Diese Adresse wird nicht verwendet.")
            params = dict(urllib.parse.parse_qsl(url.query, keep_blank_values=True))
            if "error" in params:
                result["error"] = f'Shopify meldet: {params.get("error")} {params.get("error_description", "")}'.strip()
            elif params.get("state") != state:
                result["error"] = "state stimmt nicht überein. Bitte das Skript neu starten."
            elif params.get("shop", "").lower() != shop:
                result["error"] = f'Rückleitung kam für einen anderen Shop ({params.get("shop")}).'
            elif not valid_hmac(params, client_secret):
                result["error"] = "HMAC-Prüfung fehlgeschlagen. Stimmt das Client Secret?"
            elif not params.get("code"):
                result["error"] = "Kein Autorisierungscode in der Rückleitung."
            else:
                result["code"] = params["code"]
            if "error" in result:
                self.reply(400, "Fehlgeschlagen", result["error"] + " Details im Terminal.")
            else:
                self.reply(200, "Freigabe erhalten", "Du kannst dieses Fenster schließen und zurück ins Terminal wechseln.")
            done.set()

    try:
        server = http.server.HTTPServer(("127.0.0.1", port), Handler)
    except OSError as e:
        raise SystemExit(f"Port {port} ist belegt ({e}). Anderen Port mit --port wählen und in der App eintragen.")
    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. Freigabe im Browser bestätigen. Falls sich kein Browser öffnet, diese Adresse aufrufen:\n")
    print("   " + auth_url + "\n")
    if open_browser:
        webbrowser.open(auth_url)

    if not done.wait(TIMEOUT_SECONDS):
        server.shutdown()
        raise SystemExit(f"Keine Rückleitung innerhalb von {TIMEOUT_SECONDS // 60} Minuten. Abbruch.")
    server.shutdown()
    if "error" in result:
        raise SystemExit("Fehler: " + result["error"])

    print("2. Freigabe erhalten, tausche Code gegen 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-Abruf fehlgeschlagen: 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. Fertig.\n")
    print(f"   Shop:          {shop}")
    print(f"   Scopes:        {tok.get('scope')}")
    print(f"   Gültigkeit:    {'%s Sekunden (Refresh Token gespeichert)' % tok.get('expires_in') if tok.get('expires_in') else 'läuft nicht ab (bis die App deinstalliert oder das Secret erneuert wird)'}")
    print(f"   Gespeichert:   {os.path.abspath(out)} (nur für deinen Benutzer lesbar)")
    if missing:
        print(f"   ACHTUNG: nicht freigegeben wurden {', '.join(missing)}")
    return tok


def main():
    p = argparse.ArgumentParser(description="Offline Access Token für eine eigene Shopify-App holen.")
    p.add_argument("--shop", required=True, help="z. B. mein-shop.myshopify.com")
    p.add_argument("--client-id", required=True, help="Client ID aus den App-Settings im Dev Dashboard")
    p.add_argument("--scopes", required=True, help="Kommagetrennt, z. B. read_orders,write_products")
    p.add_argument("--port", type=int, default=3456, help="Port für die Rückleitung (Standard 3456)")
    p.add_argument("--expiring", action="store_true", help="Ablaufenden Token (1h) plus Refresh Token holen")
    p.add_argument("--out", default="shopify-token.json", help="Datei für den Token (Standard shopify-token.json)")
    p.add_argument("--no-browser", action="store_true", help="Browser nicht automatisch öffnen")
    p.add_argument("--check", action="store_true", help="Token danach mit einer Testabfrage prüfen")
    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 (Eingabe bleibt unsichtbar): ").strip()
    if not secret:
        raise SystemExit("Kein Client Secret angegeben.")

    print(f"\nRückleitungsadresse, die in der App eingetragen sein muss: 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"   Testabfrage:   OK, {check_token(shop, tok['access_token'])}")
        except Exception as e:  # noqa: BLE001
            print(f"   Testabfrage fehlgeschlagen: {e}")


if __name__ == "__main__":
    main()