devplacepy/devplacepy/services/xmlrpc/server.py

# retoor <retoor@molodetz.nl>

from __future__ import annotations

import logging
from socketserver import ForkingMixIn
from xmlrpc.server import SimpleXMLRPCDispatcher, SimpleXMLRPCRequestHandler, SimpleXMLRPCServer

from defusedxml.xmlrpc import monkey_patch as _harden_xmlrpc

from devplacepy.config import XMLRPC_BIND, XMLRPC_PORT

_harden_xmlrpc()
from . import bridge
from .registry import Method, build_index, method_help, method_signature

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("xmlrpc.server")

RPC_PATHS = ("/", "/RPC2", "/xmlrpc")


def _auth_from_headers(headers) -> dict[str, str]:
    auth: dict[str, str] = {}
    api_key = headers.get("X-API-KEY") or headers.get("x-api-key")
    authorization = headers.get("Authorization") or headers.get("authorization") or ""
    if not api_key and authorization.lower().startswith("bearer "):
        api_key = authorization[7:].strip()
    if authorization.lower().startswith("basic "):
        auth["authorization"] = authorization.strip()
    if api_key:
        auth["api_key"] = api_key.strip()
    real_ip = headers.get("X-Real-IP") or headers.get("x-real-ip")
    if real_ip:
        auth["real_ip"] = real_ip
    forwarded = headers.get("X-Forwarded-For") or headers.get("x-forwarded-for")
    if forwarded:
        auth["forwarded_for"] = forwarded
    return auth


class _RequestHandler(SimpleXMLRPCRequestHandler):
    rpc_paths = RPC_PATHS

    def decode_request_content(self, data):
        if b"<!DOCTYPE" in data or b"<!ENTITY" in data:
            self.send_response(400)
            self.send_header("Content-length", "0")
            self.end_headers()
            return None
        self.server.current_auth = _auth_from_headers(self.headers)
        return super().decode_request_content(data)

    def log_message(self, *args, **kwargs):
        return


class BridgeDispatcher(SimpleXMLRPCDispatcher):
    def system_methodHelp(self, method_name):
        target = self.index.get(method_name)
        if target is not None:
            return method_help(target)
        return super().system_methodHelp(method_name)

    def system_methodSignature(self, method_name):
        target = self.index.get(method_name)
        if target is not None:
            return method_signature(target)
        return super().system_methodSignature(method_name)


class ForkingXMLRPCServer(ForkingMixIn, BridgeDispatcher, SimpleXMLRPCServer):
    allow_reuse_address = True
    daemon_threads = True

    def __init__(self, addr, index):
        self.index = index
        self.current_auth: dict[str, str] = {}
        SimpleXMLRPCServer.__init__(
            self,
            addr,
            requestHandler=_RequestHandler,
            allow_none=True,
            logRequests=False,
        )


def _make_handler(server: "ForkingXMLRPCServer", method: Method):
    def handler(*args):
        params = args[0] if args else {}
        return bridge.call(method, params, server.current_auth)

    handler.__name__ = method.name
    handler.__doc__ = method_help(method)
    return handler


def build_server(host: str, port: int) -> ForkingXMLRPCServer:
    index = build_index()
    server = ForkingXMLRPCServer((host, port), index)
    server.register_introspection_functions()
    server.register_multicall_functions()
    for name, method in index.items():
        server.register_function(_make_handler(server, method), name)
    logger.info("XML-RPC bridge registered %d methods", len(index))
    return server


def main() -> None:
    server = build_server(XMLRPC_BIND, XMLRPC_PORT)
    logger.info("Forking XML-RPC server listening on %s:%d", XMLRPC_BIND, XMLRPC_PORT)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("XML-RPC server interrupted")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()