zenai-kpc-python/status_webhook.py

#!/usr/bin/env python3
"""
Standalone status webhook (stdlib only, HTTP + HTTPS).

Receives GET /?lokasi=IN|OUT&status=ON|OFF.

State stored in JSON as IN | OUT | OFF.

Success (200) only when a valid transition applies; otherwise 400 and JSON unchanged:
  - lokasi=IN,  status=ON  + current OFF -> IN
  - lokasi=OUT, status=ON  + current OFF -> OUT
  - lokasi=IN,  status=OFF + current IN  -> OFF
  - lokasi=OUT, status=OFF + current OUT -> OFF

Fixed ports: HTTP 8002, HTTPS 8443 (TLS cannot share a port with plain HTTP).
"""

from __future__ import annotations

import json
import os
import ssl
import subprocess
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

# Fixed ports — always the same (do not override via env).
HTTP_PORT = 8002
HTTPS_PORT = 8443
BASE_DIR = Path(os.path.dirname(os.path.abspath(__file__)))
STATE_FILE = Path(
    os.getenv("STATUS_WEBHOOK_STATE", str(BASE_DIR / "status_webhook_state.json"))
)
CERT_FILE = Path(os.getenv("STATUS_WEBHOOK_CERT", str(BASE_DIR / "status_webhook_cert.pem")))
KEY_FILE = Path(os.getenv("STATUS_WEBHOOK_KEY", str(BASE_DIR / "status_webhook_key.pem")))
ALLOWED_LOKASI = frozenset({"IN", "OUT"})
ALLOWED_STATUS = frozenset({"ON", "OFF"})
ALLOWED_STORED = frozenset({"IN", "OUT", "OFF"})

_lock = threading.Lock()


def normalize_lokasi(value: str | None) -> str | None:
    if value is None:
        return None
    value = value.strip().upper()
    return value if value in ALLOWED_LOKASI else None


def normalize_request_status(value: str | None) -> str | None:
    if value is None:
        return None
    value = value.strip().upper()
    return value if value in ALLOWED_STATUS else None


def normalize_stored(value: str | None) -> str | None:
    if value is None:
        return None
    value = str(value).strip().upper()
    return value if value in ALLOWED_STORED else None


def load_status() -> str | None:
    try:
        data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return None
    except (OSError, json.JSONDecodeError) as exc:
        print(f"status_webhook: failed to read state from {STATE_FILE}: {exc}")
        return None
    return normalize_stored(data.get("status"))


def is_run_status(status: str | None = None) -> bool:
    """True when counter should run (IN or OUT)."""
    if status is None:
        status = load_status()
    return normalize_stored(status) in ("IN", "OUT")


def save_status(status: str) -> None:
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    tmp = STATE_FILE.with_suffix(".tmp")
    payload = {"status": status}
    tmp.write_text(json.dumps(payload) + "\n", encoding="utf-8")
    tmp.replace(STATE_FILE)


def apply_transition(
    lokasi: str, status: str, current: str | None
) -> tuple[str | None, bool]:
    """Return (new_status, ok). ok True only for a valid transition that stores."""
    # Treat missing/empty state as OFF so first ON can start a session.
    effective = current if current is not None else "OFF"

    if status == "ON":
        if effective == "OFF":
            return lokasi, True
        return None, False

    # status == "OFF"
    if effective == lokasi:
        return "OFF", True
    return None, False


def ensure_tls_certs() -> None:
    if CERT_FILE.is_file() and KEY_FILE.is_file():
        return
    CERT_FILE.parent.mkdir(parents=True, exist_ok=True)
    print(
        f"status_webhook: generating self-signed TLS cert at {CERT_FILE} "
        f"(set STATUS_WEBHOOK_CERT / STATUS_WEBHOOK_KEY to use your own)"
    )
    subprocess.run(
        [
            "openssl",
            "req",
            "-x509",
            "-newkey",
            "rsa:2048",
            "-keyout",
            str(KEY_FILE),
            "-out",
            str(CERT_FILE),
            "-days",
            "365",
            "-nodes",
            "-subj",
            "/CN=status-webhook",
        ],
        check=True,
        capture_output=True,
        text=True,
    )


class StatusHandler(BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args) -> None:
        print(f"status_webhook: {self.address_string()} - {fmt % args}")

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path not in ("/", ""):
            self._send_text(404, "Not Found")
            return

        params = parse_qs(parsed.query)
        lokasi_values = params.get("lokasi", [])
        status_values = params.get("status", [])
        if (
            not lokasi_values
            or lokasi_values[0] == ""
            or not status_values
            or status_values[0] == ""
        ):
            self._send_text(400, "fail")
            return

        lokasi = normalize_lokasi(lokasi_values[0])
        status = normalize_request_status(status_values[0])
        if lokasi is None or status is None:
            self._send_text(400, "fail")
            return

        with _lock:
            current = load_status()
            new_status, ok = apply_transition(lokasi, status, current)
            if not ok or new_status is None:
                self._send_text(400, "fail")
                return
            save_status(new_status)

        self._send_text(200, "success")

    def _send_text(self, code: int, body: str) -> None:
        raw = body.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)


def make_http_server(port: int = HTTP_PORT) -> ThreadingHTTPServer:
    return ThreadingHTTPServer(("0.0.0.0", port), StatusHandler)


def make_https_server(port: int = HTTPS_PORT) -> ThreadingHTTPServer:
    ensure_tls_certs()
    server = ThreadingHTTPServer(("0.0.0.0", port), StatusHandler)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(certfile=str(CERT_FILE), keyfile=str(KEY_FILE))
    server.socket = ctx.wrap_socket(server.socket, server_side=True)
    return server


def _try_make_server(kind: str, factory) -> ThreadingHTTPServer | None:
    try:
        return factory()
    except OSError as exc:
        print(f"status_webhook: {kind} port already in use ({exc}); skipping bind")
        return None


def start_background() -> tuple[ThreadingHTTPServer | None, ThreadingHTTPServer | None]:
    """Start HTTP + HTTPS servers on daemon threads. Safe if ports are taken."""
    http_server = _try_make_server("HTTP", make_http_server)
    https_server = _try_make_server("HTTPS", make_https_server)

    if http_server is not None:
        threading.Thread(
            target=http_server.serve_forever,
            name="status-webhook-http",
            daemon=True,
        ).start()
        print(
            f"status_webhook HTTP  http://0.0.0.0:{HTTP_PORT}/"
            f"?lokasi=IN|OUT&status=ON|OFF"
        )

    if https_server is not None:
        threading.Thread(
            target=https_server.serve_forever,
            name="status-webhook-https",
            daemon=True,
        ).start()
        print(
            f"status_webhook HTTPS https://0.0.0.0:{HTTPS_PORT}/"
            f"?lokasi=IN|OUT&status=ON|OFF"
        )

    print(f"status_webhook state file: {STATE_FILE}")
    return http_server, https_server


def stop_servers(
    http_server: ThreadingHTTPServer | None,
    https_server: ThreadingHTTPServer | None,
) -> None:
    for server in (http_server, https_server):
        if server is None:
            continue
        try:
            server.shutdown()
            server.server_close()
        except Exception as exc:
            print(f"status_webhook: shutdown error: {exc}")


def main() -> None:
    http_server, https_server = start_background()
    if http_server is None and https_server is None:
        raise SystemExit("status_webhook: neither HTTP nor HTTPS could bind")

    try:
        # serve_forever already running on threads; park main until interrupt.
        while True:
            threading.Event().wait(3600)
    except KeyboardInterrupt:
        print("\nstatus_webhook stopped")
    finally:
        stop_servers(http_server, https_server)


if __name__ == "__main__":
    main()