#!/usr/bin/env python3 """CLI helper for agent-safe UMEC Modbus gateway HTTP operations.""" from __future__ import annotations import argparse import base64 import json import os import sys import urllib.error import urllib.parse import urllib.request from http.cookiejar import CookieJar from pathlib import Path from typing import Any DEFAULT_TIMEOUT = 20 DANGEROUS_CONFIRM = { "upload-firmware": "update", "reboot": "reboot", "factory-reset": "factory-reset", } class GatewayError(RuntimeError): def __init__(self, message: str, status: int | None = None, body: str = "") -> None: super().__init__(message) self.status = status self.body = body class GatewayClient: def __init__( self, base_url: str, token: str | None = None, username: str | None = None, password: str | None = None, timeout: int = DEFAULT_TIMEOUT, ) -> None: self.base_url = base_url.rstrip("/") self.token = token self.username = username self.password = password self.timeout = timeout self.cookie_jar = CookieJar() self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cookie_jar)) def request( self, method: str, path: str, body: bytes | None = None, content_type: str | None = None, headers: dict[str, str] | None = None, auth: bool = True, timeout: int | None = None, ) -> tuple[int, str, dict[str, str]]: url = self._url(path) req_headers = dict(headers or {}) if content_type: req_headers["Content-Type"] = content_type if auth: self._apply_auth(req_headers) request = urllib.request.Request(url, data=body, headers=req_headers, method=method.upper()) try: with self.opener.open(request, timeout=timeout or self.timeout) as response: data = response.read().decode("utf-8", errors="replace") return response.status, data, dict(response.headers.items()) except urllib.error.HTTPError as exc: data = exc.read().decode("utf-8", errors="replace") raise GatewayError(f"HTTP {exc.code} for {method} {path}", exc.code, data) from exc except urllib.error.URLError as exc: raise GatewayError(f"Request failed for {method} {path}: {exc.reason}") from exc def login(self) -> dict[str, Any]: if not self.username or not self.password: raise GatewayError("login requires username and password") payload = json.dumps({"username": self.username, "password": self.password}).encode("utf-8") _, body, _ = self.request("POST", "/login", payload, "application/json", auth=False) return parse_json(body) def _apply_auth(self, headers: dict[str, str]) -> None: if self.token: headers["Authorization"] = f"Bearer {self.token}" return if self.username is not None and self.password is not None: raw = f"{self.username}:{self.password}".encode("utf-8") headers["Authorization"] = "Basic " + base64.b64encode(raw).decode("ascii") def _url(self, path: str) -> str: if path.startswith("http://") or path.startswith("https://"): return path if not path.startswith("/"): path = "/" + path return self.base_url + path def parse_json(text: str) -> dict[str, Any]: if not text.strip(): return {} try: value = json.loads(text) except json.JSONDecodeError as exc: raise GatewayError(f"Expected JSON response: {exc}") from exc if not isinstance(value, dict): raise GatewayError("Expected JSON object response") return value def load_json_arg(value: str) -> bytes: if value.startswith("@"): return Path(value[1:]).read_bytes() json.loads(value) return value.encode("utf-8") def print_result(status: int, body: str, raw: bool = False) -> None: if raw: sys.stdout.write(body) if body and not body.endswith("\n"): sys.stdout.write("\n") return try: parsed = json.loads(body) if body.strip() else {"status": "ok"} except json.JSONDecodeError: parsed = {"status": "ok", "http_status": status, "body": body} if isinstance(parsed, dict): parsed.setdefault("http_status", status) print(json.dumps(parsed, ensure_ascii=False, indent=2)) def require_confirm(command: str, actual: str | None) -> None: expected = DANGEROUS_CONFIRM[command] if actual != expected: raise GatewayError(f"{command} requires --confirm {expected}") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Agent-safe UMEC Modbus gateway HTTP tool. Prefer Bearer token auth for agents." ) parser.add_argument("--base-url", default=os.getenv("GATEWAY_BASE_URL"), help="Gateway base URL") parser.add_argument("--token", default=os.getenv("GATEWAY_AGENT_TOKEN"), help="Bearer agent token") parser.add_argument("--username", default=os.getenv("GATEWAY_USERNAME"), help="Local admin username") parser.add_argument("--password", default=os.getenv("GATEWAY_PASSWORD"), help="Local admin password") parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("login", help="Create a browser-compatible cookie session using username/password") sub.add_parser("session", help="Read current auth metadata without secrets") sub.add_parser("status", help="Read gateway runtime status") rotate = sub.add_parser("rotate-token", help="Rotate the single agent token") rotate.add_argument("--scopes", default="read,test,configure", help="Comma-separated scopes") sub.add_parser("delete-token", help="Delete the active agent token") get = sub.add_parser("get", help="GET an API path") get.add_argument("path") get.add_argument("--raw", action="store_true") post = sub.add_parser("post-json", help="POST JSON to an API path") post.add_argument("path") post.add_argument("--json", required=True, help="JSON string or @file") put = sub.add_parser("put-json", help="PUT JSON to an API path") put.add_argument("path") put.add_argument("--json", required=True, help="JSON string or @file") delete = sub.add_parser("delete", help="DELETE an API path") delete.add_argument("path") add_device = sub.add_parser("add-device", help="POST a device JSON payload to /api/modbus/device") add_device.add_argument("--json", required=True, help="JSON string or @file") upload = sub.add_parser("upload-firmware", help="Upload firmware through /update") upload.add_argument("firmware") upload.add_argument("--confirm", required=True) upload.add_argument("--timeout", type=int, default=180) reboot = sub.add_parser("reboot", help="Request gateway reboot") reboot.add_argument("--confirm", required=True) reset = sub.add_parser("factory-reset", help="Request full factory reset") reset.add_argument("--confirm", required=True) return parser def make_client(args: argparse.Namespace) -> GatewayClient: if not args.base_url: raise GatewayError("Set --base-url or GATEWAY_BASE_URL") return GatewayClient(args.base_url, args.token, args.username, args.password, args.timeout) def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: client = make_client(args) if args.command == "login": print(json.dumps(client.login(), ensure_ascii=False, indent=2)) return 0 if args.command == "session": status, body, _ = client.request("GET", "/api/auth/session") print_result(status, body) return 0 if args.command == "status": status, body, _ = client.request("GET", "/api/status") print_result(status, body) return 0 if args.command == "rotate-token": scopes = [item.strip() for item in args.scopes.split(",") if item.strip()] payload = json.dumps({"scopes": scopes}).encode("utf-8") status, body, _ = client.request("POST", "/api/auth/agent-token", payload, "application/json") print_result(status, body) return 0 if args.command == "delete-token": status, body, _ = client.request("DELETE", "/api/auth/agent-token") print_result(status, body) return 0 if args.command == "get": status, body, _ = client.request("GET", args.path) print_result(status, body, args.raw) return 0 if args.command == "post-json": status, body, _ = client.request("POST", args.path, load_json_arg(args.json), "application/json") print_result(status, body) return 0 if args.command == "put-json": status, body, _ = client.request("PUT", args.path, load_json_arg(args.json), "application/json") print_result(status, body) return 0 if args.command == "delete": status, body, _ = client.request("DELETE", args.path) print_result(status, body) return 0 if args.command == "add-device": status, body, _ = client.request("POST", "/api/modbus/device", load_json_arg(args.json), "application/json") print_result(status, body) return 0 if args.command == "upload-firmware": require_confirm(args.command, args.confirm) body = Path(args.firmware).read_bytes() status, text, _ = client.request( "POST", "/update", body, "application/octet-stream", {"X-Gateway-Confirm": "update"}, timeout=args.timeout, ) print_result(status, text) return 0 if args.command == "reboot": require_confirm(args.command, args.confirm) status, body, _ = client.request("POST", "/reboot", headers={"X-Gateway-Confirm": "reboot"}) print_result(status, body) return 0 if args.command == "factory-reset": require_confirm(args.command, args.confirm) status, body, _ = client.request( "POST", "/factory-reset", headers={"X-Gateway-Confirm": "factory-reset"} ) print_result(status, body) return 0 except GatewayError as exc: error = {"status": "error", "message": str(exc)} if exc.status is not None: error["http_status"] = exc.status if exc.body: error["body"] = exc.body print(json.dumps(error, ensure_ascii=False, indent=2), file=sys.stderr) return 2 return 1 if __name__ == "__main__": raise SystemExit(main())