82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
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
|