#!/usr/bin/env python3 """fglair-discover — облачный provisioning устройств FGLair (Ayla). Получает lanip_key/lanip_key_id и параметры устройств из облака Ayla (docs/PROTOCOL.md §7) и печатает конфиг для fglair-core. Ключ статичен (зашит в модуль), облако используется только здесь. Использование: fglair-discover --region eu --email user@example.com [--password ...] [--device "Living Room"] [--format json|esphome-secrets] [--out FILE] Пароль: --password или env FGLAIR_PASSWORD (иначе getpass). --api-base URL — внутренний флаг для тестов (мок облака). Форматы: json — конфиг в формате config_*.json (совместим с legacy и fglctl) esphome-secrets — блок для secrets.yaml (dsn/lanip_key/lanip_key_id) """ import argparse import getpass import json import os import ssl import sys import urllib.error import urllib.request # Секреты приложений FGLair (PROTOCOL.md §7, из APK FGLair 3.4.3). REGIONS = { "eu": { "user_server": "user-field-eu.aylanetworks.com", "device_server": "ads-eu.aylanetworks.com", "app_id": "FGLair-eu-id", "app_secret": "FGLair-eu-gpFbVBRoiJ8E3QWJ-QRULLL3j3U", }, "us": { "user_server": "user-field.aylanetworks.com", "device_server": "ads-field.aylanetworks.com", "app_id": "CJIOSP-id", "app_secret": "CJIOSP-Vb8MQL_lFiYQ7DKjN0eCFXznKZE", }, "cn": { "user_server": "user-field.ayla.com.cn", "device_server": "ads-field.ayla.com.cn", "app_id": "FGLairField-cn-id", "app_secret": "FGLairField-cn-zezg7Y60YpAvy3HPwxvWLnd4Oh4", }, } USER_AGENT = "fglair-discover/0.1 (fglair-core)" def _http_json(method, url, payload=None, headers=None, insecure=False): """Запрос с JSON-ответом; возвращает (status, parsed).""" data = json.dumps(payload).encode() if payload is not None else None hdrs = {"User-Agent": USER_AGENT, "Accept": "application/json"} if data is not None: hdrs["Content-Type"] = "application/json" if headers: hdrs.update(headers) req = urllib.request.Request(url, data=data, headers=hdrs, method=method) ctx = None if insecure: ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=30, context=ctx) as resp: return resp.status, json.loads(resp.read().decode() or "null") except urllib.error.HTTPError as e: try: body = json.loads(e.read().decode() or "null") except Exception: body = None return e.code, body except urllib.error.URLError as e: sys.exit(f"Сеть недоступна ({url}): {e.reason}") def sign_in(args, region, build): payload = { "user": { "email": args.email, "password": args.password, "application": { "app_id": region["app_id"], "app_secret": region["app_secret"], }, } } status, data = _http_json( "POST", build(region["user_server"], "/users/sign_in.json"), payload, insecure=args.insecure) if status != 200 or not isinstance(data, dict): detail = data.get("error") if isinstance(data, dict) else data sys.exit(f"Ошибка входа ({status}): {detail}") token = data.get("access_token") if not token: sys.exit(f"Вход без access_token (статус {status})") return token def get_devices(args, region, token, build): status, data = _http_json( "GET", build(region["device_server"], "/apiv1/devices.json"), headers={"Authorization": f"auth_token {token}"}, insecure=args.insecure) if status != 200 or not isinstance(data, list): sys.exit(f"Ошибка списка устройств ({status}): {data}") devices = [d.get("device", {}) for d in data if isinstance(d, dict)] if not all(d.get("dsn") for d in devices): sys.exit("Облако вернуло устройство без dsn — неожиданный формат") return devices def get_lanip(args, region, token, dsn, build): status, data = _http_json( "GET", build(region["device_server"], f"/apiv1/dsns/{dsn}/lan.json"), headers={"Authorization": f"auth_token {token}"}, insecure=args.insecure) if status != 200 or not isinstance(data, dict): sys.exit(f"Ошибка lan.json для {dsn} ({status}): {data}") return data.get("lanip") or {} def main(): ap = argparse.ArgumentParser( description="Облачный discovery устройств FGLair (fglair-core)") ap.add_argument("--region", choices=sorted(REGIONS), default="eu") ap.add_argument("--email", default=os.environ.get("FGLAIR_EMAIL")) ap.add_argument("--password", default=os.environ.get("FGLAIR_PASSWORD")) ap.add_argument("--device", help="Фильтр по имени устройства (product_name)") ap.add_argument("--format", choices=["json", "esphome-secrets"], default="json") ap.add_argument("--out", help="Записать в файл (иначе stdout)") ap.add_argument("--insecure", action="store_true", help="Не проверять TLS-сертификаты (самоподписанные прокси)") ap.add_argument("--api-base", help="ВНУТРЕННЕЕ: базовый URL мок-облака (тесты)") args = ap.parse_args() if not args.email: sys.exit("Укажите --email или FGLAIR_EMAIL") if not args.password: args.password = getpass.getpass("Пароль FGLair: ") region = dict(REGIONS[args.region]) if args.api_base: # тестовый мок: оба сервера на одном хосте, plain http host = args.api_base.removeprefix("http://").removeprefix("https://") region["user_server"] = host region["device_server"] = host scheme = "http" else: scheme = "https" def build(server, path): return f"{scheme}://{server}{path}" token = sign_in(args, region, build) devices = get_devices(args, region, token, build) if not devices: sys.exit("У аккаунта нет устройств") outputs = [] for dev in devices: if args.device and dev.get("product_name") != args.device: continue dsn = dev["dsn"] lanip = get_lanip(args, region, token, dsn, build) lanip_key = lanip.get("lanip_key") lanip_key_id = lanip.get("lanip_key_id") if not lanip_key or lanip_key_id is None: print(f"ВНИМАНИЕ: {dev.get('product_name')}: lan.json без ключа, " "пропуск", file=sys.stderr) continue outputs.append({ "name": dev.get("product_name", dsn), "app": f"fglair-{args.region}", "model": dev.get("oem_model", ""), "dsn": dsn, "temp_type": "C" if args.region == "eu" else "F", "mac_address": (dev.get("mac") or "").replace(":", "").lower(), "ip_address": dev.get("lan_ip", ""), "lanip_key": lanip_key, "lanip_key_id": lanip_key_id, }) if not outputs: sys.exit("Подходящих устройств не найдено") if args.format == "esphome-secrets": lines = [] for d in outputs: slug = "".join(c for c in d["name"].lower() if c.isalnum()) or "ac" lines.append(f"# {d['name']} ({d['model']})") lines.append(f"{slug}_dsn: \"{d['dsn']}\"") lines.append(f"{slug}_lanip_key: \"{d['lanip_key']}\"") lines.append(f"{slug}_lanip_key_id: {d['lanip_key_id']}") lines.append("") text = "\n".join(lines) else: # По одной компактной строке на устройство (парсится построчно; # каждая строка — самостоятельный конфиг config_*.json). text = "\n".join(json.dumps(d, ensure_ascii=False) for d in outputs) + "\n" if args.out: # Ключ устройства — секрет: файл только для владельца. fd = os.open(args.out, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w") as f: f.write(text) print(f"Записано: {args.out} (права 0600)", file=sys.stderr) else: sys.stdout.write(text) if __name__ == "__main__": main()