filter_plugins package

Submodules

filter_plugins.active_docker module

Ansible filter to count active docker services for current host.

Active means: - application key is in group_names - application key matches prefix regex (default: ^(web-|svc-).* ) - under applications[app][‘docker’][‘services’] each service is counted if:

• ‘enabled’ is True, OR

• ‘enabled’ is missing/undefined (treated as active)

Returns an integer. If ensure_min_one=True, returns at least 1.

class filter_plugins.active_docker.FilterModule

Bases: object

filters()

filter_plugins.active_docker.active_docker_container_count(applications: Mapping[str, Any], group_names: Iterable[str], prefix_regex: str = '^(web-|svc-).*', ensure_min_one: bool = False) → int

filter_plugins.application_allowed module

class filter_plugins.application_allowed.FilterModule

Bases: object

filters()

filter_plugins.application_allowed.application_allowed(application_id: str, group_names: list, allowed_applications: list = [])

Return True if:

• application_id exists in group_names, AND

• either allowed_applications is not provided (or empty), OR application_id is in allowed_applications.

Parameters:

• application_id (str) – Name of the application/role to check.

• group_names (list) – List of groups the current host belongs to.

• allowed_applications (list, optional) – List of application IDs to allow.

Returns:

True if this application is allowed to deploy, False otherwise.

Return type:

bool

filter_plugins.applications_if_group_and_deps module

class filter_plugins.applications_if_group_and_deps.FilterModule

Bases: object

applications_if_group_and_deps(applications, group_names)

Return only those applications whose key is either:

1. directly in group_names, or

2. the application_id of any role reachable (recursively) from any group in group_names via meta/dependencies.

filters()

filter_plugins.canonical_domains_map module

class filter_plugins.canonical_domains_map.FilterModule

Bases: object

canonical_domains_map(apps, domain_primary, *, recursive: bool = False, roles_base_dir: str | None = None, seed: Iterable[str] | None = None)

Build { app_id: [canonical domains…] }.

Rekursiv werden nur include_role, import_role und meta/main.yml:dependencies verfolgt. ‘run_after’ wird hier absichtlich ignoriert.

filters()

filter_plugins.csp_filters module

class filter_plugins.csp_filters.FilterModule

Bases: object

Jinja filters for building a robust, CSP3-aware Content-Security-Policy header. Safari/CSP2 compatibility is ensured by merging the -elem/-attr variants into the base directives (style-src, script-src). We intentionally do NOT mirror back into -elem/-attr to allow true CSP3 granularity on modern browsers.

build_csp_header(applications, application_id, domains, web_protocol='https', matomo_feature_name='matomo')

Builds the Content-Security-Policy header value dynamically based on application settings.

Key points:

• CSP3-aware: supports base/elem/attr for styles and scripts.

• Safari/CSP2 fallback: base directives (style-src, script-src) always include the union of their -elem/-attr variants.

• We do NOT mirror back into -elem/-attr; finer CSP3 rules remain effective on modern browsers if you choose to use them.

• If the app explicitly disables a token on the base (e.g. style-src.unsafe-inline: false), that token is removed from the merged base even if present in elem/attr.

• Inline hashes are added ONLY if that directive does NOT include ‘unsafe-inline’.

• Whitelists/flags/hashes read from:

  server.csp.whitelist.<directive> server.csp.flags.<directive> server.csp.hashes.<directive>

• “Smart defaults”:

  • internal CDN for style/script elem and connect

  • Matomo endpoints (if feature enabled) for script-elem/connect

  • Simpleicons (if feature enabled) for connect

  • reCAPTCHA (if feature enabled) for script-elem/frame-src

  • frame-ancestors extended for desktop/logout/keycloak if enabled

filters()

static get_csp_flags(applications, application_id, directive)

Returns CSP flag tokens (e.g., “‘unsafe-eval’”, “‘unsafe-inline’”) for a directive, merging sane defaults with app config.

Defaults:

• For styles we enable ‘unsafe-inline’ by default (style-src, style-src-elem, style-src-attr), because many apps rely on inline styles / style attributes.

• For scripts we do NOT enable ‘unsafe-inline’ by default.

static get_csp_hash(content)

Computes the SHA256 hash of the given inline content and returns a CSP token like “‘sha256-<base64>’”.

static get_csp_inline_content(applications, application_id, directive)

Returns inline script/style snippets to hash for a given directive. Accepts both scalar and list in config; always returns a list.

static get_csp_whitelist(applications, application_id, directive)

Returns a list of additional whitelist entries for a given directive. Accepts both scalar and list in config; always returns a list.

static is_feature_enabled(applications: dict, feature: str, application_id: str) → bool

Returns True if applications[application_id].features[feature] is truthy.

filter_plugins.csp_hashes module

class filter_plugins.csp_hashes.FilterModule

Bases: object

filters()

filter_plugins.csp_hashes.append_csp_hash(applications, application_id, code_one_liner)

Ensures that applications[application_id].csp.hashes[‘script-src-elem’] exists and appends the given one-liner (if not already present).

filter_plugins.docker_service_enabled module

class filter_plugins.docker_service_enabled.FilterModule

Bases: object

Custom filter to safely check if a docker service is enabled for an application_id

filters()

static is_docker_service_enabled(applications, application_id, service_name)

Returns True if applications[application_id].docker.services[service_name].enabled is truthy, otherwise returns False (even if intermediate keys are missing).

filter_plugins.domain_redirect_mappings module

class filter_plugins.domain_redirect_mappings.FilterModule

Bases: object

domain_mappings(apps, primary_domain, auto_build_alias)

Build a flat list of redirect mappings for all apps:

• source: each alias domain

• target: the first canonical domain

Skip mappings where source == target, since they make no sense.

filters()

filter_plugins.domain_tools module

class filter_plugins.domain_tools.FilterModule

Bases: object

filters()

filter_plugins.domain_tools.to_zone(hostname: str) → str

filter_plugins.generate_all_domains module

class filter_plugins.generate_all_domains.FilterModule

Bases: object

filters()

generate_all_domains(domains_dict, include_www: bool = True)

Transform a dict of domains (values: str, list, dict) into a flat list, optionally add ‘www.’ prefixes, dedupe and sort alphabetically.

filter_plugins.generate_base_sld_domains module

class filter_plugins.generate_base_sld_domains.FilterModule

Bases: object

filters()

generate_base_sld_domains(domains_list)

Given a list of hostnames, extract the second-level domain (SLD.TLD) for any hostname with two or more labels, return single-label hostnames as-is, and reject IPs, empty or malformed strings, and non-strings. Deduplicate and sort.

filter_plugins.get_all_application_ids module

class filter_plugins.get_all_application_ids.FilterModule

Bases: object

Ansible filter plugin for retrieving application IDs.

filters()

filter_plugins.get_all_application_ids.get_all_application_ids(roles_dir='roles')

Ansible filter to retrieve all unique application_id values defined in roles/*/vars/main.yml files.

Parameters:

roles_dir – Base directory for Ansible roles (default: ‘roles’)

Returns:

Sorted list of unique application_id strings

filter_plugins.get_all_invokable_apps module

class filter_plugins.get_all_invokable_apps.FilterModule

Bases: object

filters()

filter_plugins.get_all_invokable_apps.get_all_invokable_apps(categories_file=None, roles_dir=None)

Return all application_ids (or role names) for roles whose directory names match invokable paths from categories.yml. :param categories_file: Path to categories.yml (default: roles/categories.yml at project root) :param roles_dir: Path to roles directory (default: roles/ at project root) :return: List of application_ids (or role names)

filter_plugins.get_app_conf module

exception filter_plugins.get_app_conf.AppConfigKeyError(message: str = '', obj: Any = None, show_content: bool = True, suppress_extended_error: bool | EllipsisType = Ellipsis, orig_exc: BaseException | None = None, help_text: str | None = None)

Bases: AnsibleTemplatePluginError, ValueError

Raised when a required application config key is missing (strict mode). Compatible with Ansible error handling and Python ValueError.

exception filter_plugins.get_app_conf.ConfigEntryNotSetError(message: str = '', obj: Any = None, show_content: bool = True, suppress_extended_error: bool | EllipsisType = Ellipsis, orig_exc: BaseException | None = None, help_text: str | None = None)

Bases: AppConfigKeyError

Raised when a config entry is defined in schema but not set in application.

filter_plugins.get_app_conf.get_app_conf(applications, application_id, config_path, strict=True, default=None, skip_missing_app=False)

filter_plugins.get_category_entries module

class filter_plugins.get_category_entries.FilterModule

Bases: object

Custom filters for Ansible

filters()

filter_plugins.get_category_entries.get_category_entries(prefix, roles_path='roles')

Returns a list of role names under the given roles_path that start with the specified prefix.

Parameters:

• prefix – String prefix to match role names.

• roles_path – Path to the roles directory (default: ‘roles’).

Returns:

List of matching role names.

filter_plugins.get_docker_image module

class filter_plugins.get_docker_image.FilterModule

Bases: object

filters()

filter_plugins.get_docker_image.get_docker_image(applications, application_id, image_key: str = None)

filter_plugins.get_docker_paths module

class filter_plugins.get_docker_paths.FilterModule

Bases: object

filters()

filter_plugins.get_docker_paths.get_docker_paths(application_id: str, path_docker_compose_instances: str) → dict

Build the docker_compose dict based on path_docker_compose_instances and application_id. Uses get_entity_name to extract the entity name from application_id.

filter_plugins.get_domain module

class filter_plugins.get_domain.FilterModule

Bases: object

filters()

filter_plugins.get_entity_name module

class filter_plugins.get_entity_name.FilterModule

Bases: object

filters()

filter_plugins.get_role module

Ansible filter plugin: get_role

This filter inspects each role under the given roles directory, loads its vars/main.yml, and returns the role folder name whose application_id matches the provided value.

class filter_plugins.get_role.FilterModule

Bases: object

Register the get_role filter

filters()

filter_plugins.get_role.get_role(application_id, roles_path='roles')

Find the role directory under roles_path whose vars/main.yml contains the given application_id.

Parameters:

• application_id – The application_id to match.

• roles_path – Path to the roles directory (default: ‘roles’).

Returns:

The name of the matching role directory.

Raises:

AnsibleFilterError – If vars file is unreadable or no match is found.

filter_plugins.get_service_name module

Custom Ansible filter to build a systemctl unit name (always lowercase).

Rules: - If systemctl_id ends with ‘@’: drop the ‘@’ and return

“{systemctl_id_without_at}.{software_name}@{suffix_handling}”.

• Else: return “{systemctl_id}.{software_name}{suffix_handling}”.

Suffix handling: - Default “” → automatically pick:

• “.service” if no ‘@’ in systemctl_id

• “.timer” if ‘@’ in systemctl_id

• Explicit False → no suffix at all

• Any string → “.{suffix}” (lowercased)

class filter_plugins.get_service_name.FilterModule

Bases: object

filters()

filter_plugins.get_service_name.get_service_name(systemctl_id, software_name, suffix='')

filter_plugins.get_service_script_path module

class filter_plugins.get_service_script_path.FilterModule

Bases: object

Custom filters for Ansible

filters()

filter_plugins.get_service_script_path.get_service_script_path(systemctl_id, script_type)

Build the path to a service script based on systemctl_id and type.

Parameters:

• systemctl_id – The identifier of the system service.

• script_type – The script type/extension (e.g., sh, py, yml).

Returns:

The full path string.

filter_plugins.get_url module

class filter_plugins.get_url.FilterModule

Bases: object

Infinito.Nexus application config extraction filters

filters()

filter_plugins.has_env module

class filter_plugins.has_env.FilterModule

Bases: object

filters()

filter_plugins.has_env.has_env(application_id, base_dir='.')

Check if env.j2 exists under roles/{{ application_id }}/templates/env.j2

filter_plugins.invokable_paths module

class filter_plugins.invokable_paths.FilterModule

Bases: object

filters()

filter_plugins.invokable_paths.get_invokable_paths(roles_file: str | None = None, suffix: str | None = None) → List[str]

Load nested roles YAML and return dash-joined paths where ‘invokable’ is True. Appends suffix if provided.

filter_plugins.invokable_paths.get_non_invokable_paths(roles_file: str | None = None, suffix: str | None = None) → List[str]

Load nested roles YAML and return dash-joined paths where ‘invokable’ is False or missing. Appends suffix if provided.

filter_plugins.memory_filters module

class filter_plugins.memory_filters.FilterModule

Bases: object

filters()

filter_plugins.memory_filters.jvm_max_mb(apps: dict, app_id: str) → int

Compute recommended JVM Xmx in MB using: Xmx = min(

floor(0.7 * mem_limit), mem_limit - 1024, 12288

) with a lower bound of 1024 MB.

filter_plugins.memory_filters.jvm_min_mb(apps: dict, app_id: str) → int

Compute recommended JVM Xms in MB using: Xms = min(

floor(Xmx / 2), mem_reservation, Xmx

) with a lower bound of 512 MB.

filter_plugins.memory_filters.redis_maxmemory_mb(apps: dict, app_id: str, factor: float = 0.8, min_mb: int = 64) → int

Compute recommended Redis maxmemory in MB.

• factor: fraction of allowed memory used for Redis data (default 0.8)

• min_mb: minimum floor value (default 64 MB)

maxmemory = max(min_mb, floor(factor * mem_limit))

filter_plugins.merge_mapping module

class filter_plugins.merge_mapping.FilterModule

Bases: object

filters()

filter_plugins.merge_mapping.merge_mapping(list1, list2, key_name='source')

Merge two lists of dicts on a given key. - list1, list2: each must be a List[Dict] - key_name: the field to match on If both lists contain an item with the same key_name value, their dictionaries are merged (fields from list2 overwrite or add to list1).

filter_plugins.merge_with_defaults module

class filter_plugins.merge_with_defaults.FilterModule

Bases: object

Custom merge filter for Infinito.Nexus: merge_with_defaults

filters()

filter_plugins.merge_with_defaults.merge_with_defaults(defaults, customs)

Recursively merge two dicts (customs into defaults). For each top-level key in customs, ensure all dict keys from defaults are present (at least empty dict). Customs always take precedence.

filter_plugins.node_autosize module

class filter_plugins.node_autosize.FilterModule

Bases: object

filters()

filter_plugins.node_autosize.node_max_old_space_size(applications: dict, application_id: str, service_name: str, pct: float = 0.35, min_mb: int = 768, hardcap_mb: int = 3072, safety_cap_pct: float = 0.6) → int

Derive Node.js –max-old-space-size (MB) from the service’s mem_limit in app config.

Looks up: docker.services.<service_name>.mem_limit for the given application_id.

Raises:

AnsibleFilterError if mem_limit is missing/invalid OR if mem_limit (MB) < min_mb. –

filter_plugins.redirect_filters module

class filter_plugins.redirect_filters.FilterModule

Bases: object

Custom filters for redirect domain mappings

static add_redirect_if_group(redirect_list, group, source, target, group_names)

Append {“source”: source, “target”: target} to redirect_list only if group is contained in group_names.

Usage in Jinja:

{{ redirect_list

add_redirect_if_group(‘lam’, ‘ldap.’ ~ DOMAIN_PRIMARY, domains | get_domain(‘web-app-lam’), group_names) }}

filters()

filter_plugins.reserved_users module

class filter_plugins.reserved_users.FilterModule

Bases: object

User filters for extracting reserved and non-reserved subsets.

filters()

filter_plugins.reserved_users.non_reserved_users(users_dict)

Return a dict of users where reserved != true.

filter_plugins.reserved_users.reserved_usernames(users_dict)

Return a list of usernames where reserved: true. Usernames are regex-escaped to be safely embeddable.

filter_plugins.resource_filter module

class filter_plugins.resource_filter.FilterModule

Bases: object

filters()

filter_plugins.resource_filter.resource_filter(applications: dict, application_id: str, key: str, service_name: str, hard_default)

Lookup order:

1. docker.services.<service_name or get_entity_name(application_id)>.<key>

2. hard_default (mandatory)

• service_name may be “” → will resolve to get_entity_name(application_id).

• hard_default is mandatory (no implicit None).

• required=False always.

filter_plugins.role_path_by_app_id module

class filter_plugins.role_path_by_app_id.FilterModule

Bases: object

Provides the filters abs_role_path_by_application_id and rel_role_path_by_application_id.

filters()

filter_plugins.role_path_by_app_id.abs_role_path_by_application_id(application_id)

Searches all roles/*/vars/main.yml for application_id and returns the absolute path of the role that matches. Raises an error if zero or more than one match is found.

filter_plugins.role_path_by_app_id.rel_role_path_by_application_id(application_id)

Searches all roles/*/vars/main.yml for application_id and returns the relative path (from the project root) of the role that matches. Raises an error if zero or more than one match is found.

filter_plugins.text_filters module

class filter_plugins.text_filters.FilterModule

Bases: object

filters()

filter_plugins.text_filters.to_one_liner(s)

Collapse any multi-line string into a single line, trim extra whitespace, and remove JavaScript comments. Supports removal of both ‘//’ line comments and ‘/…/’ block comments, but preserves ‘//’ inside string literals and templating expressions.

filter_plugins.timeout_start_sec_for_domains module

class filter_plugins.timeout_start_sec_for_domains.FilterModule

Bases: object

filters()

timeout_start_sec_for_domains(domains_dict, include_www=True, per_domain_seconds=25, overhead_seconds=30, min_seconds=120, max_seconds=3600)

Parameters:

• domains_dict (dict | list[str] | str) – Either the domain mapping dict (values can be str | list[str] | dict[str,str]) or an already flattened list of domains, or a single domain string.

• include_www (bool) – If true, add ‘www.<domain>’ for non-www entries.

• ...

filter_plugins.to_primary_domain module

class filter_plugins.to_primary_domain.FilterModule

Bases: object

Custom filter to extract the primary/zone domain from a full domain name

filters()

to_primary_domain(domain)

Converts a full domain or subdomain into its primary/zone domain. E.g. ‘foo.bar.example.co.uk’ -> ‘example.co.uk’

filter_plugins.url_join module

Ansible filter plugin that safely joins URL components from a list. - Requires a valid ‘<scheme>://’ in the first element (any RFC-3986-ish scheme) - Preserves the double slash after the scheme, collapses other duplicate slashes - Supports query parts introduced by elements starting with ‘?’ or ‘&’

• first query element uses ‘?’, subsequent use ‘&’ (regardless of given prefix)

• each query element must be exactly one ‘key=value’ pair

• query elements may only appear after path elements; once query starts, no more path parts

• Raises specific AnsibleFilterError messages for common misuse

class filter_plugins.url_join.FilterModule

Bases: object

filters()

filter_plugins.url_join.url_join(parts)

Join a list of URL parts, URL-aware (scheme, path, query).

Parameters:

parts (list|tuple) – URL segments. First element MUST include ‘<scheme>://’. Path elements are plain strings. Query elements must start with ‘?’ or ‘&’ and contain exactly one ‘key=value’.

Returns:

Joined URL.

Return type:

str

Raises:

AnsibleFilterError – with specific, descriptive messages.

filter_plugins.volume_path module

class filter_plugins.volume_path.FilterModule

Bases: object

Docker volume path filters.

filters()

filter_plugins.volume_path.docker_volume_path(volume_name: str) → str

Returns the absolute filesystem path of a Docker volume.

Example

“akaunting_data” -> “/var/lib/docker/volumes/akaunting_data/_data/”

Module contents