feat(edge): CloudClient for config poll + heartbeat (HMAC-signed)

edge/alert.py

@@ -6,7 +6,9 @@ import json
 from edge.events import Event
 
 
-def sign(secret: str, nonce: str, ts: str, body: bytes) -> str:
+def sign(secret: str, nonce: str, ts: str, body: str | bytes) -> str:
+    if isinstance(body, str):
+        body = body.encode()
     msg = nonce.encode() + ts.encode() + body
     return hmac.new(secret.encode(), msg, hashlib.sha256).hexdigest()
 

edge/cloudclient.py

@@ -0,0 +1,50 @@
+import json
+
+from edge.alert import sign
+
+
+class CloudClient:
+    def __init__(self, base_url: str, device_id: str, secret: str,
+                 nonce_fn, now_fn, http_get=None, http_post=None, timeout: float = 5.0):
+        self.base_url = base_url.rstrip("/")
+        self.device_id = device_id
+        self.secret = secret
+        self.nonce_fn = nonce_fn
+        self.now_fn = now_fn
+        self.timeout = timeout
+        if http_get is None or http_post is None:
+            import requests
+            http_get = http_get or requests.get
+            http_post = http_post or requests.post
+        self._get = http_get
+        self._post = http_post
+
+    def _headers(self, body: str) -> dict:
+        nonce = self.nonce_fn()
+        ts = str(self.now_fn())
+        return {"X-Nonce": nonce, "X-Timestamp": ts,
+                "X-Signature": sign(self.secret, nonce, ts, body)}
+
+    def fetch_config(self) -> dict | None:
+        url = f"{self.base_url}/api/config?device_id={self.device_id}"
+        try:
+            resp = self._get(url, headers=self._headers(""), timeout=self.timeout)
+        except Exception:
+            return None
+        if 200 <= resp.status_code < 300:
+            try:
+                return resp.json()
+            except Exception:
+                return None
+        return None
+
+    def send_heartbeat(self, metrics: dict) -> bool:
+        payload = {"device_id": self.device_id, **metrics}
+        body = json.dumps(payload, sort_keys=True)
+        headers = {"Content-Type": "application/json", **self._headers(body)}
+        try:
+            resp = self._post(f"{self.base_url}/api/heartbeat", data=body,
+                              headers=headers, timeout=self.timeout)
+        except Exception:
+            return False
+        return 200 <= resp.status_code < 300

tests/test_cloudclient.py

@@ -0,0 +1,32 @@
+import json
+from edge.cloudclient import CloudClient
+from edge.alert import sign
+
+class _Resp:
+    def __init__(self, code, body=""): self.status_code = code; self.text = body
+    def json(self): return json.loads(self.text)
+
+def test_fetch_config_signs_and_parses():
+    captured = {}
+    def fake_get(url, headers=None, timeout=None):
+        captured["url"] = url; captured["headers"] = headers
+        return _Resp(200, json.dumps({"version": 3, "zones": [], "thresholds": {}}))
+    c = CloudClient("https://x", "pi-1", "sek", nonce_fn=lambda: "n1", now_fn=lambda: 1000.0, http_get=fake_get)
+    cfg = c.fetch_config()
+    assert cfg["version"] == 3
+    assert "device_id=pi-1" in captured["url"]
+    assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", "")
+
+def test_fetch_config_none_on_error():
+    c = CloudClient("https://x", "pi-1", "sek", lambda: "n", lambda: 1.0, http_get=lambda *a, **k: _Resp(401))
+    assert c.fetch_config() is None
+
+def test_send_heartbeat_signs_body():
+    captured = {}
+    def fake_post(url, data=None, headers=None, timeout=None):
+        captured["data"] = data; captured["headers"] = headers; return _Resp(200)
+    c = CloudClient("https://x", "pi-1", "sek", lambda: "n1", lambda: 1000.0, http_post=fake_post)
+    assert c.send_heartbeat({"fps": 12.0}) is True
+    body = captured["data"]
+    assert json.loads(body)["device_id"] == "pi-1"
+    assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", body)