Handlers

Handlers in MatrixCtl are used to handle the communication between the server and the Commands or to load config files.

YAML

Read and parse the configuration file with this module.

class matrixctl.handlers.yaml.JinjaUndefined(hint=None, obj=missing, name=None, exc=<class 'jinja2.exceptions.UndefinedError'>)[source]

Bases: Undefined

Use this class as undefined argument in a Jinja2 Template.

The class replaces every undefined template with an enpty string.

class matrixctl.handlers.yaml.YAML(paths=None, server=None)[source]

Bases: object

Use the YAML class to read and parse the configuration file(s).

DEFAULT_PATHS: ClassVar[list[Path]] = [PosixPath('/etc/matrixctl/config'), PosixPath('/home/runner/.config/matrixctl/config')]

JINJA_PREDEFINED: ClassVar[dict[str, str | int]] = {'default_api_concurrent_limit': 4, 'default_ssh_port': 22, 'home': '/home/runner', 'user': 'runner', 'well_knowen_path': '.well-known/openid-configuration'}

api_auth_prepared: bool

static apply_defaults(config, server)[source]

Apply defaults to the configuration.

Return type:

Config

Parameters:

configmatrixctl.structures.Config: The configuration.
serverstr: The selected server.

Returns:

servermatrixctl.structures.Config: The altered configuration.

ensure_api_auth()[source]

Ensure the API authentication configuration is valid and prepared.

This method checks the authentication type specified in the configuration and validates that all required fields for the selected authentication method are present. It supports ‘token’ and ‘oidc’ authentication types.

For ‘token’ authentication, it verifies the presence of ‘auth_token’, ‘username’, and ‘token’. For ‘oidc’ authentication, it checks for required OIDC endpoints and credentials, and if necessary, fetches OIDC configuration from a discovery endpoint. It also initializes the TokenManager and retrieves payload and user information.

Return type:

None

Raises:

ConfigFileError: If required configuration fields are missing or invalid.
ValueError: If the OIDC authorization endpoint is not found.

get(*keys, or_else=Ellipsis)[source]

Get a value from a config entry safely.

Usage

Pass strings, describing the path in the self.__yaml dictionary. Let’s say, you are looking for the synapse path:

Return type:

Any

Parameters:

*keysstr: A tuple of strings describing the values you are looking for.

Returns:

answerany: The value of the entry you described.

Examples

from matrixctl.handlers.yaml import YAML

yaml: YAML = YAML()
port: int = yaml.get("server", "ssh", "port")
print(port)
# Output: 22

get_api_token()[source]

Retrieve the API token for authentication.

This method ensures API authentication is set up, then retrieves the token based on the configured authentication type. Supports ‘token’ and ‘oidc’ auth types.

Return type:

str

Returns:

tokenstr: The API token string.

Raises:

ShouldNeverHappenError: If the token manager is not initialized or the auth type is unknown.

get_api_username(*, full=False)[source]

Retrieve the API token for authentication.

This method ensures API authentication is set up, then retrieves the username based on the configured authentication type. Supports ‘token’ and ‘oidc’ auth types.

Return type:

str

Parameters:

fullbool, optional: If True, returns the full user identifier. If False (default), it only returns the localpart.

Returns:

usernamestr: The username.

Raises:

ShouldNeverHappenError: If the token manager is not initialized or the auth type is unknown.

static get_paths_to_config()[source]

Generate a tuple of path which may contain a configuration file.

Note

This function preserves the order. The priority of the user configuration in XDG_CONFIG_HOME is higher than the global configuration in /etc/matrixctl/. The priority of the file extension yaml is greater than the priority of the file extension yml.

Warning

The paths returned by this function might not exist.

Returns:

config_pathstuple of pathlib.Path: A tuple of paths, which might contain a config file.

rtype:: tuple[Path, ...] ..

get_room_alias(alias)[source]

Retrieve the room ID associated with a given alias name.

Return type:

str | None

Parameters:

aliasstr: The name of the alias to look up.

Returns:

room_idstr, optional: The room ID if found, otherwise None.

Raises:

InternalResponseError: If multiple aliases with the same name are found.

get_server_config(paths, server)[source]

Read and concentrate the config in one dict.

The servers: ... will be removed form the dict. A new entry server will be created, which represents the selected server.

Return type:

Config

Parameters:

pathsIterable of pathlib.Path: The paths to the configfiles.
serverstr: The selected server. (Default: “default”)

Returns:

server_configmatrixctl.typehints.Config: The config for the selected server.

Notes

When all files were empty or don’t exist, an empty dict will be returned.

static read_from_file(yaml, path)[source]

Read the config from a YAML file and render the Jinja2 tmplates.

Note

The Renderer does one pass. This means, you can only render templated strings but not the templated string of another templated string.
If the file was empty or does not exist, an empty dict will be returned.

Parameters:

yamlruamel.yaml.Yaml: The yaml object.
pathPath: The path where the config file is located.

Returns:

full_configmatrixctl.typehints.Config: The full (with server name) config file as dict.

rtype:: Config ..

server: str

token_manager: TokenManager | None

matrixctl.handlers.yaml.secrets_filter(tree, key)[source]

Redact secrets when printing the configuration file.

Return type:

Any

Parameters:

treedict [str, str]: A patrial of tree from tree_printer. (Can only be this type) afterwards.
keystr: A dict key. (Can only be this type)

Returns:

None

matrixctl.handlers.yaml.tree_printer(tree, depth=0)[source]

Print the configuration file recursively.

Return type:

None

Parameters:

treeany: Initial a matrixctl.typehints.Config and partials of it afterwards.
depthint: The depth of the table

Returns:

None

OIDC

An OIDC Client for Matrix Authentication Service.

If you are reading this you might probably ask yourself: “why yet another OIDC client?”. The answer is unfortunately. It was easier to write one from scratch (for this specific usecase) than to use any of the existing ones I found. Either the documentation was in a devastating state or they were unmaintained for years. It’s pretty sad.

Nevertheless, this should not stop us from having nice things, too. So here we are with yet another OIDC client.

class matrixctl.handlers.oidc.OidcTCPServer(server_address, RequestHandlerClass)[source]

Bases: TCPServer

TCP server wrapper for handling OIDC authentication callbacks.

This server listens for incoming HTTP requests containing the OIDC authorization code and stores it for later retrieval.

class matrixctl.handlers.oidc.TokenManager(token_endpoint, client_id, client_secret, auth_endpoint=None, userinfo_endpoint=None, jwks_uri=None, cache_path=None)[source]

Bases: object

Manager for OIDC tokens handling.

It supports both client credentials and authorization code flows.

Parameters:

token_endpointstr: OIDC provider’s token endpoint URL
client_idstr: Client ID for OIDC authentication
client_secretstr: Client secret for OIDC authentication
auth_endpointstr | None, optional: Authorization endpoint URL for user authentication flow, by default None
userinfo_endpointstr | None, optional: Userinfo endpoint URL, by default None
jwks_uristr | None, optional: JWKS endpoint URL, by default None
cache_pathstr, optional: Path to token cache file, by default “~/.oidc_token_cache.json”

get_client_credentials_token()[source]

Get access token using client credentials flow.

Return type:

str

Returns:

str: Valid access token

Raises:

httpx.HTTPStatusError: For HTTP request failures
ValueError: If token response is invalid

get_payload()[source]

Decode payload from the ID token.

Return type:

dict[str, Any]

Returns:

dict[str, Any]: Decoded ID token payload

Raises:

ValueError: If no ID token available

get_user_info()[source]

Retrieve user information from the userinfo endpoint.

Return type:

dict[str, Any]

Returns:

dict[str, Any]: User claims dictionary

Raises:

ValueError: If userinfo endpoint not configured or no access token
httpx.HTTPStatusError: For HTTP request failures

get_user_token(claims)[source]

Get access token using authorization code flow with PKCE.

Return type:

str

Returns:

str: Valid access token

Raises:

TimeoutError: If user doesn’t complete authentication within 5 minutes
httpx.HTTPStatusError: For HTTP request failures
ValueError: If token response is invalid

recall_cached_token(key)[source]

Load and validate cached tokens from disk.

Return type:

bool

Returns:

bool: True if valid token was loaded, False otherwise

Notes

Sets instance attributes: - access_token: Loaded access token or None - refresh_token: Loaded refresh token or None - expires_at: Expiration timestamp or 0

refresh_access_token()[source]

Refresh access token using refresh token.

Return type:

str | None

Returns:

str | None: New access token if successful, None otherwise

store_cache_token(access_token, refresh_token, id_token, expires_in, key)[source]

Cache tokens to disk with expiration information.

Return type:

None

Parameters:

access_tokenstr: New access token to cache
refresh_tokenstr | None: Optional refresh token to cache
expires_inint: Time in seconds until token expiration

Notes

Updates instance attributes: - access_token - refresh_token - expires_at

wait_for_auth_code_timeout: int = 300

matrixctl.handlers.oidc.discover_oidc_endpoints(issuer_url)[source]

Retrieve OIDC provider configuration via discovery.

Return type:

dict[str, Any]

Parameters:

issuer_urlstr: Base URL of the OIDC issuer

Returns:

dict[str, t.Any]: OIDC provider configuration

Raises:

httpx.HTTPStatusError: For HTTP request failures
ValueError: If discovery document is invalid

API

Get access to the API of your homeserver.

class matrixctl.handlers.api.RequestBuilder(token, domain, path, scheme='https', subdomain='matrix', data=None, json=None, content=None, method='GET', params=NOTHING, headers=NOTHING, concurrent_limit=4, timeout=5.0, success_codes=(200, 201, 202, 203, 204, 205, 206, 207, 226))[source]

Bases: object

Build the URL for an API request.

concurrent_limit: int

content: str | bytes | Iterable[bytes] | None

data: dict[str, Any] | None

domain: str

headers: dict[str, str]

property headers_with_auth: dict[str, str]

Get the headers with bearer token.

Parameters:

None

Returns:

headersdict [str, str]: Headers with auth. token.

json: dict[str, Any] | None

method: str

params: dict[str, str | int]

path: str

scheme: str

subdomain: str

success_codes: tuple[int, ...]

timeout: float

token: str

class matrixctl.handlers.api.RequestStrategy(limit: int, step_size: int, concurrent_limit: int, offset: int, iterations: int)[source]

Bases: NamedTuple

Use this NamedTuple as request strategy data.

This NamedTuple is only used in this module.

concurrent_limit: int: Alias for field number 2

iterations: int: Alias for field number 4

limit: int: Alias for field number 0

offset: int: Alias for field number 3

step_size: int: Alias for field number 1

async matrixctl.handlers.api.async_worker(input_queue, output_queue)[source]

Use this coro as worker to make (a)synchronous request.

Return type:

None

Attributes:

input_queueasyncio.Queue: The input queue, which provides the RequestBuilder.
output_queueasyncio.Queue: The output queue, which gets the responses of there requests.

Returns:

None

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

matrixctl.handlers.api.download_media_to_buf(token, domain, media_id)[source]

Make a (a)synchronous request to the synapse API and receive a response.

Return type:

bytes

Attributes:

tokenstr: The token to authenticate against the homeserver’s API.
domainstr: The domain of the homeserver.
media_idstr: The media ID

Returns:

bufbytes: A buffer containing the raw media data.

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

async matrixctl.handlers.api.exec_async_request(request_config)[source]

Use this coro to generate and run workers and group the responses.

Return type:

Response | list[Response]

Attributes:

request_configRequestBuilder or Generator [RequestBuilder, None, None]: An instance of an RequestBuilder or a list of RequestBuilder. If the function gets a RequestBuilder, the request will be synchronous. If it gets a Generator, the request will be asynchronous.
concurrent_limitint: The maximum of concurrent workers. (This information must be pulled from the config.)

Returns:

responseslist of httpx.Response or httpx.Response: Depending on concurrent_limit an request_config.

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

matrixctl.handlers.api.generate_worker_configs(request_config, next_token, limit)[source]

Create workers for async requests (minus the already done sync request).

Return type:

Generator[RequestBuilder, None, None]

Attributes:

request_configmatrixctl.handlers.api.RequestBuilder: An instance of an RequestBuilder from which was used for an initial synchronous request to get the first part of the data and the other two arguments from the response.
next_tokenint: The value, which defines from where to start in the next request. You get this value from the response of an initial synchronous request.
totalint: The value which defines how many entries there are. You get this value from the response of an initial synchronous request.

Yields:

request_configmatrixctl.handlers.api.RequestBuilder: Yields a fully configured RequestsBuilder for every request that has to be done to get all entries.

Notes

Warning Call-By-Reference like behavior! The param limit and the concurrent_limit in request_config will get changed in this function. Make sure to only use them after using this function!

async matrixctl.handlers.api.group_async_results(input_size, output_queue)[source]

Use this coro to group the requests afterwards in a single list.

Return type:

list[Exception | Response]

Attributes:

input_sizeint: The number of items in the queue.
output_queueasyncio.Queue: The output queue, which holds the responses of there requests.
concurrentbool: When True, make requests concurrently. When False, make requests synchronously.

Returns:

responseslist of httpx.Response or httpx.Response: Depending on concurrent, it is a httpx.Response if concurrent is true, otherwise it is a list of httpx.Response.

matrixctl.handlers.api.handle_sync_response_status_code(response, success_codes=None)[source]

Handle the response status code of a synchronous request.

Return type:

None

Attributes:

responsehttps.Response: The response of the synchronous request.
success_codestuple of int, optional: A tuple of success codes. For example: 200, 201, 202, 203. If the parameter is not set, the default success codes are used.

Returns:

None: The function either returns None or raises an exception.

matrixctl.handlers.api.preplan_request_strategy(limit, concurrent_limit, max_step_size=100)[source]

Use this functiona as helper for optimizing asynchronous requests.

Return type:

RequestStrategy

Attributes:

limitint: A user entered limit or total.
concurrent_limit: int: The concurrent limit from the config file.
max_step_sizeint, default=100: The maximal step size, which is a soft limit. It is usually 100, but that value might be different. Check out the API documentation. We usually take the default one.

Returns:

RequestStrategymatrixctl.handlers.api.RequestStrategy: A Named tuple with the RequestStrategy values.

matrixctl.handlers.api.request(request_config)[source]

Make a (a)synchronous request to the synapse API and receive a response.

Return type:

list[Response] | Response

Attributes:

request_configRequestBuilder or Generator [RequestBuilder, None, None]: An instance of an RequestBuilder or a list of RequestBuilder. If the function gets a RequestBuilder, the request will be synchronous. If it gets a Generator, the request will be asynchronous.
concurrent_limitint: The maximum of concurrent workers. (This information must be pulled from the config.)

Returns:

responsehttpx.Response: Returns the response

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

matrixctl.handlers.api.streamed_download(request_config, download_path)[source]

Make a (a)synchronous request to the synapse API and receive a response.

Return type:

None

Attributes:

request_configRequestBuilder or Generator [RequestBuilder, None, None]: An instance of an RequestBuilder or a list of RequestBuilder. If the function gets a RequestBuilder, the request will be synchronous. If it gets a Generator, the request will be asynchronous.
concurrent_limitint: The maximum of concurrent workers. (This information must be pulled from the config.)

Returns:

responsehttpx.Response: Returns the response

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

Ansible

Run a ansible playbook with this module.

matrixctl.handlers.ansible.ansible_run(playbook, tags=None, extra_vars=None)[source]

Run an ansible playbook.

Return type:

None

Parameters:

playbookpathlib.Path: The path to the ansible Playbook
tagsstr, optional: The tags to use
extra_varsdict [str, str], optional: The extra_vars to use.

Returns:

None

Git (VCS)

Update and manage the synapse playbook repository with this module.

class matrixctl.handlers.vcs.VCS(path)[source]

Bases: object

Update and manage a repository.

property datetime_last_pulled_commit: datetime

Get the datetime the commit was pulled last from git.

This is used to determine which messages will be produced in the table.

Parameters:

None

Returns:

datetimedatetime.datetime: The datetime object.

log(since=None)[source]

Print a table of date, user and commit message since the last pull.

Return type:

None

Parameters:

sincedatetime.datetime, optional, default=None: The datetime the last commit was puled.

Returns:

None

pull()[source]

Git pull the latest commits from GH.

Return type:

None

Parameters:

None

Returns:

None

SSH

Run and evaluate commands on the host machine of your synapse server.

class matrixctl.handlers.ssh.SSH(address, user=None, port=22)[source]

Bases: object

Run and evaluate commands on the host machine of your synapse server.

address: str

port: int

run_cmd(cmd)[source]

Run a command on the host machine and receive a response.

Return type:

SSHResponse

Parameters:

cmdstr: The command to run.
ttybool: Request a pseudo-terminal from the server (default: False)

Returns:

responsematrixctl.handlers.ssh.SSHResponse: Receive stdin, stdout and stderr as response.

user: str

class matrixctl.handlers.ssh.SSHResponse(stdin: str | None, stdout: str | None, stderr: str | None)[source]

Bases: NamedTuple

Store the response of a SSH command as response.

stderr: str | None: Alias for field number 2

stdin: str | None: Alias for field number 0

stdout: str | None: Alias for field number 1

Table

Use this handler to generate and print tables.

matrixctl.handlers.table.cells_to_str(part, none)[source]

Convert all cells to strings and format None values.

Return type:

list[list[str]]

Parameters:

partcollections.abc.Sequence of collections.abc.Sequence of str: Data or header, in which every cell will be to casted to to strings.
nonestr: A string, which is used to replace None with the specific string.

Returns:

partlist of list of str
The part, where every cell is of type str.

matrixctl.handlers.table.find_newlines(data)[source]

Find newlines and return a dict with positions (key) and occurrences.

Return type:

dict[int, int]

Parameters:

datalist of str: Data or headers of the table.

Returns:

posdict [int, int]: A dictionary {r: n}, where n are newlines in row r.

Notes

The function only adds an entry to the dict, if there is at least one newline in a row.

matrixctl.handlers.table.format_table_row(line, max_column_len)[source]

Format a table row into a str.

Return type:

str

Parameters:

linelist of str: A data or headers row, which will be formatted to a string.
max_column_lentuple of int: A n-tuple which describes the longest string per column. (n is the number of columns)

Returns:

row_stringstr: A formatted string, which represents a table row.

matrixctl.handlers.table.get_colum_length(data, headers)[source]

Transpose rows and find longest line.

Return type:

tuple[int, ...]

Parameters:

datalist of list of str: The data part of the table.
headersNone or list of list of str: The headers part of the table.

Returns:

column_length_tupletuple of int: A n-tuple which describes the longest string per column. (n is the number of columns)

matrixctl.handlers.table.handle_newlines(part, newlines)[source]

Update and insert new lines.

Return type:

tuple[list[list[str]], set[int]]

Parameters:

partlist of list of str: Data or headers of the table.
newlinesdict [int, int]: A dictionary {r: n}, where n are newlines in row r.

Returns:

partlist of list of str: The part contains the supplemented and updated rows.
inhibit_sepset of int: The inhibit_sep set contains the line numbers where a separator yhould be inhibited because the lines handled by this function are belonging together.

matrixctl.handlers.table.newlines_in_row(row)[source]

Get the highest number of newlines per row.

The highest number of newlines for a row is used to determine in how many rows the row gets expanded, to get one row per newline - 1.

Return type:

int

Parameters:

rowlist of str: Data or headers of the table.

Returns:

max_newlinesint: The highest number of newlines ber row.

matrixctl.handlers.table.table(table_data, table_headers=None, none='-', *, sep=True)[source]

Create a table from data and a optional headers.

Return type:

Generator[str, None, None]

Parameters:

table_datacollections.abc.Sequence of collections.abc.Sequence of str: Data.
table_headerscollections.abc.Sequence of str, Optional: Headers.
sepbool, default = True: True, when there should be a separator between every row of data.
nonestr, default = “-”: A string, which is used to replace None with the specific string.

Yields:

tablestr: The table (row for row).

matrixctl.handlers.table.transpose_newlines_to_rows(split, occurrences)[source]

Transpose newlines in new rows.

Return type:

Generator[list[str], None, None]

Parameters:

splitlist of list of str: A list of substring-lists, split from one row, which contains newline characters. The substing-lists are containing strings, which have been split into substings.
occurrencesint: The maximal number of newlines across the row.

Yields:

rowlist of str: A row for each occurrence.

Database

Talk to the the database.

class matrixctl.handlers.db.DBConnectionBuilder(host: str, database: str, username: str, password: str, port: int = 5432, timeout: int = 10, scheme: str = 'postgresql')[source]

Bases: NamedTuple

Build the URL for an API request.

database: str: Alias for field number 1

host: str: Alias for field number 0

password: str: Alias for field number 3

port: int: Alias for field number 4

scheme: str: Alias for field number 6

timeout: int: Alias for field number 5

username: str: Alias for field number 2

matrixctl.handlers.db.db_connect(yaml)[source]

Connect to a PostgreSQL database.

Return type:

Iterator[Connection]

Parameters:

yamlmatrixctl.handlers.yaml.YAML: The configuration file handler.

Yields:

connpsycopg.Connection: A new Connection instance.

matrixctl.handlers.db.ssh_tunnel(host, username, remote_port, port=22, *, enabled=True)[source]

Create an SSH tunnel.

Return type:

Iterator[int | None]

Parameters:

hoststr: The remote host e.g. 127.0.0.1 or host.domain.tld.
usernamestr: The username of the user.
remote_portint: The port of the application, which should be tunneled.
enabledbool, default: True: True if the tunnel should be enabled or False if not.
portint, default: 22: The ssh port
private_keyPath or str, optioal: The path to the private key (Currently Disabled)

Yields:

tunint: The remote port
NoneNone: Yields none, when the tunnel is disabled (enabled = False).

Notes

The tunnel will only be created, when it is enabled. If the tunnel is disabled (enabled = False), the function will yield None instead of the local bind port.

Examples

with ssh_tunnel("127.0.0.1", myuser, 5432) as remote_port:
    print(f"The local bind port is: {local_bind_port}")
# The local bind port is: 8765