plugins.filter package

Subpackages

Submodules

plugins.filter.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][‘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 plugins.filter.active_docker.FilterModule

Bases: object

filters()
plugins.filter.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

plugins.filter.application_allowed module

class plugins.filter.application_allowed.FilterModule

Bases: object

filters()
plugins.filter.application_allowed.application_allowed(application_id: str, group_names: list, allowed_applications: list | None = None)
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

plugins.filter.async_failures module

Filter async_failures: reduce reaped async_status results to the failures a deploy should stop for.

poll: 0 makes a task fire and forget, so its own changed_when and failed_when never see an outcome. The outcome only exists once the job is collected with async_status, and collecting it is where the verdict has to be made – for a whole batch at once, so one run reports every broken job rather than the first.

A job counts as failed when it never finished, when the module reported failed, or when a command returned a non-zero rc. What distinguishes a real failure from an idempotent no-op differs per API and lives in different keys: modules put it in msg (cloudflare answers “An identical record already exists”), commands put it in stdout (ollama pull prints “up to date” and exits non-zero). Both are searched, so a caller passes the tolerated phrasing and does not have to know which key carries it.

class plugins.filter.async_failures.FilterModule

Bases: object

filters() → dict[str, Any]
plugins.filter.async_failures.async_failures(results: Any, tolerate: Any) → list[str]

Describe every reaped job that failed and was not tolerated.

Parameters:

  • results – the results list of a looped async_status task.

  • tolerate – substrings that mark an answer as an accepted no-op.

plugins.filter.bcrypt_hash module

class plugins.filter.bcrypt_hash.FilterModule

Bases: object

static bcrypt_hash(value, salt_seed=None)
filters()

plugins.filter.ca_trust_paths module

Filter ca_cert_host: SPOT for the host-side root-CA cert location.

{{ SOFTWARE_DOMAIN | ca_cert_host }} -> /etc/<software_domain>/ca/root-ca.crt

ca_cert_host is where sys-ca-selfsigned provisions the root CA on the host. The container-side bind-mount targets live in the CA_TRUST group_vars SPOT (inject_cert_container / inject_wrapper_container) and reach the standalone runtime scripts as env vars and CLI arguments.

class plugins.filter.ca_trust_paths.FilterModule

Bases: object

filters()
plugins.filter.ca_trust_paths.ca_cert_host(software_domain: str) → str

plugins.filter.canonical_domains_map module

class plugins.filter.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()

plugins.filter.cookie_scope module

class plugins.filter.cookie_scope.FilterModule

Bases: object

filters()
plugins.filter.cookie_scope.common_dns_suffix(domains)

Return the longest dot-delimited DNS suffix shared by every domain. A single-domain input returns that domain unchanged, so single-domain apps keep their exact host while a multi-domain app (e.g. filer/master/api under one base) collapses to the shared parent that a session cookie can span across all of them.

plugins.filter.cookie_scope.domain_strings(domains)

Flatten the value returned by lookup(‘domains’, app) into a flat list of domain strings. The lookup yields a dict ({key: domain}), a list, or a bare string depending on the role’s domains shape, so normalise all three.

plugins.filter.csp_filters module

class plugins.filter.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.

static add_csp_hash(current, application_id, directive, snippet)

Return a new webserver_csp_hashes_extra_by_app-shaped dict with snippet appended (deduplicated) to <application_id>.<directive>.

Replaces the verbose inline combine(..., recursive=True) pattern previously duplicated across sys-front-inj-* roles.

build_csp_header(applications, application_id, domains, web_protocol, extra_whitelist=None, extra_hashes=None, domain_primary=None)

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:

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

  • “Smart defaults”:

    • internal CDN for style/script elem and connect

    • Matomo endpoints (if services.matomo.enabled) for script-elem/connect

    • Simpleicons service (if services.simpleicons.enabled) for connect

    • reCAPTCHA (if services.recaptcha.enabled) for script-elem/frame-src

    • hCaptcha (if services.hcaptcha.enabled) for script-elem/frame-src

    • frame-ancestors extended for dashboard/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 get_extra_values(extra_mapping, directive)
static is_feature_enabled(applications: dict, feature: str, application_id: str) → bool

Returns True if the docker service flag is enabled for this application.

New flag layout (examples):

  • services.matomo.enabled

  • services.dashboard.enabled

  • services.simpleicons.enabled

  • services.logout.enabled

  • services.hcaptcha.enabled

  • services.recaptcha.enabled

plugins.filter.curl module

Render the timeout-hardened, retry-hardened curl base command (SPOT).

{{ 30 | curl }} -> curl -s –connect-timeout 5 –max-time 30 –retry 3 –retry-all-errors –retry-delay 2 {{ 300 | curl }} -> curl -s –connect-timeout 5 –max-time 300 –retry 3 –retry-all-errors –retry-delay 2 {{ lookup(‘timeout’, 30) | curl }} -> TIMEOUT_FACTOR-scaled variant

param max_time:

Seconds for the whole transfer; without it curl hangs forever on an accepted-but-silent peer and task retries never fire.

param connect_timeout:

Seconds curl waits for the TCP/TLS connect (default 5).

param retries:

–retry count; –retry-all-errors makes it cover transport-layer resets (exit 35/52/56), the class that hard-fails deploys on flaky CI egress. Pass 0 to opt a call out of retrying.

Callers append their own flags (-S, -f, -o, …) after the filter.

class plugins.filter.curl.FilterModule

Bases: object

filters()
plugins.filter.curl.curl(max_time: Any, connect_timeout: Any = 5, retries: Any = 3) → str

plugins.filter.domain_tools module

class plugins.filter.domain_tools.FilterModule

Bases: object

filters()
plugins.filter.domain_tools.to_zone(hostname: str) → str

plugins.filter.dotenv module

class plugins.filter.dotenv.FilterModule

Bases: object

filters() → dict[str, Any]
plugins.filter.dotenv.dotenv_quote(ctx: Context, value: Any = <object object>) → str

Dotenv-safe quote, switched on DEPLOYMENT_MODE from the render context.

Also callable directly from Python (unit tests, helpers) without a Jinja context: in that case the only positional argument is the value and compose-style quoting is used.

plugins.filter.gitea_auth_noop module

Filter gitea_auth_noop: classify gitea auth-source idempotency errors.

failed_when: >

result.rc != 0 and not (result.stderr | gitea_auth_noop)

Returns True when the stderr of a failed gitea admin auth add-ldap|add-oauth only reports that the login source already exists; any other CLI error stays False so the caller’s failed_when still trips.

class plugins.filter.gitea_auth_noop.FilterModule

Bases: object

filters()
plugins.filter.gitea_auth_noop.gitea_auth_noop(stderr: str | None) → bool

plugins.filter.invokable_paths module

class plugins.filter.invokable_paths.FilterModule

Bases: object

filters()
plugins.filter.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.

plugins.filter.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.

plugins.filter.ldap_modify_noop module

Filter ldap_modify_noop: classify ldapmodify/ldapadd idempotency errors.

failed_when: >

result.rc != 0 and not (result.stderr | ldap_modify_noop)

Returns True when the stderr of a failed ldapmodify/ldapadd only reports that the attribute, value or entry already exists; any other LDAP error stays False so the caller’s failed_when still trips.

class plugins.filter.ldap_modify_noop.FilterModule

Bases: object

filters()
plugins.filter.ldap_modify_noop.ldap_modify_noop(stderr: str | None) → bool

plugins.filter.mirror_filters module

Jinja filter: external_asset_origins extracts the third-party asset hosts an application (or every application) declares in csp.whitelist.*.

The result feeds the web-svc-mirror privacy proxy: the mirror’s nginx vhost whitelists exactly these hosts as proxy upstreams, and the body-filter rewrite maps https://<host>/… to <mirror>/<host>/… in served HTML.

Only literal https://<host> tokens count as external assets:

  • tokens containing unresolved Jinja ({{ … }}) point at deployment-own domains and are skipped,

  • wildcard tokens (*) cannot be mirrored deterministically,

  • hosts on the deployment’s primary domain (or .onion) are already local.

class plugins.filter.mirror_filters.FilterModule

Bases: object

filters()
plugins.filter.mirror_filters.external_asset_origins(applications, application_id=None, primary_domain='')

Return the sorted list of external asset hosts.

application_id: restrict to one application; None aggregates across all. primary_domain: the deployment’s primary domain (its hosts are not external).

plugins.filter.on_off module

class plugins.filter.on_off.FilterModule

Bases: object

filters()
plugins.filter.on_off.on_off(value)

plugins.filter.redirect_sources module

class plugins.filter.redirect_sources.FilterModule

Bases: object

filters()
plugins.filter.redirect_sources.redirect_sources(host: str) → list[str]

Ordered match prefixes for a redirect that targets host.

Parameters:

host – the upstream host, as it appears in the proxied URL.

Returns:

The two host-qualified forms, then the bare / for a target the upstream expressed as an absolute path.

plugins.filter.reserved_users module

class plugins.filter.reserved_users.FilterModule

Bases: object

User filters for identity-directory subsets and reserved usernames.

filters()
plugins.filter.reserved_users.non_reserved_users(users_dict)

Return the subset of users that have an identity-directory account (identity in accounts). This is the set Keycloak/LDAP manage.

plugins.filter.reserved_users.reserved_usernames(users_dict)

Return the list of usernames that do NOT have an identity-directory account (identity not in accounts). These names are blocked from being registered as Keycloak usernames. Usernames are regex-escaped so the result is safe to embed in a deny pattern.

plugins.filter.resource_filter module

class plugins.filter.resource_filter.FilterModule

Bases: object

filters()
plugins.filter.resource_filter.resource_filter(applications: dict, application_id: str, key: str, service_name: str, hard_default)
Lookup order:

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

  2. services.<get_entity_name(application_id)>.<key>

  3. 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.

plugins.filter.role_path_by_app_id module

class plugins.filter.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()
plugins.filter.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.

plugins.filter.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.

plugins.filter.ruby module

class plugins.filter.ruby.FilterModule

Bases: object

filters()
static ruby_double_quoted_string(value)

Escape a value for use inside a Ruby double-quoted string and wrap it in double quotes.

Ruby rules handled: - “ -> ” - -> \ - newlines ->

  • carriage return ->

  • tab ->

NOT escaped on purpose: - $ - € - ! - ‘

plugins.filter.seaweedfs module

class plugins.filter.seaweedfs.FilterModule

Bases: object

filters()
plugins.filter.seaweedfs.min_free_space()
plugins.filter.seaweedfs.seaweedfs_anonymous_deny_policy(bucket)

Build the bucket policy that denies every unsigned read of bucket.

SeaweedFS honours a public-read object or bucket ACL on an unsigned read unless a bucket policy explicitly denies it first. A bare Principal: "*" deny would also hit the consumer’s own signed and presigned requests; the Null condition on aws:username limits it to unsigned ones, because SeaweedFS derives that key from a signed caller’s IAM user ARN and evaluates an unsigned request as principal * without it.

Parameters:

bucket – bucket the policy is attached to; scopes both the bucket ARN (listings) and its object ARNs (GET/HEAD).

plugins.filter.seaweedfs.seaweedfs_bucket_names(output)

Return the bucket names a weed shell s3.bucket.list printed.

Parameters:

output – stdout of the shell; every bucket line carries the name followed by a tab and its size: column, and the first one may follow the shell prompt on the same line.

plugins.filter.seaweedfs.seaweedfs_bucket_policy_target(stored, bucket, public)

Return the policy bucket converges to.

Every statement except DenyAnonymousRead is kept as stored; that one is replaced by the current seaweedfs_anonymous_deny_policy() statement on a private bucket and dropped on a public one.

Parameters:

  • stored – policy document the bucket carries, an empty dict for none.

  • bucket – bucket the deny statement scopes.

  • public – whether the consumer is marked public; must be a bool.

Returns:

The target policy document, or an empty dict when no statement remains and the bucket must carry no policy.

plugins.filter.seaweedfs.seaweedfs_command(s3_config='', collections=None)

Build the SeaweedFS all-in-one server command.

Single source for the args shared by the standalone provider stack (compose.yml.j2) and the local sidecar (service.yml.j2). -ip=localhost is deliberate: the all-in-one server otherwise advertises the auto-detected swarm ingress IP, which is unreachable for gRPC and hangs weed shell during bucket provisioning.

Parameters:

  • s3_config – path to the mounted s3.json; appended as -s3.config when truthy. The local sidecar passes "" since it does not mount that config.

  • collections – number of buckets this server serves, passed to volume_slots(). Required; without it the stock entrypoint injects -volume.max=0 and derives the ceiling from free disk space instead.

plugins.filter.seaweedfs.seaweedfs_sidecar_script(bucket, s3_port, access_key, secret_key)

Build the sh -c bootstrap script for the embedded sidecar.

The sidecar creates its consumer’s bucket itself at startup because a swarm stack’s converge gate blocks before any post-deploy task could create it, deadlocking consumers that hard-fail on a missing bucket (e.g. the OpenTalk controller). The stock entrypoint routes every argv through weed (sh as a command would become weed sh), so the template overrides it with sh -c; the server itself is still launched via /entrypoint.sh to keep its /data chown, privilege drop, and default server args.

The script must stay free of every shell $: the compose and swarm render paths launder dollars differently (a $pid reached the container as "" and crash-looped the sidecar on a sh syntax error). The bucket bootstrap therefore polls /status from a backgrounded subshell while exec makes the server PID 1; the poll loop dies with the container and readiness is gated by the consumer’s bounded wait.

The s3.configure grant is required since SeaweedFS 4.39: a server started without an -s3.config has zero identities, and ListBuckets then returns an empty list to the consumer’s access key, so the consumer reads the bucket as missing even though it exists. WriteAcp is required since SeaweedFS 4.49: a PutObject carrying x-amz-acl is also authorized as s3:PutObjectAcl and answers AccessDenied without it.

Parameters:

  • bucket – consumer bucket name created idempotently at startup.

  • s3_port – S3 API port polled via /status before bucket creation.

  • access_key – consumer S3 access key granted access to the bucket.

  • secret_key – consumer S3 secret key.

plugins.filter.seaweedfs.seaweedfs_stored_bucket_policy(output)

Return the policy SeaweedFS reports for a bucket.

Parameters:

output – stdout of the admin curl GET /<bucket>?policy run with -w '\n%{http_code}': the response body, a newline, and the HTTP status.

Returns:

The stored policy document on HTTP 200, an empty dict on HTTP 404 NoSuchBucketPolicy, and None on every other answer, a missing bucket (NoSuchBucket) included.

plugins.filter.seaweedfs.volume_slots(collections)

Volume-slot ceiling for a server hosting collections buckets.

Parameters:

collections – number of buckets this server serves. Each bucket is a SeaweedFS collection, and the master grows a collection in batches of VOLUME_GROW_BATCH; the default collection takes one batch of its own, hence the +1.

plugins.filter.security_headers module

Expose the response security headers of one application to templates.

class plugins.filter.security_headers.FilterModule

Bases: object

filters()

plugins.filter.sed module

class plugins.filter.sed.FilterModule

Bases: object

filters()
static sed_escape(value, delimiter: str = '|') → str

Escape a value for use as the replacement part of a sed s/// command.

Example

{{ password | sed_escape(‘|’) }}

plugins.filter.served_domains module

class plugins.filter.served_domains.FilterModule

Bases: object

filters()

plugins.filter.split_postgres_connections module

class plugins.filter.split_postgres_connections.FilterModule

Bases: object

filters()
plugins.filter.split_postgres_connections.split_postgres_connections(total_connections, roles_dir='roles')

Return an integer average: total_connections / number_of_roles_with_database_type_postgres. Uses max(count, 1) to avoid division-by-zero.

plugins.filter.status_codes module

Expose the status codes that prove a domain is served to templates.

class plugins.filter.status_codes.FilterModule

Bases: object

filters()

plugins.filter.swarm_ingress_guard module

Ports the swarm ingress publishes on every node that no application declares public.

class plugins.filter.swarm_ingress_guard.FilterModule

Bases: object

filters() → dict[str, Any]
plugins.filter.swarm_ingress_guard.swarm_ingress_guarded(published: Iterable[Mapping[str, Any]], public: Iterable[int]) → list[dict[str, Any]]

Drop the ports that are public by declaration from the published ones.

Parameters:

  • published – the output of swarm_ingress_published.

  • public – the output of swarm_public_ports.

Returns:

The published entries whose port no application declares public.

plugins.filter.swarm_ingress_guard.swarm_ingress_published(inspect_lines: Iterable[str]) → list[dict[str, Any]]

Return every port the routing mesh publishes, one entry per protocol and port.

Parameters:

inspect_lines – one docker service inspect --format '{{json .Endpoint.Ports}}' document per service; null for a service that publishes nothing.

Returns:

{"protocol": "tcp"|"udp", "port": int} for each ingress entry, sorted and without duplicates; host-mode entries are left out.

Raises:

AnsibleFilterError – a line is not JSON, or an entry names a protocol other than tcp or udp.

plugins.filter.swarm_ingress_guard.swarm_node_addresses(inspect_lines: Iterable[str]) → dict[str, list[str]]

Return the addresses the swarm nodes reach each other from, split by family.

Parameters:

inspect_lines – one docker node inspect --format '{{.Status.Addr}}' line per node.

Returns:

{"ipv4": [...], "ipv6": [...]}, sorted and without duplicates; the unspecified address a node may report is left out.

Raises:

AnsibleFilterError – a line is not an IP address.

plugins.filter.swarm_ingress_guard.swarm_public_ports(applications: Mapping[str, Any]) → list[int]

Return every port any application declares under services.*.ports.public.

Parameters:

applications – the merged lookup('applications') mapping.

Returns:

The declared public ports, {start, end} ranges expanded, sorted and without duplicates.

plugins.filter.text_filters module

class plugins.filter.text_filters.FilterModule

Bases: object

filters()
plugins.filter.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.

plugins.filter.timeout_start_sec_for_domains module

class plugins.filter.timeout_start_sec_for_domains.FilterModule

Bases: object

filters()
timeout_start_sec_for_domains(domains_dict, include_www=True, per_domain_seconds=25, onion_per_domain_seconds=None, 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.

  • per_domain_seconds (int) – Budget a clearnet domain contributes.

  • onion_per_domain_seconds (int | None) – Budget a ‘.onion’ domain contributes; None gives it per_domain_seconds.

  • ...

plugins.filter.to_primary_domain module

class plugins.filter.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’

plugins.filter.value_generator module

class plugins.filter.value_generator.FilterModule

Bases: object

filters()
plugins.filter.value_generator.strong_password(length: int = 32) → str

Module contents