petr.polezhaev
97842276c8
- fglair-discover: fchmod 0600 и при перезаписи существующего --out. - fglctl: #include <cstdlib>; точная проверка обрезки config (n == sizeof-1 && !feof); валидация 'monitor <sec>' (endptr/errno). - docs: отчёт soak — формулировка «ответы на batch GET»; индекс reports. Ревью M4 независимым агентом: 2 круга, APPROVED.
221 lines
8.9 KiB
Python
Executable File
221 lines
8.9 KiB
Python
Executable File
#!/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)
|
||
os.fchmod(fd, 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()
|