stormwind-nixfiles/fde-lock/api.py

# HTTP trigger for fde-lock, called by Home Assistant.
#   POST /lock    lock Nethergarde, ashenvale_crypt and duskwood_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()