forked from TillRepo/IrSender
setup
This commit is contained in:
Executable
+499
@@ -0,0 +1,499 @@
|
||||
#!/usr/bin/env python3
|
||||
import html
|
||||
import json
|
||||
import re
|
||||
import secrets
|
||||
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)
|
||||
|
||||
|
||||
def load_config():
|
||||
with CONFIG_PATH.open("r", encoding="utf-8") as f:
|
||||
cfg = json.load(f)
|
||||
|
||||
cfg.setdefault("host", "0.0.0.0")
|
||||
cfg.setdefault("port", 8080)
|
||||
cfg.setdefault("max_body_bytes", 65536)
|
||||
cfg.setdefault("commands", {})
|
||||
|
||||
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:
|
||||
raise RuntimeError(
|
||||
f"Command '{command_id}' uses undefined args: {', '.join(sorted(missing_args))}"
|
||||
)
|
||||
|
||||
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:
|
||||
raise RuntimeError(
|
||||
f"Command '{command_id}' arg '{arg_name}' has invalid regex: {exc}"
|
||||
)
|
||||
|
||||
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}'"
|
||||
)
|
||||
|
||||
|
||||
CONFIG = load_config()
|
||||
|
||||
|
||||
class ApiHandler(BaseHTTPRequestHandler):
|
||||
server_version = "RpiScriptApi/1.3"
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
print(
|
||||
"%s [%s] %s"
|
||||
% (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")
|
||||
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'"
|
||||
)
|
||||
|
||||
def read_body_raw(self):
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
except ValueError:
|
||||
raise ValueError("invalid content-length")
|
||||
|
||||
if length > int(CONFIG["max_body_bytes"]):
|
||||
raise ValueError("request body too large")
|
||||
|
||||
if length == 0:
|
||||
return b""
|
||||
|
||||
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"))
|
||||
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 {}
|
||||
|
||||
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)
|
||||
|
||||
if api_match:
|
||||
self.handle_run_api(api_match.group(1))
|
||||
return
|
||||
|
||||
if web_match:
|
||||
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)
|
||||
|
||||
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
|
||||
|
||||
try:
|
||||
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)
|
||||
|
||||
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)})
|
||||
|
||||
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", "<p>Unknown command.</p>"))
|
||||
return
|
||||
|
||||
if not command_cfg.get("web_enabled", False):
|
||||
self.send_html(403, render_page("Forbidden", "<p>Command is not enabled for web UI.</p>"))
|
||||
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")
|
||||
|
||||
args = {}
|
||||
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))
|
||||
|
||||
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"<pre>{html.escape(str(exc))}</pre>"))
|
||||
|
||||
|
||||
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))
|
||||
}
|
||||
for command_id, command in CONFIG["commands"].items()
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def validate_args(command_cfg, incoming_args):
|
||||
arg_rules = command_cfg.get("args", {})
|
||||
validated = {}
|
||||
|
||||
unknown = set(incoming_args.keys()) - set(arg_rules.keys())
|
||||
if unknown:
|
||||
raise ValueError(f"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 = ""
|
||||
|
||||
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)
|
||||
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
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
|
||||
def timeout_result(command_id, exc):
|
||||
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"
|
||||
}
|
||||
|
||||
|
||||
def render_index():
|
||||
cards = []
|
||||
|
||||
for command_id, command in CONFIG["commands"].items():
|
||||
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"""
|
||||
<form method="post" action="/web/run/{html.escape(command_id)}" class="card">
|
||||
<input type="hidden" name="csrf" value="{html.escape(WEB_CSRF_TOKEN)}">
|
||||
<h2>{title}</h2>
|
||||
<p>{desc}</p>
|
||||
<button type="submit">Run</button>
|
||||
</form>
|
||||
""")
|
||||
|
||||
if not cards:
|
||||
cards.append("<p>No web-enabled commands configured.</p>")
|
||||
|
||||
return render_page("Raspberry Pi Script API", "\n".join(cards))
|
||||
|
||||
|
||||
def render_result(command_id, result):
|
||||
body = f"""
|
||||
<p><a href="/">Back</a></p>
|
||||
<h2>Result: {html.escape(command_id)}</h2>
|
||||
|
||||
<table>
|
||||
<tr><th>OK</th><td>{html.escape(str(result.get("ok")))}</td></tr>
|
||||
<tr><th>Exit Code</th><td>{html.escape(str(result.get("exit_code")))}</td></tr>
|
||||
<tr><th>Timed out</th><td>{html.escape(str(result.get("timed_out")))}</td></tr>
|
||||
<tr><th>Duration</th><td>{html.escape(str(result.get("duration_ms")))} ms</td></tr>
|
||||
<tr><th>Request ID</th><td>{html.escape(str(result.get("request_id")))}</td></tr>
|
||||
</table>
|
||||
|
||||
<h3>stdout</h3>
|
||||
<pre>{html.escape(str(result.get("stdout", "")))}</pre>
|
||||
|
||||
<h3>stderr</h3>
|
||||
<pre>{html.escape(str(result.get("stderr", "")))}</pre>
|
||||
"""
|
||||
|
||||
return render_page("Command Result", body)
|
||||
|
||||
|
||||
def render_page(title, body):
|
||||
return f"""<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<title>{html.escape(title)}</title>
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<style>
|
||||
body {{
|
||||
font-family: system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
|
||||
margin: 2rem;
|
||||
background: #f6f7f8;
|
||||
color: #222;
|
||||
}}
|
||||
.card {{
|
||||
background: white;
|
||||
border: 1px solid #ddd;
|
||||
border-radius: 12px;
|
||||
padding: 1rem;
|
||||
margin-bottom: 1rem;
|
||||
max-width: 520px;
|
||||
box-shadow: 0 1px 3px rgba(0,0,0,.08);
|
||||
}}
|
||||
button {{
|
||||
font-size: 1rem;
|
||||
padding: .65rem 1rem;
|
||||
border: 0;
|
||||
border-radius: 8px;
|
||||
cursor: pointer;
|
||||
background: #222;
|
||||
color: white;
|
||||
}}
|
||||
pre {{
|
||||
background: #111;
|
||||
color: #eee;
|
||||
padding: 1rem;
|
||||
border-radius: 8px;
|
||||
overflow: auto;
|
||||
}}
|
||||
table {{
|
||||
border-collapse: collapse;
|
||||
background: white;
|
||||
margin-bottom: 1rem;
|
||||
}}
|
||||
th, td {{
|
||||
border: 1px solid #ddd;
|
||||
padding: .5rem .8rem;
|
||||
text-align: left;
|
||||
}}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<h1>{html.escape(title)}</h1>
|
||||
{body}
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
||||
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.serve_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user