import re from .constants import (ARG_MAX, HOST_MAX, TOOL_DOWNTIMES, TOOL_HOST_SERVICES, TOOL_PROBLEMS, TOOL_SERVICE_STATUS, TOOL_STATE_HISTORY) HOST_RE = re.compile(r"^[A-Za-z0-9_.\-]+$") SERVICE_RE = re.compile(r"^[\w .:/()\[\]@+,=%#\-]+$") class ToolArgumentError(ValueError): pass def _string(description: str) -> dict: return {"type": "string", "description": description} def _schema(name: str, description: str, properties: dict, required: list[str]) -> dict: return {"type": "function", "function": { "name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}, }} HOST_PROP = _string("Hostname in Checkmk") SERVICE_PROP = _string("Servicename in Checkmk, z.B. 'CPU load'") TOOL_SCHEMAS = [ _schema(TOOL_SERVICE_STATUS, "Aktueller Status eines Services.", {"host": HOST_PROP, "service": SERVICE_PROP}, ["host", "service"]), _schema(TOOL_HOST_SERVICES, "Alle Services eines Hosts mit Status.", {"host": HOST_PROP}, ["host"]), _schema(TOOL_STATE_HISTORY, "Letzte Statuswechsel eines Services.", {"host": HOST_PROP, "service": SERVICE_PROP}, ["host", "service"]), _schema(TOOL_DOWNTIMES, "Geplante Wartungen (Downtimes), optional für einen Host.", {"host": HOST_PROP}, []), _schema(TOOL_PROBLEMS, "Alle aktuell nicht-OK Services.", {}, []), ] SIGNATURES: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = { TOOL_SERVICE_STATUS: (("host", "service"), ()), TOOL_HOST_SERVICES: (("host",), ()), TOOL_STATE_HISTORY: (("host", "service"), ()), TOOL_DOWNTIMES: ((), ("host",)), TOOL_PROBLEMS: ((), ()), } def validate_host(value: object) -> str: if not isinstance(value, str) or not value or len(value) > HOST_MAX or not HOST_RE.match(value): raise ToolArgumentError("host: ungültig (erlaubt: Buchstaben, Ziffern, _ . -)") return value def validate_service(value: object) -> str: if not isinstance(value, str) or not value.strip() or len(value) > ARG_MAX or not SERVICE_RE.match(value): raise ToolArgumentError("service: ungültig (leer, zu lang oder unerlaubte Zeichen)") return value.strip() _VALIDATORS = {"host": validate_host, "service": validate_service} def validate_arguments(tool: str, arguments: object) -> dict[str, str]: if tool not in SIGNATURES: raise ToolArgumentError(f"unbekanntes Tool: {tool[:50]!r}") if not isinstance(arguments, dict): raise ToolArgumentError("Argumente müssen ein Objekt sein") required, optional = SIGNATURES[tool] unknown = set(arguments) - set(required) - set(optional) if unknown: raise ToolArgumentError(f"unbekannte Argumente: {', '.join(sorted(map(str, unknown)))[:100]}") missing = [name for name in required if name not in arguments] if missing: raise ToolArgumentError(f"fehlende Argumente: {', '.join(missing)}") cleaned: dict[str, str] = {} for name in (*required, *optional): if arguments.get(name) in (None, "") and name in optional: continue if name in arguments: cleaned[name] = _VALIDATORS[name](arguments[name]) return cleaned