forked from TillRepo/IrSender
1885 lines
25 KiB
Python
Executable File
1885 lines
25 KiB
Python
Executable File
#!/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
|
|
|
|
|
|
# ============================================================
|
|
# Configuration
|
|
# ============================================================
|
|
|
|
CONFIG_PATH = Path(
|
|
sys.argv[1]
|
|
if len(sys.argv) > 1
|
|
else "./commands.json"
|
|
)
|
|
|
|
WEB_CSRF_TOKEN = secrets.token_urlsafe(32)
|
|
|
|
|
|
# ============================================================
|
|
# Load configuration
|
|
# ============================================================
|
|
|
|
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",
|
|
8082
|
|
)
|
|
|
|
cfg.setdefault(
|
|
"max_body_bytes",
|
|
65536
|
|
)
|
|
|
|
cfg.setdefault(
|
|
"commands",
|
|
{}
|
|
)
|
|
|
|
|
|
if not cfg.get("api_key"):
|
|
|
|
raise RuntimeError(
|
|
"Global api_key missing"
|
|
)
|
|
|
|
|
|
if len(cfg["api_key"]) < 16:
|
|
|
|
raise RuntimeError(
|
|
"api_key must contain at least 16 characters"
|
|
)
|
|
|
|
|
|
if not isinstance(
|
|
cfg["commands"],
|
|
dict
|
|
) or not cfg["commands"]:
|
|
|
|
raise RuntimeError(
|
|
"No commands configured"
|
|
)
|
|
|
|
|
|
validate_config(cfg)
|
|
|
|
|
|
return cfg
|
|
|
|
|
|
|
|
# ============================================================
|
|
# Validate configuration
|
|
# ============================================================
|
|
|
|
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 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"
|
|
)
|
|
|
|
|
|
|
|
#
|
|
# Check placeholders
|
|
#
|
|
|
|
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))
|
|
)
|
|
|
|
|
|
|
|
#
|
|
# Validate argument rules
|
|
#
|
|
|
|
for arg_name, rule in command["args"].items():
|
|
|
|
|
|
if not re.fullmatch(
|
|
r"[a-zA-Z0-9_]{1,64}",
|
|
arg_name
|
|
):
|
|
|
|
raise RuntimeError(
|
|
f"Invalid argument 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"Invalid regex for {arg_name}: {exc}"
|
|
)
|
|
|
|
|
|
|
|
CONFIG = load_config()
|
|
|
|
|
|
|
|
# ============================================================
|
|
# HTTP Handler
|
|
# ============================================================
|
|
|
|
class ApiHandler(
|
|
BaseHTTPRequestHandler
|
|
):
|
|
|
|
|
|
server_version = (
|
|
"RpiScriptApi/3.0"
|
|
)
|
|
|
|
|
|
|
|
def log_message(
|
|
self,
|
|
fmt,
|
|
*args
|
|
):
|
|
|
|
print(
|
|
"%s [%s] %s"
|
|
%
|
|
(
|
|
self.client_address[0],
|
|
self.log_date_time_string(),
|
|
fmt % args
|
|
)
|
|
)
|
|
|
|
|
|
|
|
# --------------------------------------------------------
|
|
# Responses
|
|
# --------------------------------------------------------
|
|
|
|
def send_security_headers(self):
|
|
|
|
self.send_header(
|
|
"X-Content-Type-Options",
|
|
"nosniff"
|
|
)
|
|
|
|
self.send_header(
|
|
"Content-Security-Policy",
|
|
"frame-ancestors 'self' http://192.168.178.60:8080"
|
|
)
|
|
|
|
self.send_header(
|
|
"Referrer-Policy",
|
|
"no-referrer"
|
|
)
|
|
|
|
self.send_header(
|
|
"Cache-Control",
|
|
"no-store"
|
|
)
|
|
|
|
|
|
|
|
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
|
|
)
|
|
|
|
|
|
|
|
# --------------------------------------------------------
|
|
# Request body handling
|
|
# --------------------------------------------------------
|
|
|
|
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()
|
|
}
|
|
# --------------------------------------------------------
|
|
# HTTP GET
|
|
# --------------------------------------------------------
|
|
|
|
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"
|
|
}
|
|
)
|
|
|
|
|
|
|
|
# --------------------------------------------------------
|
|
# HTTP POST
|
|
# --------------------------------------------------------
|
|
|
|
def do_POST(self):
|
|
|
|
path = urlparse(
|
|
self.path
|
|
).path
|
|
|
|
|
|
|
|
#
|
|
# OpenHAB API
|
|
#
|
|
if path == "/run":
|
|
|
|
self.handle_run_api()
|
|
|
|
return
|
|
|
|
|
|
|
|
#
|
|
# Web Button UI
|
|
#
|
|
web_match = re.fullmatch(
|
|
r"/web/run/([a-zA-Z0-9_-]+)",
|
|
path
|
|
)
|
|
|
|
|
|
if web_match:
|
|
|
|
self.handle_run_web(
|
|
web_match.group(1)
|
|
)
|
|
|
|
return
|
|
|
|
|
|
|
|
self.send_json(
|
|
404,
|
|
{
|
|
"ok": False,
|
|
"error": "not found"
|
|
}
|
|
)
|
|
|
|
|
|
|
|
# --------------------------------------------------------
|
|
# API execution
|
|
# --------------------------------------------------------
|
|
|
|
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()
|
|
|
|
|
|
|
|
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(
|
|
"unknown",
|
|
exc
|
|
)
|
|
)
|
|
|
|
|
|
except Exception as exc:
|
|
|
|
|
|
self.send_json(
|
|
400,
|
|
{
|
|
"ok": False,
|
|
"error": str(exc)
|
|
}
|
|
)
|
|
|
|
|
|
|
|
# --------------------------------------------------------
|
|
# Web UI execution
|
|
# --------------------------------------------------------
|
|
|
|
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 disabled.</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"]
|
|
|
|
|
|
|
|
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>"
|
|
)
|
|
)
|
|
|
|
|
|
|
|
# ============================================================
|
|
# Command handling functions
|
|
# ============================================================
|
|
|
|
|
|
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
|
|
):
|
|
|
|
rules = command_cfg.get(
|
|
"args",
|
|
{}
|
|
)
|
|
|
|
|
|
validated = {}
|
|
|
|
|
|
|
|
unknown = (
|
|
set(incoming_args.keys())
|
|
-
|
|
set(rules.keys())
|
|
)
|
|
|
|
|
|
|
|
if unknown:
|
|
|
|
raise ValueError(
|
|
"unknown args: "
|
|
+
|
|
", ".join(sorted(unknown))
|
|
)
|
|
|
|
|
|
|
|
for name, rule in 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
|
|
|
|
|
|
|
|
#
|
|
# Optional allowlist support
|
|
#
|
|
|
|
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(
|
|
"not allowed"
|
|
)
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================
|
|
# Execute 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} "
|
|
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":
|
|
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"
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================
|
|
# Web Interface
|
|
# ============================================================
|
|
|
|
def render_index():
|
|
|
|
groups = {}
|
|
|
|
|
|
for command_id, command in CONFIG["commands"].items():
|
|
|
|
if not command.get(
|
|
"web_enabled",
|
|
False
|
|
):
|
|
continue
|
|
|
|
|
|
title = command.get(
|
|
"web_title",
|
|
command_id
|
|
)
|
|
|
|
|
|
description = command.get(
|
|
"description",
|
|
""
|
|
)
|
|
|
|
|
|
# Gruppe anhand des ersten Wortes
|
|
group_name = title.split(
|
|
" ",
|
|
1
|
|
)[0]
|
|
|
|
|
|
groups.setdefault(
|
|
group_name,
|
|
[]
|
|
)
|
|
|
|
|
|
groups[group_name].append(
|
|
{
|
|
"id": command_id,
|
|
"title": title,
|
|
"description": description
|
|
}
|
|
)
|
|
|
|
|
|
|
|
sections = []
|
|
|
|
|
|
for group_name, commands in sorted(
|
|
groups.items()
|
|
):
|
|
|
|
|
|
buttons = []
|
|
|
|
|
|
for cmd in commands:
|
|
|
|
|
|
buttons.append(
|
|
f"""
|
|
<form method="post"
|
|
action="/web/run/{html.escape(cmd["id"])}">
|
|
|
|
<input type="hidden"
|
|
name="csrf"
|
|
value="{WEB_CSRF_TOKEN}">
|
|
|
|
|
|
<button class="action-button"
|
|
type="submit">
|
|
|
|
{html.escape(cmd["title"])}
|
|
|
|
</button>
|
|
|
|
</form>
|
|
"""
|
|
)
|
|
|
|
|
|
|
|
sections.append(
|
|
f"""
|
|
|
|
<div class="group-card">
|
|
|
|
|
|
<div class="group-title">
|
|
|
|
{html.escape(group_name)}
|
|
|
|
</div>
|
|
|
|
|
|
<div class="button-grid">
|
|
|
|
{"".join(buttons)}
|
|
|
|
</div>
|
|
|
|
|
|
</div>
|
|
|
|
"""
|
|
)
|
|
|
|
|
|
|
|
if not sections:
|
|
|
|
sections.append(
|
|
"""
|
|
<div class="empty">
|
|
Keine Befehle verfügbar.
|
|
</div>
|
|
"""
|
|
)
|
|
|
|
|
|
|
|
return render_page(
|
|
|
|
"IR Steuerung",
|
|
|
|
"\n".join(sections)
|
|
|
|
)
|
|
def render_page(
|
|
title,
|
|
body
|
|
):
|
|
|
|
return f"""
|
|
<!doctype html>
|
|
|
|
<html>
|
|
|
|
<head>
|
|
|
|
<meta charset="utf-8">
|
|
|
|
|
|
<title>
|
|
{html.escape(title)}
|
|
</title>
|
|
|
|
|
|
<meta name="viewport"
|
|
content="width=device-width, initial-scale=1">
|
|
|
|
|
|
<style>
|
|
|
|
|
|
:root {{
|
|
|
|
--bg:
|
|
#111827;
|
|
|
|
--card:
|
|
#1f2937;
|
|
|
|
--card-hover:
|
|
#374151;
|
|
|
|
--text:
|
|
#f9fafb;
|
|
|
|
--secondary:
|
|
#9ca3af;
|
|
|
|
--accent:
|
|
#3b82f6;
|
|
|
|
--success:
|
|
#10b981;
|
|
|
|
--error:
|
|
#ef4444;
|
|
|
|
}}
|
|
|
|
|
|
|
|
* {{
|
|
|
|
box-sizing:
|
|
border-box;
|
|
|
|
}}
|
|
|
|
|
|
|
|
body {{
|
|
|
|
margin:
|
|
0;
|
|
|
|
padding:
|
|
20px;
|
|
|
|
min-height:
|
|
100vh;
|
|
|
|
background:
|
|
var(--bg);
|
|
|
|
color:
|
|
var(--text);
|
|
|
|
font-family:
|
|
system-ui,
|
|
-apple-system,
|
|
BlinkMacSystemFont,
|
|
"Segoe UI",
|
|
sans-serif;
|
|
|
|
}}
|
|
|
|
|
|
|
|
h1 {{
|
|
|
|
font-size:
|
|
1.8rem;
|
|
|
|
margin-bottom:
|
|
25px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.grid {{
|
|
|
|
display:
|
|
grid;
|
|
|
|
grid-template-columns:
|
|
repeat(
|
|
auto-fit,
|
|
minmax(
|
|
220px,
|
|
1fr
|
|
)
|
|
);
|
|
|
|
gap:
|
|
20px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.card {{
|
|
|
|
background:
|
|
var(--card);
|
|
|
|
border-radius:
|
|
16px;
|
|
|
|
padding:
|
|
20px;
|
|
|
|
box-shadow:
|
|
0 8px 20px
|
|
rgba(
|
|
0,
|
|
0,
|
|
0,
|
|
.25
|
|
);
|
|
|
|
}}
|
|
|
|
|
|
|
|
.card form {{
|
|
|
|
display:
|
|
flex;
|
|
|
|
flex-direction:
|
|
column;
|
|
|
|
gap:
|
|
12px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.card-title {{
|
|
|
|
font-size:
|
|
1.25rem;
|
|
|
|
font-weight:
|
|
600;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.card-description {{
|
|
|
|
color:
|
|
var(--secondary);
|
|
|
|
min-height:
|
|
20px;
|
|
|
|
}}
|
|
|
|
.group-card {{
|
|
|
|
background:
|
|
var(--card);
|
|
|
|
border-radius:
|
|
16px;
|
|
|
|
padding:
|
|
20px;
|
|
|
|
margin-bottom:
|
|
25px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.group-title {{}
|
|
|
|
font-size:
|
|
1.4rem;
|
|
|
|
font-weight:
|
|
700;
|
|
|
|
margin-bottom:
|
|
15px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.button-grid {{
|
|
|
|
display:
|
|
grid;
|
|
|
|
grid-template-columns:
|
|
repeat(
|
|
auto-fit,
|
|
minmax(
|
|
150px,
|
|
1fr
|
|
)
|
|
);
|
|
|
|
gap:
|
|
12px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.action-button {{
|
|
|
|
width:
|
|
100%;
|
|
|
|
margin:
|
|
0;
|
|
|
|
}}
|
|
|
|
|
|
|
|
button {{
|
|
|
|
margin-top:
|
|
10px;
|
|
|
|
padding:
|
|
14px;
|
|
|
|
border:
|
|
none;
|
|
|
|
border-radius:
|
|
12px;
|
|
|
|
background:
|
|
var(--accent);
|
|
|
|
color:
|
|
white;
|
|
|
|
font-size:
|
|
1rem;
|
|
|
|
font-weight:
|
|
600;
|
|
|
|
cursor:
|
|
pointer;
|
|
|
|
}}
|
|
|
|
|
|
|
|
button:hover {{
|
|
|
|
opacity:
|
|
.85;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.success {{
|
|
|
|
border-left:
|
|
5px solid var(--success);
|
|
|
|
}}
|
|
|
|
|
|
|
|
.error {{
|
|
|
|
border-left:
|
|
5px solid var(--error);
|
|
|
|
}}
|
|
|
|
|
|
|
|
.status {{
|
|
|
|
margin:
|
|
15px 0;
|
|
|
|
}}
|
|
|
|
|
|
|
|
pre {{
|
|
|
|
background:
|
|
#030712;
|
|
|
|
padding:
|
|
15px;
|
|
|
|
border-radius:
|
|
10px;
|
|
|
|
overflow:
|
|
auto;
|
|
|
|
color:
|
|
#d1d5db;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.back {{
|
|
|
|
display:
|
|
inline-block;
|
|
|
|
margin-top:
|
|
20px;
|
|
|
|
color:
|
|
white;
|
|
|
|
text-decoration:
|
|
none;
|
|
|
|
background:
|
|
#374151;
|
|
|
|
padding:
|
|
12px 18px;
|
|
|
|
border-radius:
|
|
10px;
|
|
|
|
}}
|
|
|
|
|
|
|
|
.empty {{
|
|
|
|
background:
|
|
var(--card);
|
|
|
|
padding:
|
|
20px;
|
|
|
|
border-radius:
|
|
15px;
|
|
|
|
color:
|
|
var(--secondary);
|
|
|
|
}}
|
|
|
|
|
|
|
|
</style>
|
|
|
|
|
|
</head>
|
|
|
|
|
|
<body>
|
|
|
|
|
|
<h1>
|
|
{html.escape(title)}
|
|
</h1>
|
|
|
|
|
|
{body}
|
|
|
|
|
|
</body>
|
|
|
|
|
|
</html>
|
|
"""
|
|
|
|
|
|
|
|
# ============================================================
|
|
# Main
|
|
# ============================================================
|
|
|
|
def main():
|
|
|
|
server = ThreadingHTTPServer(
|
|
|
|
(
|
|
CONFIG["host"],
|
|
int(CONFIG["port"])
|
|
),
|
|
|
|
ApiHandler
|
|
|
|
)
|
|
|
|
|
|
print(
|
|
f"Listening on "
|
|
f"http://{CONFIG['host']}:{CONFIG['port']}"
|
|
)
|
|
|
|
|
|
print(
|
|
f"Config: {CONFIG_PATH}"
|
|
)
|
|
|
|
|
|
server.serve_forever()
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
main() |