IrSender2/script-api.py

#!/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", 8082)
    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 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"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"Invalid regex {arg_name}: {exc}"
                )


CONFIG = load_config()



class ApiHandler(BaseHTTPRequestHandler):

    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
            )
        )


    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_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"
                )
            )

        except ValueError:
            raise ValueError(
                "invalid content-length"
            )


        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")
            )

        except json.JSONDecodeError as exc:

            raise ValueError(
                f"invalid json: {exc}"
            )



    def do_POST(self):

        path = urlparse(
            self.path
        ).path


        if path == "/run":

            self.handle_run_api()

            return


        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"
            }
        )



    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 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 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 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(
            "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



    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} "
        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,


        "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
        ):

            continue



        cards.append(
            f"""
<form method="post" action="/web/run/{command_id}">
<input type="hidden" name="csrf" value="{WEB_CSRF_TOKEN}">
<h2>{command.get("web_title", command_id)}</h2>
<p>{command.get("description","")}</p>
<button type="submit">Run</button>
</form>
"""
        )



    if not cards:

        cards.append(
            "<p>No commands.</p>"
        )


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





def render_result(command_id, result):

    return render_page(

        "Command Result",

        f"""
<h2>{html.escape(command_id)}</h2>

<p>
OK:
{html.escape(str(result.get("ok")))}
</p>

<p>
Exit:
{html.escape(str(result.get("exit_code")))}
</p>

<pre>
{html.escape(result.get("stdout",""))}
</pre>

<pre>
{html.escape(result.get("stderr",""))}
</pre>
"""

    )





def render_page(title, body):

    return f"""
<!doctype html>

<html>

<head>

<title>{html.escape(title)}</title>

</head>

<body>

<h1>{html.escape(title)}</h1>

{body}

</body>

</html>
"""





def main():

    server = ThreadingHTTPServer(

        (
            CONFIG["host"],
            int(CONFIG["port"])
        ),

        ApiHandler

    )


    print(
        f"Listening on {CONFIG['host']}:{CONFIG['port']}"
    )


    server.serve_forever()





if __name__ == "__main__":

    main()