diff --git a/script-api.py b/script-api.py index 8582f64..31e980e 100755 --- a/script-api.py +++ b/script-api.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 + import html import json import re @@ -7,10 +8,12 @@ import subprocess import sys import time import uuid + from pathlib import Path from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs + CONFIG_PATH = Path(sys.argv[1] if len(sys.argv) > 1 else "./commands.json") WEB_CSRF_TOKEN = secrets.token_urlsafe(32) @@ -24,476 +27,884 @@ def load_config(): cfg.setdefault("max_body_bytes", 65536) cfg.setdefault("commands", {}) + if not cfg.get("api_key") or len(cfg["api_key"]) < 16: + raise RuntimeError( + "Global api_key missing or shorter than 16 characters" + ) + if not isinstance(cfg["commands"], dict) or not cfg["commands"]: raise RuntimeError("No commands configured") validate_config(cfg) + return cfg def validate_config(cfg): + for command_id, command in cfg["commands"].items(): - if not re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", command_id): - raise RuntimeError(f"Invalid command id: {command_id}") - if "command" not in command or not isinstance(command["command"], list) or not command["command"]: - raise RuntimeError(f"Command '{command_id}' needs a non-empty command list") - - command.setdefault("args", {}) - command.setdefault("timeout_seconds", 30) - - if int(command["timeout_seconds"]) <= 0: - raise RuntimeError(f"Command '{command_id}' has invalid timeout_seconds") - - if not command.get("web_enabled", False): - if not command.get("api_key") or len(command["api_key"]) < 16: - raise RuntimeError(f"Command '{command_id}' needs api_key with at least 16 chars") - - placeholders = set() - for part in command["command"]: - for match in re.findall(r"{([a-zA-Z0-9_]+)}", str(part)): - placeholders.add(match) - - missing_args = placeholders - set(command["args"].keys()) - if missing_args: + if not re.fullmatch( + r"[a-zA-Z0-9_-]{1,64}", + command_id + ): raise RuntimeError( - f"Command '{command_id}' uses undefined args: {', '.join(sorted(missing_args))}" + f"Invalid command id: {command_id}" ) - for arg_name, rule in command["args"].items(): - if not re.fullmatch(r"[a-zA-Z0-9_]{1,64}", arg_name): - raise RuntimeError(f"Command '{command_id}' has invalid arg name: {arg_name}") - pattern = rule.get("pattern", r"^[a-zA-Z0-9_.:/ -]{0,128}$") - try: - re.compile(pattern) - except re.error as exc: + if ( + "command" not in command + or not isinstance(command["command"], list) + or not command["command"] + ): + raise RuntimeError( + f"Command '{command_id}' needs command list" + ) + + + command.setdefault("args", {}) + command.setdefault( + "timeout_seconds", + 30 + ) + + + if int(command["timeout_seconds"]) <= 0: + raise RuntimeError( + f"Command '{command_id}' has invalid timeout" + ) + + + placeholders = set() + + for part in command["command"]: + for match in re.findall( + r"{([a-zA-Z0-9_]+)}", + str(part) + ): + placeholders.add(match) + + + missing = placeholders - set( + command["args"].keys() + ) + + if missing: + raise RuntimeError( + f"Command '{command_id}' missing args: " + + ", ".join(sorted(missing)) + ) + + + for arg_name, rule in command["args"].items(): + + if not re.fullmatch( + r"[a-zA-Z0-9_]{1,64}", + arg_name + ): raise RuntimeError( - f"Command '{command_id}' arg '{arg_name}' has invalid regex: {exc}" + f"Invalid arg name: {arg_name}" ) - if command.get("web_enabled", False) and rule.get("required", False): - if "web_default" not in rule and "default" not in rule: - raise RuntimeError( - f"Web command '{command_id}' needs web_default/default for required arg '{arg_name}'" - ) + pattern = rule.get( + "pattern", + r"^[a-zA-Z0-9_.:/ -]{0,128}$" + ) + + try: + re.compile(pattern) + + except re.error as exc: + raise RuntimeError( + f"Invalid regex {arg_name}: {exc}" + ) CONFIG = load_config() + class ApiHandler(BaseHTTPRequestHandler): - server_version = "RpiScriptApi/1.3" + + server_version = "RpiScriptApi/2.0" + def log_message(self, fmt, *args): + print( "%s [%s] %s" - % (self.client_address[0], self.log_date_time_string(), fmt % args) + % + ( + self.client_address[0], + self.log_date_time_string(), + fmt % args + ) ) + def send_json(self, status, payload): - data = json.dumps(payload, indent=2).encode("utf-8") - self.send_response(status) - self.send_header("Content-Type", "application/json; charset=utf-8") - self.send_header("Content-Length", str(len(data))) - self.send_security_headers() - self.end_headers() - self.wfile.write(data) - def send_html(self, status, body): - data = body.encode("utf-8") + data = json.dumps( + payload, + indent=2 + ).encode("utf-8") + + self.send_response(status) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(data))) - self.send_security_headers() - self.end_headers() - self.wfile.write(data) - def send_security_headers(self): - self.send_header("X-Content-Type-Options", "nosniff") - self.send_header("X-Frame-Options", "DENY") - self.send_header("Referrer-Policy", "no-referrer") - self.send_header("Cache-Control", "no-store") self.send_header( - "Content-Security-Policy", - "default-src 'self'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'" + "Content-Type", + "application/json; charset=utf-8" ) + self.send_header( + "Content-Length", + str(len(data)) + ) + + self.send_security_headers() + + self.end_headers() + + self.wfile.write(data) + + + + def send_security_headers(self): + + self.send_header( + "X-Content-Type-Options", + "nosniff" + ) + + self.send_header( + "X-Frame-Options", + "DENY" + ) + + self.send_header( + "Cache-Control", + "no-store" + ) + + + def read_body_raw(self): + try: - length = int(self.headers.get("Content-Length", "0")) + length = int( + self.headers.get( + "Content-Length", + "0" + ) + ) + except ValueError: - raise ValueError("invalid content-length") + raise ValueError( + "invalid content-length" + ) - if length > int(CONFIG["max_body_bytes"]): - raise ValueError("request body too large") - if length == 0: - return b"" + if length > CONFIG["max_body_bytes"]: + raise ValueError( + "request too large" + ) + return self.rfile.read(length) + + def read_json_body(self): + raw = self.read_body_raw() + if not raw: return {} + try: - return json.loads(raw.decode("utf-8")) + + return json.loads( + raw.decode("utf-8") + ) + except json.JSONDecodeError as exc: - raise ValueError(f"invalid json: {exc}") - def read_form_body(self): - raw = self.read_body_raw() - if not raw: - return {} + raise ValueError( + f"invalid json: {exc}" + ) - content_type = self.headers.get("Content-Type", "") - if "application/x-www-form-urlencoded" not in content_type: - raise ValueError("invalid form content-type") - parsed = parse_qs(raw.decode("utf-8"), keep_blank_values=True) - return {k: v[0] if v else "" for k, v in parsed.items()} - - def do_GET(self): - path = urlparse(self.path).path - - if path == "/": - self.send_html(200, render_index()) - return - - if path == "/health": - self.send_json(200, {"ok": True}) - return - - if path == "/commands": - self.send_json(200, list_commands()) - return - - self.send_json(404, {"ok": False, "error": "not found"}) def do_POST(self): - path = urlparse(self.path).path - api_match = re.fullmatch(r"/run/([a-zA-Z0-9_-]+)", path) - web_match = re.fullmatch(r"/web/run/([a-zA-Z0-9_-]+)", path) + path = urlparse( + self.path + ).path + + + if path == "/run": + + self.handle_run_api() - if api_match: - self.handle_run_api(api_match.group(1)) return + + web_match = re.fullmatch( + r"/web/run/([a-zA-Z0-9_-]+)", + path + ) + + if web_match: - self.handle_run_web(web_match.group(1)) + + self.handle_run_web( + web_match.group(1) + ) + return - self.send_json(404, {"ok": False, "error": "not found"}) - def handle_run_api(self, command_id): - command_cfg = CONFIG["commands"].get(command_id) + self.send_json( + 404, + { + "ok": False, + "error": "not found" + } + ) - if not command_cfg: - self.send_json(404, {"ok": False, "error": "unknown command"}) - return - api_key = self.headers.get("X-API-Key", "") - expected_key = command_cfg.get("api_key", "") - if not expected_key or not secrets.compare_digest(api_key, expected_key): - self.send_json(403, {"ok": False, "error": "invalid api key"}) - return + def handle_run_api(self): try: + + api_key = self.headers.get( + "X-API-Key", + "" + ) + + + if not secrets.compare_digest( + api_key, + CONFIG["api_key"] + ): + + self.send_json( + 403, + { + "ok": False, + "error": "invalid api key" + } + ) + + return + + + payload = self.read_json_body() - args = payload.get("args", {}) - if not isinstance(args, dict): - raise ValueError("args must be an object") - validated_args = validate_args(command_cfg, args) - command = build_command(command_cfg, validated_args) - result = run_command(command_id, command, command_cfg) - status = 200 if result["ok"] else 500 - self.send_json(status, result) + command_id = payload.get( + "command" + ) + + + if not isinstance( + command_id, + str + ): + + raise ValueError( + "missing command" + ) + + + command_cfg = CONFIG["commands"].get( + command_id + ) + + + if not command_cfg: + + self.send_json( + 404, + { + "ok": False, + "error": "unknown command" + } + ) + + return + + + + args = payload.get( + "args", + {} + ) + + + if not isinstance( + args, + dict + ): + + raise ValueError( + "args must be object" + ) + + + validated_args = validate_args( + command_cfg, + args + ) + + + command = build_command( + command_cfg, + validated_args + ) + + + result = run_command( + command_id, + command, + command_cfg + ) + + + self.send_json( + 200 if result["ok"] else 500, + result + ) - except subprocess.TimeoutExpired as exc: - self.send_json(504, timeout_result(command_id, exc)) except Exception as exc: - self.send_json(400, {"ok": False, "error": str(exc)}) + + self.send_json( + 400, + { + "ok": False, + "error": str(exc) + } + ) def handle_run_web(self, command_id): + command_cfg = CONFIG["commands"].get(command_id) if not command_cfg: - self.send_html(404, render_page("Unknown command", "

Unknown command.

")) + + self.send_html( + 404, + render_page( + "Unknown command", + "

Unknown command.

" + ) + ) + return - if not command_cfg.get("web_enabled", False): - self.send_html(403, render_page("Forbidden", "

Command is not enabled for web UI.

")) + + if not command_cfg.get( + "web_enabled", + False + ): + + self.send_html( + 403, + render_page( + "Forbidden", + "

Command disabled.

" + ) + ) + return + try: - form = self.read_form_body() - csrf = form.get("csrf", "") - if not secrets.compare_digest(csrf, WEB_CSRF_TOKEN): - raise ValueError("invalid csrf token") + form = self.read_form_body() + + csrf = form.get( + "csrf", + "" + ) + + + if not secrets.compare_digest( + csrf, + WEB_CSRF_TOKEN + ): + + raise ValueError( + "invalid csrf token" + ) + args = {} - for name, rule in command_cfg.get("args", {}).items(): + + for name, rule in command_cfg.get( + "args", + {} + ).items(): + if "web_default" in rule: + args[name] = rule["web_default"] + elif "default" in rule: + args[name] = rule["default"] - elif rule.get("required", False): - raise ValueError(f"missing web_default for required arg: {name}") - validated_args = validate_args(command_cfg, args) - command = build_command(command_cfg, validated_args) - result = run_command(command_id, command, command_cfg) - self.send_html(200, render_result(command_id, result)) + validated_args = validate_args( + command_cfg, + args + ) + + + command = build_command( + command_cfg, + validated_args + ) + + + result = run_command( + command_id, + command, + command_cfg + ) + + + self.send_html( + 200, + render_result( + command_id, + result + ) + ) - except subprocess.TimeoutExpired as exc: - self.send_html(504, render_result(command_id, timeout_result(command_id, exc))) except Exception as exc: - self.send_html(400, render_page("Error", f"
{html.escape(str(exc))}
")) + + self.send_html( + 400, + render_page( + "Error", + f"
{html.escape(str(exc))}
" + ) + ) + def list_commands(): + return { + "ok": True, + "commands": [ + { "id": command_id, - "description": command.get("description", ""), - "args": list(command.get("args", {}).keys()), - "web_enabled": bool(command.get("web_enabled", False)) + "description": command.get( + "description", + "" + ), + "args": list( + command.get( + "args", + {} + ).keys() + ), + "web_enabled": bool( + command.get( + "web_enabled", + False + ) + ) } - for command_id, command in CONFIG["commands"].items() + + for command_id, command + in CONFIG["commands"].items() + ] + } + + def validate_args(command_cfg, incoming_args): - arg_rules = command_cfg.get("args", {}) + + arg_rules = command_cfg.get( + "args", + {} + ) + validated = {} - unknown = set(incoming_args.keys()) - set(arg_rules.keys()) + + unknown = set( + incoming_args.keys() + ) - set( + arg_rules.keys() + ) + + if unknown: - raise ValueError(f"unknown args: {', '.join(sorted(unknown))}") + + raise ValueError( + "unknown args: " + + ", ".join(sorted(unknown)) + ) + + for name, rule in arg_rules.items(): - value = incoming_args.get(name, rule.get("default")) - if value in [None, ""]: - if rule.get("required", False): - raise ValueError(f"missing arg: {name}") + value = incoming_args.get( + name, + rule.get("default") + ) + + + if value in [ + None, + "" + ]: + + if rule.get( + "required", + False + ): + + raise ValueError( + f"missing arg: {name}" + ) + + value = "" - value = str(value) - pattern = rule.get("pattern", r"^[a-zA-Z0-9_.:/ -]{0,128}$") - if not re.fullmatch(pattern, value): - raise ValueError(f"invalid arg: {name}") + + value = str(value) + + + pattern = rule.get( + "pattern", + r"^[a-zA-Z0-9_.:/ -]{0,128}$" + ) + + + if not re.fullmatch( + pattern, + value + ): + + raise ValueError( + f"invalid arg: {name}" + ) + validated[name] = value - allowlist = command_cfg.get("allowlist") - if allowlist: - checked_values = [ - validated.get(name) - for name in command_cfg.get("allowlist_args", []) - ] - if checked_values not in allowlist: - raise ValueError(f"not allowed: {checked_values}") return validated + + + def build_command(command_cfg, args): + command = [] + for part in command_cfg["command"]: + value = str(part) + + for key, arg_value in args.items(): - value = value.replace("{" + key + "}", arg_value) + + value = value.replace( + "{" + key + "}", + arg_value + ) + + command.append(value) + return command + + + def run_command(command_id, command, command_cfg): - request_id = str(uuid.uuid4()) - started = time.time() - timeout = int(command_cfg.get("timeout_seconds", 30)) - env = { - "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", - "LANG": "C.UTF-8", - "LC_ALL": "C.UTF-8", - "REQUEST_ID": request_id - } - - env.update({str(k): str(v) for k, v in command_cfg.get("environment", {}).items()}) - - print(f"request_id={request_id} command_id={command_id} command={command}") - - completed = subprocess.run( - command, - shell=False, - capture_output=True, - text=True, - timeout=timeout, - check=False, - env=env + request_id = str( + uuid.uuid4() ) - duration_ms = int((time.time() - started) * 1000) - return { - "ok": completed.returncode == 0, - "request_id": request_id, - "command_id": command_id, - "command": command, - "exit_code": completed.returncode, - "timed_out": False, - "duration_ms": duration_ms, - "stdout": completed.stdout, - "stderr": completed.stderr + started = time.time() + + + timeout = int( + command_cfg.get( + "timeout_seconds", + 30 + ) + ) + + + + env = { + + "PATH": + "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + + "LANG": + "C.UTF-8", + + "LC_ALL": + "C.UTF-8", + + "REQUEST_ID": + request_id + } -def timeout_result(command_id, exc): + env.update( + { + str(k): str(v) + + for k,v + + in command_cfg.get( + "environment", + {} + ).items() + } + ) + + + + print( + f"request_id={request_id} " + f"command_id={command_id} " + f"command={command}" + ) + + + + completed = subprocess.run( + + command, + + shell=False, + + capture_output=True, + + text=True, + + timeout=timeout, + + check=False, + + env=env + + ) + + + + duration_ms = int( + (time.time()-started)*1000 + ) + + + return { - "ok": False, - "request_id": str(uuid.uuid4()), - "command_id": command_id, - "command": getattr(exc, "cmd", []), - "exit_code": None, - "timed_out": True, - "duration_ms": None, - "stdout": exc.stdout or "", - "stderr": exc.stderr or "command timed out" + + "ok": + completed.returncode == 0, + + + "request_id": + request_id, + + + "command_id": + command_id, + + + "command": + command, + + + "exit_code": + completed.returncode, + + + "duration_ms": + duration_ms, + + + "stdout": + completed.stdout, + + + "stderr": + completed.stderr + } + + + def render_index(): + cards = [] + for command_id, command in CONFIG["commands"].items(): - if not command.get("web_enabled", False): + + + if not command.get( + "web_enabled", + False + ): + continue - title = html.escape(command.get("web_title", command_id)) - desc = html.escape(command.get("description", "")) - cards.append(f""" -
- -

{title}

-

{desc}

- -
- """) + + cards.append( + f""" +
+ +

{command.get("web_title", command_id)}

+

{command.get("description","")}

+ +
+""" + ) + + if not cards: - cards.append("

No web-enabled commands configured.

") - return render_page("Raspberry Pi Script API", "\n".join(cards)) + cards.append( + "

No commands.

" + ) + + + return render_page( + "Raspberry Pi Script API", + "\n".join(cards) + ) + + + def render_result(command_id, result): - body = f""" -

Back

-

Result: {html.escape(command_id)}

- - - - - - -
OK{html.escape(str(result.get("ok")))}
Exit Code{html.escape(str(result.get("exit_code")))}
Timed out{html.escape(str(result.get("timed_out")))}
Duration{html.escape(str(result.get("duration_ms")))} ms
Request ID{html.escape(str(result.get("request_id")))}
+ return render_page( + + "Command Result", + + f""" +

{html.escape(command_id)}

+ +

+OK: +{html.escape(str(result.get("ok")))} +

+ +

+Exit: +{html.escape(str(result.get("exit_code")))} +

+ +
+{html.escape(result.get("stdout",""))}
+
+ +
+{html.escape(result.get("stderr",""))}
+
+""" + + ) -

stdout

- 
{html.escape(str(result.get("stdout", "")))}
-

stderr

- 
{html.escape(str(result.get("stderr", "")))}
- """ - return render_page("Command Result", body) def render_page(title, body): - return f""" - + + return f""" + + + + - - {html.escape(title)} - - + +{html.escape(title)} + + -

{html.escape(title)}

- {body} + +

{html.escape(title)}

+ +{body} + -""" + + +""" + + + def main(): - server = ThreadingHTTPServer((CONFIG["host"], int(CONFIG["port"])), ApiHandler) - print(f"Listening on http://{CONFIG['host']}:{CONFIG['port']}") - print(f"Config: {CONFIG_PATH}") + + server = ThreadingHTTPServer( + + ( + CONFIG["host"], + int(CONFIG["port"]) + ), + + ApiHandler + + ) + + + print( + f"Listening on {CONFIG['host']}:{CONFIG['port']}" + ) + + server.serve_forever() + + + if __name__ == "__main__": - main() - + main() \ No newline at end of file