epaper-dashboard/tools/netatmo_auth.py

#!/usr/bin/env python3
"""Netatmo OAuth2 Authorization Code Flow — einmaliger Setup-Helper.

Netatmo hat 2025 den OAuth-Password-Grant abgeschaltet. Stattdessen muss
einmalig der Authorization-Code-Flow durchlaufen werden (Browser-Login).

Dieses Script:
  1. Liest oder fragt client_id + client_secret
  2. Startet einen lokalen HTTP-Server auf http://localhost:8765/callback
  3. Öffnet den Browser mit der Netatmo-Authorize-URL
  4. Empfängt den Callback-Code
  5. Tauscht Code gegen access_token + refresh_token
  6. Schreibt refresh_token in config.json unter plugin_configs.netatmo

Verwendung:
  python3 tools/netatmo_auth.py              # interaktiv
  python3 tools/netatmo_auth.py --cid X --csec Y  # nicht-interaktiv

Vor dem ersten Lauf in der Netatmo-Dev-Konsole (https://dev.netatmo.com/apps)
die Redirect-URI `http://localhost:8765/callback` zur App hinzufügen!
"""
from __future__ import annotations
import argparse
import http.server
import json
import os
import socket
import sys
import urllib.parse
import urllib.request
import urllib.error
import webbrowser
from pathlib import Path

# Konstanten
TOKEN_URL = "https://api.netatmo.com/oauth2/token"
AUTHORIZE_URL = "https://api.netatmo.com/oauth2/authorize"
REDIRECT_URI = "http://localhost:8765/callback"
DEFAULT_SCOPE = "read_station"

# Pfad zur config.json (gleicher Pfad wie im Dashboard)
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"


def detect_lan_ip() -> str:
    """Findet die LAN-IP-Adresse (nicht 127.0.0.1) für Redirect-URI.

    Der Pi hat oft mehrere Interfaces (wlan0, eth0, ...). Wir nehmen die
    erste nicht-loopback IPv4-Adresse. Wenn das nicht klappt, wird
    127.0.0.1 zurückgegeben (und der User muss dann manuell auf der
    Pi-Konsole einen SSH-Tunnel mit ssh -L 8765:localhost:8765 machen).
    """
    try:
        import socket
        # Trick: connect zu einem externen Socket (kein Datenverkehr),
        # dann lies die Source-IP — das ist die Default-Route-IP.
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(("8.8.8.8", 80))
            return s.getsockname()[0]
        finally:
            s.close()
    except Exception:
        return "127.0.0.1"


def free_port(port: int = 8765) -> int:
    """Findet einen freien Port (falls 8765 belegt)."""
    for p in [port] + list(range(port + 1, port + 20)):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            try:
                s.bind(("127.0.0.1", p))
                return p
            except OSError:
                continue
    raise RuntimeError(f"Kein freier Port zwischen {port} und {port+19} gefunden")


def load_existing_config() -> dict:
    if not CONFIG_PATH.exists():
        return {}
    try:
        return json.loads(CONFIG_PATH.read_text())
    except Exception as e:
        print(f"WARN: {CONFIG_PATH} nicht lesbar: {e}", file=sys.stderr)
        return {}


def save_refresh_token(refresh_token: str, client_id: str, client_secret: str,
                        username: str) -> dict:
    """Schreibt refresh_token in config.json unter plugin_configs.netatmo."""
    cfg = load_existing_config()
    cfg.setdefault("plugin_configs", {})
    netatmo_cfg = cfg["plugin_configs"].setdefault("netatmo", {})
    netatmo_cfg["client_id"] = client_id
    netatmo_cfg["client_secret"] = client_secret
    netatmo_cfg["refresh_token"] = refresh_token
    if username:
        netatmo_cfg["username"] = username
    CONFIG_PATH.write_text(json.dumps(cfg, indent=2))
    return cfg


def exchange_code_for_tokens(code: str, client_id: str, client_secret: str) -> dict:
    """Authorization-Code → Access + Refresh Token."""
    data = json.dumps({
        "grant_type": "authorization_code",
        "client_id": client_id,
        "client_secret": client_secret,
        "code": code,
        "redirect_uri": REDIRECT_URI,
        "scope": DEFAULT_SCOPE,
    }).encode("utf-8")
    req = urllib.request.Request(TOKEN_URL, data=data,
        headers={"Content-Type": "application/json", "Accept": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read())
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "ignore")
        raise SystemExit(f"Token-Exchange fehlgeschlagen: HTTP {e.code}\n{body}")


def run_callback_server(port: int, expected_state: str) -> str:
    """Startet einen HTTP-Server, der auf den Callback wartet und den Code extrahiert."""
    captured = {}

    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *args, **kwargs):
            pass  # quiet

        def do_GET(self):
            parsed = urllib.parse.urlparse(self.path)
            qs = urllib.parse.parse_qs(parsed.query)
            if "error" in qs:
                captured["error"] = qs.get("error", ["unknown"])[0]
                self.send_response(400)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.end_headers()
                self.wfile.write(
                    b"<h1>Fehler bei Netatmo-Authentifizierung</h1>"
                    b"<p>Bitte zurueck zum Terminal gehen.</p>")
                return
            if "code" not in qs:
                self.send_response(400)
                self.end_headers()
                return
            if qs.get("state", [None])[0] != expected_state:
                self.send_response(400)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.end_headers()
                self.wfile.write(b"<h1>State mismatch (CSRF-Schutz)</h1>")
                return
            captured["code"] = qs["code"][0]
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(
                b"<h1>OK!</h1>"
                b"<p>Du kannst dieses Fenster jetzt schliessen. "
                b"Zurueck zum Terminal fuer den Refresh-Token.</p>")

    server = http.server.HTTPServer(("0.0.0.0", port), Handler)
    server.timeout = 180  # 3 min timeout
    server.handle_request()
    server.handle_request()  # 2nd for /favicon.ico falls Browser fragt
    if "error" in captured:
        raise SystemExit(f"Netatmo-Authentifizierung fehlgeschlagen: {captured['error']}")
    if "code" not in captured:
        raise SystemExit("Timeout oder kein Code empfangen.")
    return captured["code"]


def main():
    parser = argparse.ArgumentParser(description="Netatmo OAuth2-Setup")
    parser.add_argument("--cid", help="Client-ID")
    parser.add_argument("--csec", help="Client-Secret")
    parser.add_argument("--port", type=int, default=8765,
                        help="Lokaler Port für den Callback-Server")
    parser.add_argument("--scope", default=DEFAULT_SCOPE,
                        help=f"OAuth-Scopes (default: {DEFAULT_SCOPE})")
    args = parser.parse_args()

    # 1) Client-Credentials laden
    client_id = args.cid
    if not client_id:
        existing = load_existing_config().get("plugin_configs", {}).get("netatmo", {})
        client_id = existing.get("client_id") or input("Netatmo Client-ID: ").strip()
    client_secret = args.csec
    if not client_secret:
        existing = load_existing_config().get("plugin_configs", {}).get("netatmo", {})
        client_secret = existing.get("client_secret") or input("Netatmo Client-Secret: ").strip()
    if not (client_id and client_secret):
        raise SystemExit("Client-ID und Client-Secret werden benötigt.")

    # 2) Port + State + Redirect-URI
    port = free_port(args.port)
    import secrets
    state = secrets.token_urlsafe(16)

    # Wenn Chromium auf dem Pi vorhanden ist UND ein X-Server läuft,
    # öffnen wir den Browser auf dem Pi selbst. Dann ist 'localhost' die
    # korrekte Redirect-URI. Das ist der einfachste Fall — User muss
    # nichts weiter konfigurieren.
    lan_ip = detect_lan_ip()
    import os
    if os.path.exists("/tmp/.X11-unix/X0") and not os.environ.get("DISPLAY"):
        os.environ["DISPLAY"] = ":0"
    chromium_local = any(os.path.exists(p) for p in
                         ("/usr/bin/chromium", "/usr/bin/chromium-browser",
                          "/usr/bin/google-chrome"))

    if chromium_local and os.environ.get("DISPLAY"):
        # Chromium läuft auf dem Pi selbst → localhost funktioniert.
        redirect = f"http://localhost:{port}/callback"
        target_host = "localhost"
        print()
        print("Info: Chromium auf Display :0 erkannt. Verwende localhost.")
        print("      (Öffne ein Terminal-Fenster auf dem Pi und schau zu.)")
    else:
        # Kein lokaler Browser → Redirect muss LAN-IP haben, und der
        # User öffnet die URL in seinem Laptop/Phone-Browser.
        redirect = f"http://{lan_ip}:{port}/callback"
        target_host = lan_ip
        if redirect.startswith("http://localhost"):
            print()
            print("WARN: Konnte keine LAN-IP ermitteln. Verwende localhost.")
            print("      Falls du per SSH verbunden bist: exit und dann")
            print("      'ssh -L 8765:localhost:8765 pi@...' um den Tunnel zu öffnen.")

    # 3) Authorize-URL
    auth_url = (
        f"{AUTHORIZE_URL}"
        f"?client_id={urllib.parse.quote(client_id)}"
        f"&redirect_uri={urllib.parse.quote(redirect)}"
        f"&scope={urllib.parse.quote(args.scope)}"
        f"&state={urllib.parse.quote(state)}"
        f"&response_type=code"
    )

    print(f"\n=== Netatmo OAuth Setup ===\n")
    print(f"Local callback server listening on {redirect}")
    print()
    print(f"WICHTIG: Diese Redirect-URI muss EXAKT in deiner Netatmo-App")
    print(f"auf https://dev.netatmo.com/apps/ registriert sein!")
    print()
    print(f"  Aktuelle Redirect-URI:")
    print(f"    {redirect}")
    print()
    if redirect.startswith("http://localhost"):
        print(f"  Falls deine App auf https://dev.netatmo.com/apps/ mit der")
        print(f"  URI 'http://localhost:8765/callback' registriert ist, dann")
        print(f"  funktioniert der Helper NUR wenn du direkt auf dem Pi ein")
        print(f"  Terminal hast (Browser auf dem Pi). Falls du per SSH")
        print(f"  verbunden bist: 'exit' und mit 'ssh -L 8765:localhost:8765'")
        print(f"  neu verbinden, dann Browser auf localhost:8765 öffnen.")
        print()
        print(f"  Alternative: du kannst die App-URI auf")
        print(f"  '{redirect}' ändern, dann funktioniert der Helper vom")
        print(f"  Laptop/Phone aus über diese URL.")
    print()
    print(f"Oeffne in deinem Browser:\n  {auth_url}\n")
    try:
        # Wenn ein X-Server läuft (Pi mit Bildschirm, Desktop-Session),
        # explizit DISPLAY=:0 setzen und Chromium direkt öffnen. Das
        # umgeht xdg-open-Fallbacks die oft auf headless-Setups scheitern.
        import os
        if os.path.exists("/tmp/.X11-unix/X0") and not os.environ.get("DISPLAY"):
            os.environ["DISPLAY"] = ":0"
        # Chromium-Binary auf dem Pi finden
        chromium = None
        for cand in ("chromium", "chromium-browser", "google-chrome"):
            from shutil import which
            p = which(cand)
            if p:
                chromium = p
                break
        if chromium and os.environ.get("DISPLAY"):
            # Chromium im App-Mode öffnen — kein Toolbar, sieht wie eine
            # native App aus. Fenster schließt nach Redirect automatisch
            # durch unser Callback-Handler.
            print(f"Oeffne Chromium auf Display {os.environ['DISPLAY']}...")
            import subprocess
            subprocess.Popen(
                [chromium, f"--app={auth_url}",
                 "--no-default-browser-check",
                 "--no-first-run"],
                env=dict(os.environ),
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
        elif webbrowser.open(auth_url):
            print("(Browser wurde geoeffnet.)")
        else:
            print("(Browser konnte nicht automatisch geoeffnet werden — bitte URL manuell öffnen.)")
    except Exception as e:
        print(f"(Browser-Fehler: {e} — bitte URL manuell öffnen.)")

    # 4) Auf Callback warten
    print("\nWarte auf Callback...")
    code = run_callback_server(port, state)

    # 5) Code → Tokens
    print("\nTausche Code gegen Tokens...")
    tok = exchange_code_for_tokens(code, client_id, client_secret)
    refresh_token = tok.get("refresh_token")
    access_token = tok.get("access_token")
    expires_in = tok.get("expires_in", 0)
    if not refresh_token:
        raise SystemExit(f"Antwort enthält keinen refresh_token: {tok}")

    print(f"  access_token:  {access_token[:24]}...")
    print(f"  refresh_token: {refresh_token[:24]}...")
    print(f"  expires_in:    {expires_in}s ({expires_in // 60} min)")

    # 6) In config.json speichern
    save_refresh_token(refresh_token, client_id, client_secret,
                        username="")
    print(f"\nGespeichert in {CONFIG_PATH}")
    print("\nFertig! Das Plugin holt sich ab jetzt selbst neue Access-Tokens.")
    print("Tipp: dashboard.py / epaper-admin restarten damit das Plugin den "
          "neuen refresh_token liest.\n")


if __name__ == "__main__":
    main()