84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Perform Netatmo's initial OAuth authorization-code exchange."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
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.config import load_config
|
|
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("--config", type=Path, default=Path("config.yaml"))
|
|
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()
|
|
try:
|
|
config = load_config(args.config)
|
|
except RuntimeError as exc:
|
|
print(exc, file=sys.stderr)
|
|
return 2
|
|
if not config["NETATMO_CLIENT_ID"] or not config["NETATMO_CLIENT_SECRET"]:
|
|
print("Set netatmo.client_id and netatmo.client_secret in config.yaml first.", file=sys.stderr)
|
|
return 2
|
|
|
|
state = secrets.token_urlsafe(24)
|
|
url = f"{AUTHORIZE_URL}?" + urllib.parse.urlencode(
|
|
{
|
|
"client_id": config["NETATMO_CLIENT_ID"],
|
|
"redirect_uri": config["NETATMO_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)
|
|
client = NetatmoClient(config)
|
|
client.exchange_code(code, config["NETATMO_REDIRECT_URI"])
|
|
except (EOFError, ValueError, NetatmoError) as exc:
|
|
print(f"Token setup failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Tokens saved to {config['NETATMO_TOKEN_FILE']} with owner-only permissions.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|