utils package

Subpackages

Submodules

utils.ansible_log module

SPOT for the line prefix Ansible writes into its log_path file.

Every line of the file named by ANSIBLE_LOG_PATH carries ``<ts> p=<pid> u=<user> n=<name> <LEVEL>| `` in front of the message that also went to stdout. Left unstripped, that prefix defeats every line-anchored matcher a reader applies, so a reader that mixes an anchored match with an unanchored one silently sees an empty log instead of failing.

utils.api module

Proprietary API credential SPOT, read from group_vars/all/18_api.yml.

API is a plain mapping, so an inventory that configures three providers replaces the whole declaration instead of merging into it, and every provider it does not name disappears. Resolving a hidden provider then raises, and a raising lookup inside a value the applications renderer is templating leaves the raw {{ ... }} text in place: the consumer stores template text, and | bool coerces it to False with a deprecation warning that fails the deploy on the warning gate.

Merging the scope’s override over the declared defaults keeps every declared provider resolvable, so an override narrows values, never the provider set.

utils.api.declared_api() → dict[str, Any]

The API mapping as declared in the group_vars SPOT.

Returns:

Provider name to credential mapping, every value a plain string.

Raises:

TypeError – the SPOT does not declare API as a mapping.

utils.api.provider_enabled(api: Mapping[str, Any], provider: str) → bool

Whether every credential of a provider carries a value.

Parameters:

  • api – an effective API mapping, as returned by resolve_api.

  • provider – the provider name, e.g. github.

Returns:

True when the provider declares at least one credential and none of them is empty.

Raises:

KeyError – the provider is not declared in the SPOT.

utils.api.resolve_api(variables: Mapping[str, Any] | None, templar: Any = None) → dict[str, Any]

The effective API mapping for a render scope.

Parameters:

  • variables – the scope’s variables; its API entry overrides the SPOT.

  • templar – optional templar; renders the override and the merged values, so an inventory may source a credential from an env lookup.

Returns:

The SPOT declaration with the scope’s override merged over it.

Raises:

TypeError – the scope’s API is present but not a mapping.

utils.cert_utils module

class utils.cert_utils.CertUtils

Bases: object

classmethod build_snapshot(cert_base_path)
classmethod ensure_cert_mapping(cert_base_path, debug=False)
static extract_sans(cert_text)
classmethod find_cert_for_domain(domain, cert_base_path, debug=False)
static list_cert_files(cert_base_path)
static matches(domain, san)

RFC compliant SAN matching.

classmethod refresh_cert_mapping(cert_base_path, debug=False)

Build mapping: SAN -> list of entries entry = {

‘folder’: str, ‘cert_path’: str, ‘mtime’: float, ‘not_before’: int|None, ‘not_after’: int|None, ‘is_wildcard’: bool

}

static run_openssl(cert_path)
static run_openssl_dates(cert_path)

Returns (not_before_ts, not_after_ts) as POSIX timestamps or (None, None) on failure.

classmethod snapshot_changed(cert_base_path)

utils.distros module

Distribution SPOT.

meta/distros.yml declares every distribution the project builds, tests and deploys on: the id used as INFINITO_DISTRO, the os_family ansible reports for it, the Galaxy platform name, and the templates the pkgmgr base image and the environment image references are rendered from. The upstream image the development-environment matrix boots each distro from is pinned in roles/sys-os/meta/services.yml under the distro id, where the docker updater, the mirror and the vulnerability scan find it.

Consumers MUST read the distro set through this module. A local tuple of distro names or an f-string spelling out ghcr.io/.../pkgmgr-... is a second source of truth and drifts; tests/lint/repository/test_distros_spot.py guards against reintroducing one.

Registry owner, repository and tag are not declared here — they stay in default.env, which is their SPOT. This module only renders them into the templates.

Keep this module free of Ansible imports and cheap to import: it is pulled in by the env layer and by utils.packages.schema, which reach contexts where Ansible is not installed. It must also stay importable on the bare bootstrap python, so the SPOT is read with the stdlib block reader instead of PyYAML.

exception utils.distros.UnknownDistroError

Bases: ValueError

Raised for a distro id or image that meta/distros.yml does not declare.

utils.distros.dev_runtime_image(distro: str) → str

Upstream base image the development environment boots a distro from.

utils.distros.dev_runtime_images() → tuple[str, ...]

Upstream base images of the development-environment test matrix.

utils.distros.distro_family() → dict[str, str]

Map each distro id onto the os_family ansible reports for it.

utils.distros.distro_names() → tuple[str, ...]

Every declared distro id, in declaration order.

utils.distros.distro_of_dev_runtime_image(image: str) → str

Distro id whose development environment boots from image.

utils.distros.environment_image(distro: str, owner: str, repository: str, tag: str) → str

Reference of the environment image built for a distro.

utils.distros.galaxy_platforms() → list[dict[str, object]]

Canonical galaxy_info.platforms block for every role’s meta/main.yml.

Sorted by platform name, which is the order the role files carry and which the schema lint compares against verbatim.

utils.distros.image_template(kind: str) → str

Raw reference template, for callers rendering a non-distro slug.

The shell-side installer scripts loop over distros with their own variable, so they render {slug} themselves rather than per distro.

utils.distros.pkgmgr_image(distro: str, owner: str, tag: str) → str

Reference of the pkgmgr base image a distro’s container builds FROM.

utils.distros.pkgmgr_virgin_image(distro: str, owner: str, tag: str) → str

Reference of the untouched pkgmgr image the install tests boot.

utils.get_url module

utils.get_url.get_url(domains, application_id, protocol)

utils.kcadm_json module

utils.kcadm_json.json_from_noisy_stdout(text: Any) → Any

Parse JSON from output that may contain leading non-JSON noise, including lines that start with ‘[’ (e.g. JVM warnings like “[0.001s][warning]…”).

Deterministic strategy:

  • scan for all occurrences of ‘[’ and ‘{’ in the output

  • attempt json.loads() starting from each occurrence (in order)

  • return the first successfully parsed JSON value

  • raise ValueError if none work

utils.keycloak_scopes module

Client-scope list convergence for keycloak_kcadm_update.

A client PUT ignores default/optionalClientScopes (sub-resources with their own endpoints; verified against a live deploy) — these helpers converge them explicitly: DELETE unassigns, bodyless PUT assigns, removals first so a scope can move between the two lists.

run_kcadm is injected by the calling module (it owns retry/CID-recovery semantics); every helper takes it as its first argument.

utils.keycloak_scopes.converge_client_scope_lists(run_kcadm, module, object_id, desired, realm, kcadm_exec)

Converge a client’s default/optional client-scope lists to desired.

utils.keycloak_scopes.named_id_map(run_kcadm, module, cmd)

Run a kcadm list command and return a {name: id} map (empty on noise).

utils.parse_trace module

utils.paths module

Filesystem SPOT for the Python side, read from group_vars/all/05_paths.yml.

group_vars/all/05_paths.yml is the single source of truth for host paths. The env layer (utils/env/handlers, exported by scripts/meta/env/load.sh) derives the INFINITO_* env keys from it; a set env wins so callers can override. When a tool runs outside that layer, e.g. a bare subprocess that does not inherit the env build, the value is read straight from the group_vars SPOT, so there is still no hardcoded literal default.

The SPOT is parsed with a plain-string line parser instead of PyYAML: the env build runs during bootstrap (make install-system-python) before any Python dependency exists, so this module must stay stdlib-only. Only plain KEY: "value" entries can be read this way; a Jinja expression or a missing key is a hard error.

utils.paths.read_group_path(key: str) → str

Plain-string path value from the group_vars paths SPOT.

Parameters:

key – top-level variable name in group_vars/all/05_paths.yml.

Returns:

The literal string value.

Raises:

  • KeyError – the key is not defined in the SPOT.

  • ValueError – the value is not a plain string (e.g. a Jinja template).

utils.paths.read_group_paths(key: str) → list[str]

Plain-string list value from the group_vars paths SPOT.

Parameters:

key – top-level variable name whose value is a YAML block sequence.

Returns:

The literal string entries, in file order.

Raises:

KeyError – the key is not defined in the SPOT or carries no entry.

utils.sizes module

Parse human-readable memory size strings to bytes.

SI/decimal units for the bare suffixes (k=10^3, m=10^6, g=10^9, …) and IEC binary for the explicit *ib suffixes (kib=1024, mib=1024^2, …). Shared by the node_max_old_space_size lookup and the peertube install_mem_limit filter.

utils.sizes.to_bytes(val)

Convert a numeric or string size (e.g. ‘512m’, ‘2GiB’, 1024) to bytes.

Returns None for None/empty input. Raises AnsibleFilterError on an unparseable value.

utils.symbol_glossary module

Single source of truth mapping short words to display emojis, and back.

Every emoji used as a STRUCTURED marker across the project (complexity-matrix columns, lifecycle stages, deploy modes, boolean/state cells) is defined here exactly once, so a glyph means the same thing everywhere it appears. Emojis are unique: the module fails to import if two words share a glyph.

Look up with to_emoji() (word -> emoji) and to_word() (emoji -> canonical word); both pass unknown input through unchanged.

utils.symbol_glossary.to_emoji(word: str) → str

Return the emoji mapped to word, or word itself when unmapped.

utils.symbol_glossary.to_word(emoji: str) → str

Return the canonical word for emoji, or emoji itself when unmapped.

utils.terminal_aliases module

Fetch-once cache and parser for the operator’s terminal aliases.

Configuration comes exclusively from the generated .env (make dotenv), which already folds in default.env, every handler, and the operator’s custom.env overrides. The alias repository (INFINITO_ALIAS_REPOSITORY) holds one alias <name>='<command>' line per shortcut in an aliases file at its root.

utils.terminal_aliases.alias_repository() → str

Effective terminal alias repository from the generated .env.

utils.terminal_aliases.fetch_cached(repository: str, timeout: int = 15) → str

Return the aliases file content, downloading once per repository.

Parameters:

  • repository – git repository URL to fetch the aliases file from.

  • timeout – socket timeout in seconds for the one-time download.

utils.terminal_aliases.local_alias_names() → list[str]

Infinito.Nexus-specific alias names from the repo’s aliases file.

utils.terminal_aliases.parse_alias_names(text: str) → list[str]

Alias names in file order.

Parameters:

text – raw content of an aliases file.

utils.terminal_aliases.raw_aliases_url(repository: str) → str

Raw download URL of the repository’s root aliases file.

Parameters:

repository – git repository URL (GitHub-style raw path layout).

utils.tls_common module

utils.tls_common.align_domain_to_consumer(domains: dict, app_id: str, primary_domain: str, *, consumer: str = '', variables: dict | None = None, templar: Any = None) → str

Family-align a cross-app reference: when a clearnet-primary consumer resolves an onion-primary target, return the target’s first clearnet domain from the merged map; keep primary_domain whenever the consumer is unknown/onion, the target is clearnet, or no clearnet sibling exists.

utils.tls_common.as_str(value: Any) → str
utils.tls_common.collect_domains_for_app(domains: dict, app_id: str, *, err_prefix: str) → list[str]
utils.tls_common.collect_domains_global(domains: dict) → list[str]
utils.tls_common.get_path(data: Any, path: str, default: Any = None) → Any
utils.tls_common.is_onion_domain(domain: Any) → bool

True if the domain is a Tor v3 .onion address.

utils.tls_common.iter_domains(value: Any) → Iterable[str]

Flatten domains from the global domains mapping entry. Supports:

  • str

  • list[str]

  • dict[str, str] (only one level, legacy behavior)

utils.tls_common.norm_domain(value: Any) → str
utils.tls_common.override_san_list(app: dict) → list[str] | None
utils.tls_common.require(variables: dict, name: str, expected_type: type | tuple[type, ...]) → Any
utils.tls_common.resolve_app_id_from_domain(domains: dict, domain: str, *, err_prefix: str) → str

Legacy reverse-lookup from the global domains mapping. This does NOT know about per-app aliases/canonicals stored in applications.

utils.tls_common.resolve_enabled(app: dict, enabled_default: bool, *, primary_domain: str = '') → bool
utils.tls_common.resolve_le_name(app: dict, domain: str) → str
utils.tls_common.resolve_mode(app: dict, enabled: bool, mode_default: str, *, err_prefix: str) → str
utils.tls_common.resolve_primary_domain_from_app(domains: dict, app_id: str, *, err_prefix: str) → str
utils.tls_common.resolve_term(term: str, *, domains: dict, applications: dict | None = None, forced_mode: str, err_prefix: str) → tuple[str, str]

Returns (app_id, primary_domain) where primary_domain is normalized lower-case.

term can be:

  • application_id

  • domain (canonical or alias)

forced_mode: “auto” | “domain” | “app”

Behavior:

  1. If term is a domain and exists in global domains mapping -> primary_domain = that term (normalized).

  2. If term is a domain and only exists in applications[*].domains -> map to app_id, and primary_domain = canonical primary from global domains mapping (first entry).

utils.tls_common.uniq_preserve(items: Iterable[str]) → list[str]
utils.tls_common.want_get(data: Any, dotted: str) → Any

utils.tor_onion module

Offline Tor v3 onion-service key minting.

Deterministically derives a Tor v3 hidden-service address and the exact HiddenServiceDir key files from an ed25519 seed, without running Tor.

The v3 address is base32(PUBKEY || CHECKSUM || VERSION) where CHECKSUM = SHA3-256(".onion checksum" || PUBKEY || VERSION)[:2] and VERSION = 0x03 (see spec.torproject.org/address-spec). Tor stores the expanded ed25519 secret key clamp(SHA512(seed)[:32]) || SHA512(seed)[32:] in hs_ed25519_secret_key; the public key clamp(SHA512(seed)[:32]) * B equals the standard ed25519 public key, so it is taken straight from cryptography without needing scalar multiplication.

class utils.tor_onion.OnionKey(seed: bytes, address: str, hostname: bytes, public_key: bytes, secret_key: bytes)

Bases: object

A minted Tor v3 onion key and its HiddenServiceDir file contents.

address: str
files() → dict[str, bytes]

Map HiddenServiceDir filenames to their exact byte contents.

hostname: bytes
public_key: bytes
secret_key: bytes
seed: bytes
utils.tor_onion.identity_hs_dir(base_dir: str | Path) → Path

Authoritative hidden-service key dir (<base_dir>/.onion-identity/hs).

Single source of truth for the node’s onion identity: ensure_node_onion mints/reuses the key files here and the svc-net-tor role copies them into the running daemon. Lives outside anything the deploy regenerates and is mounted into the DiD at {playbook_dir}/.onion-identity/hs.

utils.tor_onion.mint(seed: bytes | None = None) → OnionKey

Mint a v3 onion key from seed (32 random bytes if omitted).

utils.tor_onion.mint_from_seed_b64(seed_b64: str) → OnionKey

Mint from a base64-encoded seed.

utils.tor_onion.onion_address(public_key: bytes) → str

Return the v3 .onion address for a 32-byte ed25519 public key.

utils.yaml_bootstrap module

Stdlib reader for the block-YAML subset the bootstrap SPOT files use.

This is a deliberate second YAML touchpoint outside utils.cache.yaml: the modules that read the distro SPOT must import on the bare bootstrap python, which runs the .env generator before any dependency is installed, so they cannot reach the PyYAML-backed cache.

Supported: nested block mappings, block sequences of scalars, plain and double-quoted string scalars, full-line and trailing comments, a single leading document marker. Every other construct raises BootstrapYamlError, so load_block(text) either equals yaml.safe_load(text) or fails loud.

The plain-scalar rule is deliberately narrower than YAML’s: a plain scalar must start with a letter, an underscore or a slash, and may then carry only [A-Za-z0-9_./+@:-] and spaces. That excludes every leading character PyYAML’s implicit resolvers key off, so no accepted plain scalar can come back as an int, float, bool, timestamp or null.

exception utils.yaml_bootstrap.BootstrapYamlError

Bases: ValueError

Raised for a construct outside the supported block subset.

utils.yaml_bootstrap.load_block(text: str) → dict

Parse text as a block mapping and return it.

Parameters:

text – YAML source restricted to the supported block subset.

Returns:

The root mapping, with every scalar as a str.

Raises:

BootstrapYamlError – for any construct outside the subset.

Module contents