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

init_poolmanager(connections, maxsize, block=False, **pool_kwargs)

Initializes a urllib3 PoolManager.

This method should not be called from user code, and is only exposed for use when subclassing the HTTPAdapter.

Parameters:

connections (int) – The number of urllib3 connection pools to cache.
maxsize (int) – The maximum number of connections to save in the pool.
block (bool) – Block when no free connections are available.
pool_kwargs (Any) – Extra keyword arguments used to initialize the Pool Manager.

Return type:

None

send(request, *args, **kwargs)

Send the request, injecting the Host header for SNI if not already set.

Return type:: Response

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.

write(key, content)

Write text content to the storage.

Parameters:

key (str) – The unique key to identify the content. It will be sanitized and used as file name. After url quoting, the key must not be longer than 255 characters, otherwise a ValueError is raised.
content (str) – The serialized content to be stored.

Raises:

A RuntimeError is raised if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set. –

Return type:

None

unset(key)

Remove key and its content from the storage.

Parameters:: key (str) – The unique key to identify the content.
Raises:: A RuntimeError is raised if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set. –
Return type:: None

read(key, default)

Read content from the storage.

Parameters:

key (str) – The unique key to identify the content.
default (TypeVar(T)) – The default value to return if the key is unknown or the content is corrupted.

Raises:

A RuntimeError is raised if SERVER_SIDE_PROGRAM_STORAGE_PATH environment variable is not set. –

Return type:

Union[str, TypeVar(T)]

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, the file will be created and all requests the program sends and their corresponding answers will be recorded in said file. If the file already exists, responses will be replayed from the tracefile. Note that if the program makes requests not covered by the tracefile, those requests will be sent to the server and appended to the tracefile. TRACEFILE must be a file path within the directory ~/tmp/check_mk/debug. Note that this feature is for debugging purposes only and might change in the future.

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