98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
|
#!/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())
|