130 lines
4.1 KiB
Python
Executable File
130 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Small SwitchBot API v1.1 proof-of-concept.
|
|
|
|
By default this lists devices from:
|
|
GET https://api.switch-bot.com/v1.1/devices
|
|
|
|
Credentials are read from environment variables so the token and secret do not
|
|
need to be committed into source files:
|
|
SWITCHBOT_TOKEN=... SWITCHBOT_SECRET=... python3 switchbot_poc.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
|
|
|
|
BASE_URL = "https://api.switch-bot.com/v1.1"
|
|
|
|
|
|
def build_headers(token: str, secret: str) -> dict[str, str]:
|
|
"""Build SwitchBot v1.1 authentication headers."""
|
|
timestamp_ms = str(int(time.time() * 1000))
|
|
nonce = str(uuid.uuid4())
|
|
message = f"{token}{timestamp_ms}{nonce}".encode("utf-8")
|
|
digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).digest()
|
|
signature = base64.b64encode(digest).decode("utf-8")
|
|
|
|
return {
|
|
"Authorization": token,
|
|
"Content-Type": "application/json; charset=utf8",
|
|
"sign": signature,
|
|
"t": timestamp_ms,
|
|
"nonce": nonce,
|
|
}
|
|
|
|
|
|
def switchbot_get(path: str, token: str, secret: str) -> dict:
|
|
"""Perform a signed GET request and parse the JSON response."""
|
|
url = f"{BASE_URL}{path}"
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers=build_headers(token, secret),
|
|
method="GET",
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
body = response.read().decode("utf-8")
|
|
except urllib.error.HTTPError as exc:
|
|
details = exc.read().decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"HTTP {exc.code} from SwitchBot: {details}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise RuntimeError(f"Could not reach SwitchBot API: {exc.reason}") from exc
|
|
|
|
return json.loads(body)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Query the SwitchBot API v1.1 using signed requests."
|
|
)
|
|
parser.add_argument(
|
|
"--endpoint",
|
|
default="/devices",
|
|
help="API path to query, for example /devices, /scenes, or /devices/<id>/status.",
|
|
)
|
|
parser.add_argument(
|
|
"--token",
|
|
default=os.environ.get("SWITCHBOT_TOKEN"),
|
|
help="SwitchBot token. Defaults to SWITCHBOT_TOKEN.",
|
|
)
|
|
parser.add_argument(
|
|
"--secret",
|
|
default=os.environ.get("SWITCHBOT_SECRET"),
|
|
help="SwitchBot secret. Defaults to SWITCHBOT_SECRET.",
|
|
)
|
|
parser.add_argument(
|
|
"--diagnose",
|
|
action="store_true",
|
|
help="Print a short hint when /devices returns an empty inventory.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if not args.token or not args.secret:
|
|
print(
|
|
"Missing credentials. Set SWITCHBOT_TOKEN and SWITCHBOT_SECRET, "
|
|
"or pass --token and --secret.",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
endpoint = args.endpoint if args.endpoint.startswith("/") else f"/{args.endpoint}"
|
|
try:
|
|
payload = switchbot_get(endpoint, args.token, args.secret)
|
|
except (RuntimeError, json.JSONDecodeError) as exc:
|
|
print(f"Request failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
if args.diagnose and endpoint == "/devices":
|
|
device_list = payload.get("body", {}).get("deviceList", [])
|
|
remote_list = payload.get("body", {}).get("infraredRemoteList", [])
|
|
if payload.get("statusCode") == 100 and not device_list and not remote_list:
|
|
print(
|
|
"\nDiagnostic: authentication succeeded, but this SwitchBot "
|
|
"account has no Cloud API-visible devices. Check that the "
|
|
"token belongs to the same app account/home as your devices, "
|
|
"that devices are added to the SwitchBot app, and that "
|
|
"Bluetooth-only devices are reachable via a SwitchBot Hub.",
|
|
file=sys.stderr,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|