server.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import mimetypes
import os
import socketserver
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

from lib.auth import check_bearer, ip_allowed, load_allow_cidrs, token_configured
from lib.netbot import (
 MOBILE_SHORTCUTS,
 bridge_health,
 load_catalog,
 local_get,
 mobile_shortcut,
 netbot_get,
 netbot_invoke,
)
from lib.runner import (
 install_tool,
 list_scripts,
 restart_api_services,
 run_script,
 run_system_probe,
 sync_api_files,
)

PORT = int(os.environ.get("PORT", "9378"))
BIND_ADDRESS = os.environ.get("BIND_ADDRESS", "").strip()
ALLOW_CIDRS = load_allow_cidrs()
STATIC_DIR = Path(__file__).resolve().parent / "static"


def json_response(handler: BaseHTTPRequestHandler, status: int, body: Any) -> None:
 payload = json.dumps(body).encode("utf-8")
 handler.send_response(status)
 handler.send_header("Content-Type", "application/json; charset=utf-8")
 handler.send_header("Cache-Control", "no-store")
 handler.send_header("Content-Length", str(len(payload)))
 handler.end_headers()
 handler.wfile.write(payload)


class FirewallaApiHandler(BaseHTTPRequestHandler):
 server_version = "array-firewalla-api/2.0"

 def log_message(self, fmt: str, *args: object) -> None:
 print(f"[firewalla-api] {self.address_string()} {fmt % args}")

 def _client_ip(self) -> str:
 return self.client_address[0]

 def _reject_unless_allowed(self) -> bool:
 ip = self._client_ip()
 if not ip_allowed(ip, ALLOW_CIDRS):
 json_response(
 self,
 403,
 {"error": "forbidden", "detail": "LAN clients only", "clientIp": ip},
 )
 return False
 return True

 def _reject_unless_authed(self) -> bool:
 if not check_bearer(self.headers.get("Authorization")):
 json_response(self, 401, {"error": "unauthorized"})
 return False
 return True

 def _read_json(self) -> dict[str, Any]:
 length = int(self.headers.get("Content-Length", "0") or 0)
 raw = self.rfile.read(length) if length else b""
 if not raw:
 return {}
 return json.loads(raw.decode("utf-8"))

 def _parsed_path(self) -> tuple[str, dict[str, list[str]]]:
 parsed = urllib.parse.urlparse(self.path)
 return parsed.path, urllib.parse.parse_qs(parsed.query)

 def _handle_health(self) -> None:
 bridge: dict[str, Any] = {"ok": False}
 try:
 bridge = bridge_health()
 except Exception as exc: # noqa: BLE001
 bridge = {"ok": False, "error": str(exc)}
 json_response(
 self,
 200,
 {
 "ok": True,
 "service": "array-firewalla-api",
 "version": 2,
 "bindAddress": BIND_ADDRESS,
 "port": PORT,
 "allowCidrs": list(ALLOW_CIDRS),
 "tokenRequired": token_configured(),
 "toolsDir": os.environ.get("FIREWALLA_TOOLS_DIR", "/home/pi/gaming-tools"),
 "netbotBridge": bridge,
 "mobileParity": "POST /api/v1/netbot with {mtype,data} — same netbot path as mobile app",
 },
 )

 def _handle_catalog(self) -> None:
 json_response(
 self,
 200,
 {
 "netbotItems": load_catalog(),
 "mobileShortcuts": sorted(MOBILE_SHORTCUTS.keys()),
 "usage": {
 "generic": "POST /api/v1/netbot {mtype: get|cmd|set, data: {item, value?}}",
 "shortcut": "GET /api/v1/mobile/{shortcut}",
 "localHost": "GET /api/v1/local/host/all (production FireAPI local)",
 },
 },
 )

 def _handle_mobile_shortcut(self, name: str) -> None:
 result = mobile_shortcut(name)
 json_response(self, 200, {"ok": True, "shortcut": name, "result": result})

 def _handle_netbot_get_item(self, item: str, query: dict[str, list[str]]) -> None:
 value_raw = (query.get("value") or [None])[0]
 value = json.loads(value_raw) if value_raw else None
 result = netbot_get(item, value)
 json_response(self, 200, {"ok": True, "result": result})

 def _handle_netbot_post(self, body: dict[str, Any]) -> None:
 mtype = body.get("mtype")
 data = body.get("data") or {}
 target = body.get("target")
 if not mtype:
 json_response(self, 400, {"error": "mtype required (get, cmd, set, ...)"})
 return
 result = netbot_invoke(str(mtype), dict(data), target=target)
 json_response(self, 200, {"ok": True, "result": result})

 def _handle_local(self, subpath: str) -> None:
 result = local_get(subpath)
 json_response(self, 200, result)

 def _serve_static(self, path: str) -> bool:
 rel = path.lstrip("/")
 if rel in ("", "index.html"):
 rel = "index.html"
 elif rel.startswith("static/"):
 rel = rel[len("static/") :]
 if not rel or ".." in rel or rel.startswith("/"):
 return False
 file_path = (STATIC_DIR / rel).resolve()
 if not str(file_path).startswith(str(STATIC_DIR.resolve())):
 return False
 if not file_path.is_file():
 return False
 content = file_path.read_bytes()
 mime, _ = mimetypes.guess_type(str(file_path))
 self.send_response(200)
 self.send_header("Content-Type", mime or "application/octet-stream")
 self.send_header("Cache-Control", "no-cache")
 self.send_header("Content-Length", str(len(content)))
 self.end_headers()
 self.wfile.write(content)
 return True

 def do_GET(self) -> None:
 if not self._reject_unless_allowed():
 return

 path, query = self._parsed_path()

 if path == "/api/health":
 self._handle_health()
 return

 if path == "/api/v1/catalog":
 if not self._reject_unless_authed():
 return
 self._handle_catalog()
 return

 if path == "/api/v1/mobile":
 if not self._reject_unless_authed():
 return
 json_response(self, 200, {"shortcuts": MOBILE_SHORTCUTS})
 return

 if path.startswith("/api/v1/mobile/"):
 if not self._reject_unless_authed():
 return
 name = path[len("/api/v1/mobile/") :]
 try:
 self._handle_mobile_shortcut(name)
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path.startswith("/api/v1/netbot/"):
 if not self._reject_unless_authed():
 return
 item = path[len("/api/v1/netbot/") :]
 try:
 self._handle_netbot_get_item(item, query)
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path.startswith("/api/v1/local/"):
 if not self._reject_unless_authed():
 return
 subpath = path[len("/api/v1/local/") :]
 try:
 self._handle_local(subpath)
 except Exception as exc: # noqa: BLE001
 json_response(self, 502, {"error": str(exc)})
 return

 if path == "/api/v1/scripts":
 if not self._reject_unless_authed():
 return
 json_response(self, 200, {"scripts": list_scripts()})
 return

 if path == "/api/v1/system":
 if not self._reject_unless_authed():
 return
 try:
 json_response(self, 200, run_system_probe())
 except Exception as exc: # noqa: BLE001
 json_response(self, 500, {"error": str(exc)})
 return

 if path == "/" or path.startswith("/static/"):
 if self._serve_static(path):
 return

 json_response(self, 404, {"error": "not found"})

 def do_POST(self) -> None:
 if not self._reject_unless_allowed():
 return

 path, _query = self._parsed_path()

 if path == "/api/v1/netbot":
 if not self._reject_unless_authed():
 return
 try:
 self._handle_netbot_post(self._read_json())
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path == "/api/v1/run":
 if not self._reject_unless_authed():
 return
 try:
 body = self._read_json()
 result = run_script(
 body.get("script", ""),
 list(body.get("args") or []),
 sudo=bool(body.get("sudo")),
 payload=body.get("payload"),
 )
 json_response(self, 200, result)
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path == "/api/v1/tools/update":
 if not self._reject_unless_authed():
 return
 try:
 body = self._read_json()
 mode = int(body.get("mode", "755"), 8)
 result = install_tool(
 body.get("name", ""),
 body.get("content", ""),
 mode=mode,
 )
 json_response(self, 200, result)
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path == "/api/v1/admin/sync":
 if not self._reject_unless_authed():
 return
 try:
 body = self._read_json()
 result = sync_api_files(list(body.get("files") or []))
 json_response(self, 200, result)
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 if path == "/api/v1/admin/restart":
 if not self._reject_unless_authed():
 return
 try:
 json_response(self, 200, restart_api_services(detach=True))
 except Exception as exc: # noqa: BLE001
 json_response(self, 400, {"error": str(exc)})
 return

 json_response(self, 404, {"error": "not found"})


def main() -> None:
 if not BIND_ADDRESS:
 raise SystemExit("BIND_ADDRESS is required (set in /etc/default/firewalla-api)")
 if not ALLOW_CIDRS:
 raise SystemExit("FIREWALLA_API_ALLOW_CIDRS is required (set in /etc/default/firewalla-api)")
 with ThreadingHTTPServer((BIND_ADDRESS, PORT), FirewallaApiHandler) as httpd:
 print(f"array-firewalla-api listening on http://{BIND_ADDRESS}:{PORT}")
 print(f" allow CIDRs: {', '.join(ALLOW_CIDRS)}")
 print(f" token required: {token_configured()}")
 httpd.serve_forever()


if __name__ == "__main__":
 main()