Server-side programs

Version 1 (UNSTABLE): cmk.server_side_programs.v1_unstable

Warning

This version of the API is work in progress and not yet stable. It is not recommended to use this version in production systems.

However: we do intend to stabilize this API version in the future and release it, so you are encouraged to experiment and give us feedback.

Scope

This API provides functionality to be used by server-side programs running on the Checkmk server. It is written with special agents and active checks in mind – we do not guarantee they work in other circumstances.

This is the first version of the server-side programs API.

class HostnameValidationAdapter(hostname)

Bases: HTTPAdapter

An HTTPAdapter that enforces hostname validation against a given hostname.

In Checkmk we often find ourselves in situations where we want to connect to a server via its IP address, but the server’s SSL certificate is issued for a hostname. This adapter allows us to enforce hostname validation against the expected hostname, even when connecting via IP.

Example

#!/usr/bin/env/python3 … address = “[12600:1406:5e00:6::17ce:bc1b]” # this how we contact the server cert_server_name = “example.com” # this is the name we expect in the server’s certificate session = requests.Session()

session.mount(f”https://{address}”, HostnameValidationAdapter(cert_server_name))

response = session.get(f”https://{address}/some/api/endpoint”) …

cert_verify(conn, url, verify, cert)

Verify a SSL certificate. This method should not be called from user code.

Since the superclass method is untyped anyway, we type the arguments as Never to indicate that this function should not be called in normal usage.

Return type:: None

class Storage(program_ident, host)

Bases: object

Storage interface for special agents and active checks.

These programs may have the need to save some form of data in between runs. This interface provides a possibility to read and write text to a persistent storage. You can instantiate the storage interface with a program identifier and a host name, these are used to namespace the storage.

Parameters:

program_ident (str) – A string identifying the program using the storage. This is used to namespace the storage.
host (str) – The host name the program is working for. This is used to namespace the storage.

read(key, default)

Read content from the storage.

If the key is unknown or the content is corrupted, return default.

Raises a RuntimeError if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set.

Return type:: Union[str, TypeVar(T)]

unset(key)

Remove key and its content from the storage.

Raises a RuntimeError if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set.

Return type:: None

write(key, content)

Write text content to the storage.

Raises a RuntimeError if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set.

Return type:: None

vcrtrace(filter_body=<function <lambda>>, filter_query_parameters=(), filter_headers=(), filter_post_data_parameters=())

Returns the class of an argparse.Action to enter a vcr context

Provided keyword arguments will be passed to the call of vcrpy.VCR. The minimal change to use vcrtrace in your program is to add this line to your argument parsing:

parser.add_argument(”–vcrtrace”, action=vcrtrace())

If this flag is set to a TRACEFILE that does not exist yet, it will be created and all requests the program sends and their corresponding answers will be recorded in said file. If the file already exists, no requests are sent to the server, but the responses will be replayed from the tracefile.

The destination attribute will be set to True if the option was specified, the provided default otherwise.

Return type:: type[Action]

report_agent_crashes(name, version)

“Decorator factory to report crashes from agents

Wrapping a function with the returned decorator will catch all exceptions raised by the function and create a crash report.

Parameters:

name (str) – The name of the agent
version (str) – The version of the agent

Return type:

Callable[[Callable[[ParamSpec(_P, bound= None)], int]], Callable[[ParamSpec(_P, bound= None)], int]]

Example

>>> @report_agent_crashes("smith", "1.0.0")
... def main() -> int:
...     # your code here
...     return 0