72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
# 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()
|