feat: fde-lock endpoint for home assistant, init_on_free

fde-lock/api.py

@@ -0,0 +1,71 @@
+# HTTP trigger for fde-lock, called by Home Assistant.
+#   POST /lock    lock Nethergarde and ashenvale_crypt (202, runs in the background)
+#   POST /check   dry run, changes nothing
+#   GET  /status  result of the last run
+# Needs "Authorization: Bearer <token>" and a source address from FDE_LOCK_ALLOWED.
+# This process has no privileges: it only touches a trigger file that a root path unit watches.
+
+import hmac
+import json
+import os
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+TOKEN = (Path(os.environ["CREDENTIALS_DIRECTORY"]) / "token").read_text().strip().encode()
+ALLOWED = set(os.environ["FDE_LOCK_ALLOWED"].split())
+TRIGGER_DIR = Path("/run/fde-lock-api")
+STATUS = Path("/run/fde-lock/status")
+
+
+def read_status():
+    if not STATUS.exists():
+        return {"state": "never-run"}
+    result = {"steps": []}
+    for line in STATUS.read_text().splitlines():
+        parts = line.split("\t")
+        if parts[0] == "step" and len(parts) >= 3:
+            result["steps"].append({"name": parts[1], "result": parts[2], "detail": "\t".join(parts[3:])})
+        elif len(parts) == 2:
+            result[parts[0]] = parts[1]
+    return result
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "fde-lock"
+
+    def authorized(self):
+        if self.client_address[0] not in ALLOWED:
+            return False
+        given = self.headers.get("Authorization", "").encode()
+        return hmac.compare_digest(given, b"Bearer " + TOKEN)
+
+    def reply(self, code, body):
+        data = json.dumps(body).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def do_POST(self):
+        if not self.authorized():
+            return self.reply(403, {"error": "forbidden"})
+        action = {"/lock": "lock", "/check": "check"}.get(self.path)
+        if action is None:
+            return self.reply(404, {"error": "not found"})
+        (TRIGGER_DIR / action).write_text(f"{time.time()}\n")
+        self.reply(202, {"accepted": action})
+
+    def do_GET(self):
+        if not self.authorized():
+            return self.reply(403, {"error": "forbidden"})
+        if self.path != "/status":
+            return self.reply(404, {"error": "not found"})
+        self.reply(200, read_status())
+
+    def log_message(self, fmt, *args):
+        print(f"{self.client_address[0]} {fmt % args}", flush=True)
+
+
+ThreadingHTTPServer((os.environ["FDE_LOCK_LISTEN"], int(os.environ["FDE_LOCK_PORT"])), Handler).serve_forever()