From 1a34118ab53d2ec42223e46de99a7f27dfee042d Mon Sep 17 00:00:00 2001 From: Ignace Date: Sat, 22 Aug 2026 16:12:14 +0200 Subject: [PATCH] first setup of the controller --- .gitignore | 3 + README.md | 38 ++++++++ home-control.service | 32 +++++++ lib/__init__.py | 1 + lib/_arrival_detection.py | 98 ++++++++++++++++++++ lib/_zyxel.py | 184 ++++++++++++++++++++++++++++++++++++++ notify.py | 95 ++++++++++++++++++++ requirements.txt | 3 + service.py | 59 ++++++++++++ service.yaml | 33 +++++++ tests/__init__.py | 1 + tests/test_service.py | 57 ++++++++++++ 12 files changed, 604 insertions(+) create mode 100644 .gitignore create mode 100644 home-control.service create mode 100644 lib/__init__.py create mode 100644 lib/_arrival_detection.py create mode 100644 lib/_zyxel.py create mode 100644 notify.py create mode 100644 requirements.txt create mode 100644 service.py create mode 100644 service.yaml create mode 100644 tests/__init__.py create mode 100644 tests/test_service.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a160f49 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ diff --git a/README.md b/README.md index 1c508db..8355e90 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,40 @@ # home_control +Detects an arrival when an iPhone reconnects to a Zyxel EX5601-T1. The service +logs into the router's encrypted web API and polls its active LAN clients. + +1. Install dependencies: `python -m pip install -r requirements.txt` +2. Fill in the router login and the iPhone Wi-Fi MAC address under + `arrival_detection` in `service.yaml`. +3. Test connectivity: `python service.py --once` +4. Run continuously: `python service.py` + +An arrival is logged after the phone was confirmed absent and then present. It +also calls `send_notification()` from `notify.py`; its message, ntfy topic, and +timeout are configured under `arrival_detection.notification`. Set +`arrival_detection.enabled` to `false` to disable this function. + +On the iPhone, open **Settings > Wi-Fi**, tap the info button beside the home +network, and copy **Wi-Fi Address**. If Private Wi-Fi Address is enabled, that +per-network address is the correct one to configure. + +## Install as a systemd service + +The included unit expects the project and its virtual environment at +`/opt/home_control`, owned by a dedicated `home-control` system user: + +```bash +sudo useradd --system --home-dir /opt/home_control --shell /usr/sbin/nologin home-control +sudo chown -R home-control:home-control /opt/home_control +sudo chmod 600 /opt/home_control/service.yaml +sudo cp /opt/home_control/home-control.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now home-control.service +``` + +Check its state and follow its logs with: + +```bash +sudo systemctl status home-control.service +sudo journalctl -u home-control.service -f +``` diff --git a/home-control.service b/home-control.service new file mode 100644 index 0000000..341aebe --- /dev/null +++ b/home-control.service @@ -0,0 +1,32 @@ +[Unit] +Description=Home Control Service +Documentation=file:/opt/home_control/README.md +Wants=network-online.target +After=network-online.target + +[Service] +Type=simple +User=home-control +Group=home-control +WorkingDirectory=/opt/home_control +ExecStart=/opt/home_control/.venv/bin/python /opt/home_control/service.py --config /opt/home_control/service.yaml +Restart=on-failure +RestartSec=15s +TimeoutStopSec=20s + +# The service only needs to read its application and configuration. +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ProtectKernelTunables=true +ProtectKernelModules=true +ProtectKernelLogs=true +ProtectControlGroups=true +RestrictSUIDSGID=true +RestrictRealtime=true +LockPersonality=true +MemoryDenyWriteExecute=true + +[Install] +WantedBy=multi-user.target diff --git a/lib/__init__.py b/lib/__init__.py new file mode 100644 index 0000000..35ad3e9 --- /dev/null +++ b/lib/__init__.py @@ -0,0 +1 @@ +"""Router integrations used by the home-control service.""" diff --git a/lib/_arrival_detection.py b/lib/_arrival_detection.py new file mode 100644 index 0000000..48868d9 --- /dev/null +++ b/lib/_arrival_detection.py @@ -0,0 +1,98 @@ +"""iPhone arrival detection backed by a Zyxel router's active-client list.""" + +from __future__ import annotations + +import logging +import time + +from lib._zyxel import ZyxelError, ZyxelRouter, normalize_mac +from notify import send_notification + +LOG = logging.getLogger("home_control.arrival_detection") + + +class ArrivalDetector: + """Debounce connection samples and emit absent -> present transitions.""" + + def __init__(self, absent_after: int, present_after: int) -> None: + if absent_after < 1 or present_after < 1: + raise ValueError("absent_after and present_after must be at least 1") + self.absent_after = absent_after + self.present_after = present_after + self.state: bool | None = None + self.present_samples = 0 + self.absent_samples = 0 + + def sample(self, present: bool) -> bool: + self.present_samples = self.present_samples + 1 if present else 0 + self.absent_samples = self.absent_samples + 1 if not present else 0 + if self.state is None: + if self.present_samples >= self.present_after: + self.state = True + elif self.absent_samples >= self.absent_after: + self.state = False + return False + if self.state and self.absent_samples >= self.absent_after: + self.state = False + elif not self.state and self.present_samples >= self.present_after: + self.state = True + return True + return False + + +def _build_router(config: dict) -> ZyxelRouter: + router = config["zyxel"] + return ZyxelRouter( + router["host"], router["username"], router["password"], + timeout=float(router.get("timeout", 10)), + verify_tls=bool(router.get("verify_tls", True)), + hosts_oid=str(router.get("hosts_oid", "lanhosts")), + ) + + +def _notify(config: dict, device: dict) -> None: + LOG.info("ARRIVAL: %s", device["name"]) + notification = config.get("notification", {}) + if not notification.get("enabled", True): + return + message = str(notification.get("message", "{name} arrived home")).format( + name=device["name"], mac=device["mac"] + ) + send_notification( + message, + topic_url=notification.get("topic_url") or None, + timeout=float(notification.get("timeout", 10)), + ) + + +def run(config: dict, stop_requested, *, once: bool = False) -> int: + """Run arrival detection until stop_requested returns true.""" + device = dict(config["device"]) + device["mac"] = normalize_mac(device["mac"]) + polling = config.get("polling", {}) + detector = ArrivalDetector( + int(polling.get("absent_after", 3)), int(polling.get("present_after", 2)) + ) + interval = float(polling.get("interval_seconds", 10)) + retry = float(polling.get("error_retry_seconds", 30)) + router = _build_router(config) + try: + while not stop_requested(): + try: + present = router.is_connected(device["mac"]) + LOG.debug("%s is %s", device["name"], "connected" if present else "absent") + if detector.sample(present): + _notify(config, device) + if once: + print("connected" if present else "absent") + return 0 + time.sleep(interval) + except (ZyxelError, OSError, RuntimeError) as error: + LOG.error("poll failed: %s", error) + router.close() + if once: + return 1 + time.sleep(retry) + finally: + router.close() + return 0 diff --git a/lib/_zyxel.py b/lib/_zyxel.py new file mode 100644 index 0000000..317cad4 --- /dev/null +++ b/lib/_zyxel.py @@ -0,0 +1,184 @@ +"""Small client for the encrypted Zyxel OPAL/DAL web API.""" + +from __future__ import annotations + +import base64 +import json +import os +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlsplit + +import requests +from Crypto.Cipher import AES, PKCS1_v1_5 +from Crypto.PublicKey import RSA +from Crypto.Util.Padding import pad, unpad + + +class ZyxelError(RuntimeError): + """The router rejected a request or returned an unexpected response.""" + + +def normalize_mac(value: str) -> str: + compact = "".join(character for character in value if character.isalnum()).lower() + if len(compact) != 12 or any(character not in "0123456789abcdef" for character in compact): + raise ValueError(f"invalid MAC address: {value!r}") + return ":".join(compact[index : index + 2] for index in range(0, 12, 2)) + + +@dataclass(frozen=True) +class LanHost: + mac: str + active: bool + hostname: str = "" + ip_address: str = "" + + +class ZyxelRouter: + """Read-only client for Zyxel routers using the OPAL DAL API.""" + + def __init__( + self, + host: str, + username: str, + password: str, + *, + timeout: float = 10, + verify_tls: bool = True, + hosts_oid: str = "lanhosts", + ) -> None: + if "://" not in host: + host = f"http://{host}" + parsed = urlsplit(host) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("zyxel.host must be a hostname/IP, or an http(s) URL") + self.url = host.rstrip("/") + self.username = username + self.password = password + self.timeout = timeout + self.verify_tls = verify_tls + self.hosts_oid = hosts_oid + self.session = requests.Session() + self._aes_key: bytes | None = None + self._session_key: str | None = None + + def __enter__(self) -> "ZyxelRouter": + self.login() + return self + + def __exit__(self, *_: object) -> None: + self.close() + + def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response: + kwargs.setdefault("timeout", self.timeout) + kwargs.setdefault("verify", self.verify_tls) + try: + response = self.session.request(method, f"{self.url}{path}", **kwargs) + response.raise_for_status() + return response + except requests.RequestException as error: + raise ZyxelError(f"Zyxel request failed: {error}") from error + + @staticmethod + def _json(response: requests.Response) -> dict[str, Any]: + try: + value = response.json() + except (requests.JSONDecodeError, ValueError) as error: + raise ZyxelError("Zyxel returned a non-JSON response") from error + if not isinstance(value, dict): + raise ZyxelError("Zyxel returned an unexpected JSON response") + return value + + def login(self) -> None: + self._request("GET", "/GetInfoNoLogin") + public_key = self._json(self._request("GET", "/getRSAPublickKey")).get("RSAPublicKey") + if not public_key: + raise ZyxelError("Zyxel did not return an RSA public key") + + self._aes_key = os.urandom(32) + iv = os.urandom(32) + login = { + "Input_Account": self.username, + "Input_Passwd": base64.b64encode(self.password.encode()).decode(), + "currLang": "en", + "RememberPassword": 0, + } + cipher = AES.new(self._aes_key, AES.MODE_CBC, iv[:16]) + content = cipher.encrypt(pad(json.dumps(login, separators=(",", ":")).encode(), 16)) + rsa = PKCS1_v1_5.new(RSA.import_key(public_key.encode())) + encrypted_key = rsa.encrypt(base64.b64encode(self._aes_key)) + payload = { + "content": base64.b64encode(content).decode(), + "key": base64.b64encode(encrypted_key).decode(), + "iv": base64.b64encode(iv).decode(), + } + result = self._decrypt(self._json(self._request("POST", "/UserLogin", json=payload))) + if result.get("result") != "ZCFG_SUCCESS" or not result.get("sessionkey"): + raise ZyxelError(f"Zyxel login failed: {result.get('result', 'unknown error')}") + self._session_key = str(result["sessionkey"]) + + def _decrypt(self, envelope: dict[str, Any]) -> dict[str, Any]: + if "content" not in envelope or "iv" not in envelope: + raise ZyxelError(str(envelope.get("result") or "unencrypted Zyxel response")) + if self._aes_key is None: + raise ZyxelError("not logged in") + try: + iv = base64.b64decode(envelope["iv"])[:16] + encrypted = base64.b64decode(envelope["content"]) + clear = unpad(AES.new(self._aes_key, AES.MODE_CBC, iv).decrypt(encrypted), 16) + result = json.loads(clear) + except (ValueError, KeyError, json.JSONDecodeError) as error: + raise ZyxelError("could not decrypt Zyxel response") from error + if not isinstance(result, dict): + raise ZyxelError("Zyxel returned an unexpected encrypted response") + return result + + def dal_get(self, oid: str) -> dict[str, Any]: + if not self._session_key: + self.login() + response = self._request( + "GET", "/cgi-bin/DAL", params={"oid": oid, "sessionkey": self._session_key} + ) + return self._decrypt(self._json(response)) + + def get_lan_hosts(self) -> list[LanHost]: + result = self.dal_get(self.hosts_oid) + if result.get("result") not in (None, "ZCFG_SUCCESS"): + raise ZyxelError(f"lanhosts query failed: {result.get('result')}") + objects = result.get("Object", []) + if isinstance(objects, dict): + objects = [objects] + hosts: list[LanHost] = [] + for item in objects: + if not isinstance(item, dict): + continue + raw_mac = next((item.get(key) for key in ("PhysAddress", "physAddress", "MACAddr", "MacAddress") if item.get(key)), None) + if not raw_mac: + continue + try: + mac = normalize_mac(str(raw_mac)) + except ValueError: + continue + raw_active = next((item.get(key) for key in ("Active", "active", "Enable", "enable") if key in item), True) + active = raw_active if isinstance(raw_active, bool) else str(raw_active).lower() in {"1", "true", "yes", "active", "enabled"} + hosts.append(LanHost( + mac=mac, + active=active, + hostname=str(item.get("HostName") or item.get("hostName") or ""), + ip_address=str(item.get("IPAddress") or item.get("ipAddress") or item.get("IPAddr") or ""), + )) + return hosts + + def is_connected(self, mac: str) -> bool: + wanted = normalize_mac(mac) + return any(host.mac == wanted and host.active for host in self.get_lan_hosts()) + + def close(self) -> None: + if self._session_key: + try: + self._request("GET", "/cgi-bin/UserLogout", params={"sessionkey": self._session_key}) + except ZyxelError: + pass + self._session_key = None + self._aes_key = None + self.session.close() diff --git a/notify.py b/notify.py new file mode 100644 index 0000000..8c18172 --- /dev/null +++ b/notify.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 + +"""Send a text notification to the Home Alert ntfy topic.""" + +import argparse +import os +import sys +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +DEFAULT_TOPIC_URL = ( + "https://ntfy.sh/VK20_Home_Alert_______4671938674123049873459" +) + + +def send_notification( + message, + topic_url=None, + timeout=10, +): + """Send *message* to ntfy and return the server response body.""" + + if not isinstance(message, str): + raise TypeError("message must be a string") + + if not message: + raise ValueError("message must not be empty") + + url = topic_url or os.environ.get( + "NTFY_TOPIC_URL", + DEFAULT_TOPIC_URL, + ) + request = Request( + url, + data=message.encode("utf-8"), + headers={"Content-Type": "text/plain; charset=utf-8"}, + method="POST", + ) + + with urlopen(request, timeout=timeout) as response: + return response.read().decode("utf-8") + + +def parse_arguments(): + """Parse command-line arguments.""" + + parser = argparse.ArgumentParser( + description="Send a text message to the Home Alert ntfy topic.", + ) + parser.add_argument( + "message", + nargs="+", + help="notification text", + ) + parser.add_argument( + "--topic-url", + help="override the ntfy topic URL", + ) + parser.add_argument( + "--timeout", + type=float, + default=10, + help="request timeout in seconds (default: 10)", + ) + return parser.parse_args() + + +def main(): + args = parse_arguments() + + try: + response = send_notification( + " ".join(args.message), + topic_url=args.topic_url, + timeout=args.timeout, + ) + print(response) + return 0 + + except HTTPError as exc: + print( + f"ntfy returned HTTP {exc.code}: {exc.reason}", + file=sys.stderr, + ) + except URLError as exc: + print(f"Could not reach ntfy: {exc.reason}", file=sys.stderr) + except (TypeError, ValueError) as exc: + print(f"Invalid notification: {exc}", file=sys.stderr) + + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..aa69174 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +pycryptodome>=3.20,<4 +PyYAML>=6.0,<7 +requests>=2.31,<3 diff --git a/service.py b/service.py new file mode 100644 index 0000000..1ddeb09 --- /dev/null +++ b/service.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Entry point and orchestrator for enabled home-control functions.""" + +from __future__ import annotations + +import argparse +import logging +import signal +from pathlib import Path + +import yaml + +from lib import _arrival_detection + + +def load_config(path: Path) -> dict: + with path.open(encoding="utf-8") as stream: + config = yaml.safe_load(stream) + if not isinstance(config, dict): + raise ValueError("configuration must be a YAML mapping") + return config + + +def run(config: dict, once: bool = False) -> int: + stopped = False + + def stop(*_: object) -> None: + nonlocal stopped + stopped = True + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + + arrival_config = config.get("arrival_detection") + if not isinstance(arrival_config, dict): + raise ValueError("arrival_detection configuration is required") + if not arrival_config.get("enabled", True): + logging.getLogger("home_control").info("arrival detection is disabled") + return 0 + return _arrival_detection.run(arrival_config, lambda: stopped, once=once) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--config", type=Path, default=Path(__file__).with_name("service.yaml")) + parser.add_argument("--once", action="store_true", help="check once and print connected/absent") + args = parser.parse_args() + config = load_config(args.config) + arrival = config.get("arrival_detection", {}) + logging_config = arrival.get("logging", {}) if isinstance(arrival, dict) else {} + logging.basicConfig( + level=getattr(logging, str(logging_config.get("level", "INFO")).upper()), + format="%(asctime)s %(levelname)s %(message)s", + ) + return run(config, args.once) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/service.yaml b/service.yaml new file mode 100644 index 0000000..ed6678e --- /dev/null +++ b/service.yaml @@ -0,0 +1,33 @@ +arrival_detection: + enabled: true + + zyxel: + host: http://192.168.178.2 + username: admin + password: KYKPXWJ8 + timeout: 10 + verify_tls: true + hosts_oid: lanhosts + + device: + name: Ignace + # Use the iPhone's Wi-Fi Address shown for your home SSID. If Private Wi-Fi + # Address is enabled, this is the per-network private address, not hardware MAC. + mac: "ce:68:53:a8:fc:07" + + polling: + interval_seconds: 5 + present_after: 2 + absent_after: 4 + error_retry_seconds: 30 + + notification: + enabled: true + # Available placeholders: {name}, {mac}. + message: "{name} arrived home" + # Leave blank to use the default ntfy topic configured in notify.py. + topic_url: "" + timeout: 10 + + logging: + level: INFO diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..12734cb --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for home_control.""" diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..9f2b4e0 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,57 @@ +import unittest +from unittest.mock import patch + +from lib._zyxel import normalize_mac +from lib._arrival_detection import ArrivalDetector, _notify + + +class ArrivalDetectorTests(unittest.TestCase): + def test_starting_present_does_not_emit_arrival(self): + detector = ArrivalDetector(absent_after=2, present_after=2) + self.assertFalse(detector.sample(True)) + self.assertFalse(detector.sample(True)) + + def test_emits_only_after_absence_and_debounced_presence(self): + detector = ArrivalDetector(absent_after=2, present_after=2) + for value in (False, False, True): + self.assertFalse(detector.sample(value)) + self.assertTrue(detector.sample(True)) + self.assertFalse(detector.sample(True)) + + def test_single_missed_poll_does_not_mark_absent(self): + detector = ArrivalDetector(absent_after=2, present_after=1) + detector.sample(True) + self.assertFalse(detector.sample(False)) + self.assertFalse(detector.sample(True)) + + +class MacTests(unittest.TestCase): + def test_normalizes_common_formats(self): + self.assertEqual(normalize_mac("AA-BB-CC-DD-EE-FF"), "aa:bb:cc:dd:ee:ff") + + def test_rejects_invalid_mac(self): + with self.assertRaises(ValueError): + normalize_mac("not-a-mac") + + +class NotificationTests(unittest.TestCase): + @patch("lib._arrival_detection.send_notification") + def test_uses_notify_module(self, send_notification): + config = { + "notification": { + "enabled": True, + "message": "Welcome home, {name} ({mac})", + "topic_url": "https://ntfy.example/home", + "timeout": 4, + } + } + _notify(config, {"name": "Ignace", "mac": "aa:bb:cc:dd:ee:ff"}) + send_notification.assert_called_once_with( + "Welcome home, Ignace (aa:bb:cc:dd:ee:ff)", + topic_url="https://ntfy.example/home", + timeout=4.0, + ) + + +if __name__ == "__main__": + unittest.main()