better authenticarion

scripts/get_netatmo_token.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Perform Netatmo's initial OAuth authorization-code exchange."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import secrets
+import sys
+import urllib.parse
+import webbrowser
+from pathlib import Path
+
+# Allow direct execution from a source checkout without installing the package.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from netatmo_service.netatmo import NetatmoClient, NetatmoError
+
+AUTHORIZE_URL = "https://api.netatmo.com/oauth2/authorize"
+
+
+def arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--client-id", default=os.getenv("NETATMO_CLIENT_ID"))
+    parser.add_argument("--client-secret", default=os.getenv("NETATMO_CLIENT_SECRET"))
+    parser.add_argument(
+        "--redirect-uri", default=os.getenv("NETATMO_REDIRECT_URI", "http://localhost:8080")
+    )
+    parser.add_argument(
+        "--token-file",
+        type=Path,
+        default=Path(os.getenv("NETATMO_TOKEN_FILE", "./rrd/netatmo_tokens.json")),
+    )
+    parser.add_argument("--no-browser", action="store_true")
+    return parser.parse_args()
+
+
+def authorization_code(value: str, expected_state: str) -> str:
+    value = value.strip()
+    if "://" not in value:
+        return value
+    query = urllib.parse.parse_qs(urllib.parse.urlparse(value).query)
+    if query.get("state", [None])[0] != expected_state:
+        raise ValueError("The callback state does not match; restart authorization")
+    if "error" in query:
+        raise ValueError(f"Authorization failed: {query['error'][0]}")
+    if not query.get("code"):
+        raise ValueError("The callback URL contains no authorization code")
+    return query["code"][0]
+
+
+def main() -> int:
+    args = arguments()
+    if not args.client_id or not args.client_secret:
+        print("Set NETATMO_CLIENT_ID and NETATMO_CLIENT_SECRET first.", file=sys.stderr)
+        return 2
+
+    state = secrets.token_urlsafe(24)
+    url = f"{AUTHORIZE_URL}?" + urllib.parse.urlencode(
+        {
+            "client_id": args.client_id,
+            "redirect_uri": args.redirect_uri,
+            "scope": "read_station",
+            "state": state,
+            "response_type": "code",
+        }
+    )
+    print("\nOpen this URL and approve access:\n")
+    print(url)
+    if not args.no_browser:
+        webbrowser.open(url)
+    print("\nAfter redirect, paste the complete callback URL (or just its code).")
+
+    try:
+        code = authorization_code(input("> "), state)
+        config = {
+            "NETATMO_CLIENT_ID": args.client_id,
+            "NETATMO_CLIENT_SECRET": args.client_secret,
+            "NETATMO_REFRESH_TOKEN": "",
+            "NETATMO_ACCESS_TOKEN": "",
+            "NETATMO_TOKEN_FILE": args.token_file.expanduser().resolve(),
+            "NETATMO_DEVICE_ID": "",
+            "NETATMO_TOKEN_URL": "https://api.netatmo.com/oauth2/token",
+            "NETATMO_STATIONS_URL": "https://api.netatmo.com/api/getstationsdata",
+        }
+        client = NetatmoClient(config)
+        client.exchange_code(code, args.redirect_uri)
+    except (EOFError, ValueError, NetatmoError) as exc:
+        print(f"Token setup failed: {exc}", file=sys.stderr)
+        return 1
+
+    print(f"Tokens saved to {args.token_file} with owner-only permissions.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())