ansible-infra/roles/keycloak_auth_flow/library/keycloak_auth_flow_lint.py

#!/usr/bin/python

from __future__ import annotations

DOCUMENTATION = r"""
module: keycloak_auth_flow_lint

short_description: Locally validate a Keycloak authentication flow definition

description:
  - Statically validates the structure of a nested Keycloak authentication flow definition, in
    the shape accepted by C(middleware_automation.keycloak.keycloak_authentication_v2)'s
    O(middleware_automation.keycloak.keycloak_authentication_v2#module:authenticationExecutions)
    option.
  - This performs no network calls and requires no Keycloak credentials, so it can catch YAML
    authoring mistakes (typos, missing keys, bad nesting) before ever touching a live server.
  - Some checks (for example whether a C(providerId) is spelled correctly) are necessarily best
    effort, since Keycloak's set of registered authenticator providers depends on the server and
    its installed SPI plugins. Those only produce warnings, not failures.

options:
  realm:
    description: The name of the realm the flow belongs to.
    required: true
    type: str
  alias:
    description: The name of the authentication flow.
    required: true
    type: str
  providerId:
    description: The C(providerId) for the flow.
    choices: [basic-flow, client-flow]
    type: str
    default: basic-flow
  authenticationExecutions:
    description: The desired execution configuration for the flow, at root level.
    required: true
    type: list
    elements: dict

author:
  - CCCHH
"""

EXAMPLES = r"""
- name: Validate a flow definition before deploying it
  keycloak_auth_flow_lint:
    realm: "{{ item.realm }}"
    alias: "{{ item.alias }}"
    providerId: "{{ item.providerId | default('basic-flow') }}"
    authenticationExecutions: "{{ item.authenticationExecutions }}"
  loop: "{{ keycloak_auth_flow__flows }}"
"""

RETURN = r"""
warnings:
  description: Non-fatal issues found in the flow definition.
  returned: always
  type: list
  elements: str
"""

from ansible.module_utils.basic import AnsibleModule

# Non-exhaustive list of Keycloak built-in authenticator provider IDs relevant to browser flows.
# An unrecognized providerId only produces a warning, not a failure, since custom SPI providers
# exist. This list is NOT an authoritative source, just a fast offline pre-check:
#   - Most entries were observed directly on the live CCCHH Keycloak server (26.x): the built-in
#     flows (browser, direct grant, registration, reset credentials, clients, docker auth, first
#     broker login) and the custom "browser passkey and token" flow, fetched via the admin API.
#   - The rest (other idp-*, registration-*, conditional-*, auth-conditional-otp-form,
#     auth-password-form, identity-provider-autolink) are from general Keycloak knowledge, not
#     verified against this server.
# To check or extend this list later:
#   - Live server (authoritative, and what keycloak_authentication_v2 itself checks against at
#     deploy time): GET /admin/realms/{realm}/authentication/authenticator-providers
#   - Keycloak source: each provider's getId() in its *AuthenticatorFactory.java, under
#     https://github.com/keycloak/keycloak/tree/main/services/src/main/java/org/keycloak/authentication/authenticators
KNOWN_PROVIDER_IDS = {
    "auth-conditional-otp-form",
    "auth-cookie",
    "auth-otp-form",
    "auth-password-form",
    "auth-spnego",
    "auth-username-password-form",
    "client-jwt",
    "client-secret",
    "client-secret-jwt",
    "client-x509",
    "conditional-credential",
    "conditional-level-of-authentication",
    "conditional-user-attribute",
    "conditional-user-configured",
    "conditional-user-role",
    "direct-grant-validate-otp",
    "direct-grant-validate-password",
    "direct-grant-validate-username",
    "docker-http-basic-authenticator",
    "identity-provider-autolink",
    "identity-provider-redirector",
    "idp-confirm-link",
    "idp-create-user-if-unique",
    "idp-email-verification",
    "idp-review-profile",
    "idp-username-password-form",
    "organization",
    "registration-page-form",
    "registration-password-action",
    "registration-recaptcha-action",
    "registration-terms-and-conditions",
    "registration-user-creation",
    "reset-credential-email",
    "reset-credentials-choose-user",
    "reset-password",
    "webauthn-authenticator",
    "webauthn-authenticator-passwordless",
}

REQUIREMENTS = {"REQUIRED", "ALTERNATIVE", "DISABLED", "CONDITIONAL"}
SUBFLOW_TYPES = {"basic-flow", "form-flow"}
ALLOWED_NODE_KEYS = {
    "requirement",
    "providerId",
    "subFlow",
    "subFlowType",
    "authenticationExecutions",
    "authenticationConfig",
}
MAX_SUBFLOW_DEPTH = 10


class FlowValidationError(Exception):
    pass


def validate_authentication_config(auth_config, path, provider_id, warnings):
    if not isinstance(auth_config, dict):
        raise FlowValidationError(f"{path}.authenticationConfig: must be a mapping")
    extra = set(auth_config) - {"alias", "config"}
    if extra:
        raise FlowValidationError(f"{path}.authenticationConfig: unexpected key(s): {', '.join(sorted(extra))}")
    if not isinstance(auth_config.get("alias"), str) or not auth_config["alias"]:
        raise FlowValidationError(f"{path}.authenticationConfig: 'alias' is required and must be a non-empty string")
    config = auth_config.get("config")
    if not isinstance(config, dict):
        raise FlowValidationError(f"{path}.authenticationConfig: 'config' is required and must be a mapping")

    if provider_id == "conditional-credential":
        credentials = config.get("credentials")
        if not credentials:
            raise FlowValidationError(
                f"{path}.authenticationConfig.config: 'conditional-credential' requires a non-empty "
                "'credentials' entry"
            )
        included = config.get("included")
        if included is not None and included not in ("true", "false"):
            raise FlowValidationError(
                f"{path}.authenticationConfig.config: 'included' must be the string 'true' or 'false' "
                f"(got {included!r}) — Keycloak stores authenticator config values as strings"
            )
        # The included=true/false semantics are unlabeled in the Keycloak admin console and easy to
        # get backwards, so restate them every time as a standing reminder. Keycloak reads the
        # value with Boolean.parseBoolean(), so an unset 'included' behaves like "false".
        if included != "true":
            meaning = f"TRUE when NONE of [{credentials}] were used (inverted)"
        else:
            meaning = f"TRUE when ANY of [{credentials}] was used"
        warnings.append(
            f"{path}: 'conditional-credential' with included={included!r} — condition is {meaning}. "
            "This is easy to get backwards since the admin console has no tooltip for it."
        )
    elif provider_id == "auth-conditional-otp-form" and not config:
        warnings.append(
            f"{path}: 'auth-conditional-otp-form' (Conditional OTP Form) has an empty config — "
            "unlike the plain 'auth-otp-form', its skip/force logic is entirely self-contained (by "
            "user attribute, role, header, or default) and does NOT inherit a wrapping Condition-* "
            "check; with no config it defaults to always showing the OTP form."
        )


def validate_node(node, path, depth, seen_subflow_names, seen_provider_ids, warnings):
    if not isinstance(node, dict):
        raise FlowValidationError(f"{path}: must be a mapping")

    extra = set(node) - ALLOWED_NODE_KEYS
    if extra:
        raise FlowValidationError(f"{path}: unexpected key(s): {', '.join(sorted(extra))}")

    requirement = node.get("requirement")
    if requirement not in REQUIREMENTS:
        raise FlowValidationError(
            f"{path}.requirement: must be one of {sorted(REQUIREMENTS)}, got {requirement!r}"
        )

    has_provider = node.get("providerId") is not None
    has_subflow = node.get("subFlow") is not None
    if has_provider and has_subflow:
        raise FlowValidationError(f"{path}: has both 'providerId' and 'subFlow' — an execution is exactly one of the two")
    if not has_provider and not has_subflow:
        raise FlowValidationError(f"{path}: must have exactly one of 'providerId' or 'subFlow'")

    if has_subflow:
        if depth >= MAX_SUBFLOW_DEPTH:
            raise FlowValidationError(
                f"{path}: nesting exceeds the maximum of {MAX_SUBFLOW_DEPTH} sub-flow levels "
                "supported by middleware_automation.keycloak.keycloak_authentication_v2"
            )

        name = node["subFlow"]
        if not isinstance(name, str) or not name:
            raise FlowValidationError(f"{path}.subFlow: must be a non-empty string")
        if name in seen_subflow_names:
            raise FlowValidationError(f"{path}.subFlow: duplicate sub-flow name {name!r} within this flow")
        seen_subflow_names.add(name)

        subflow_type = node.get("subFlowType", "basic-flow")
        if subflow_type not in SUBFLOW_TYPES:
            raise FlowValidationError(f"{path}.subFlowType: must be one of {sorted(SUBFLOW_TYPES)}, got {subflow_type!r}")

        if "authenticationConfig" in node:
            raise FlowValidationError(f"{path}: 'authenticationConfig' is not valid on a sub-flow node")

        children = node.get("authenticationExecutions")
        if not children:
            raise FlowValidationError(f"{path}.authenticationExecutions: sub-flow {name!r} must define at least one execution")
        if not isinstance(children, list):
            raise FlowValidationError(f"{path}.authenticationExecutions: must be a list")

        # Direct-child providerIds, used by the check below. Only *direct* children count:
        # Keycloak's "Condition executions can only be contained in Conditional subflow" rule
        # concerns immediate children, not deeper descendants.
        direct_provider_ids = {
            child.get("providerId")
            for child in children
            if isinstance(child, dict) and child.get("providerId")
        }

        # Gotcha: a 'conditional-*' execution is silently ignored (no error, condition just never
        # fires) unless the wrapping subflow's own requirement is literally CONDITIONAL.
        conditional_children = {pid for pid in direct_provider_ids if pid.startswith("conditional-")}
        if conditional_children and requirement != "CONDITIONAL":
            raise FlowValidationError(
                f"{path}: sub-flow {name!r} contains condition execution(s) "
                f"({', '.join(sorted(conditional_children))}) but its own requirement is "
                f"{requirement!r}, not CONDITIONAL — Keycloak silently ignores condition executions "
                "whose wrapping subflow isn't CONDITIONAL (no error is shown; the condition just "
                "never fires)"
            )

        for index, child in enumerate(children):
            validate_node(
                child,
                f"{path}.authenticationExecutions[{index}]",
                depth + 1,
                seen_subflow_names,
                seen_provider_ids,
                warnings,
            )
    else:
        if "subFlowType" in node:
            raise FlowValidationError(f"{path}: 'subFlowType' is not valid on an execution node")
        if "authenticationExecutions" in node:
            raise FlowValidationError(f"{path}: 'authenticationExecutions' is not valid on an execution node")

        provider_id = node["providerId"]
        if not isinstance(provider_id, str) or not provider_id:
            raise FlowValidationError(f"{path}.providerId: must be a non-empty string")
        seen_provider_ids.add(provider_id)
        if provider_id not in KNOWN_PROVIDER_IDS:
            warnings.append(
                f"{path}: providerId '{provider_id}' is not in the list of known Keycloak provider "
                "IDs — double-check for a typo, or ignore if this is a custom SPI provider"
            )
        if requirement == "CONDITIONAL" and not provider_id.startswith("conditional-"):
            warnings.append(
                f"{path}: requirement is CONDITIONAL but providerId '{provider_id}' does not look "
                "like a condition provider (expected something like 'conditional-*')"
            )

        auth_config = node.get("authenticationConfig")
        if auth_config is not None:
            validate_authentication_config(auth_config, path, provider_id, warnings)
        elif provider_id == "conditional-credential":
            raise FlowValidationError(
                f"{path}: 'conditional-credential' requires an authenticationConfig (with a "
                "'credentials' entry) — without it the condition doesn't check anything"
            )
        elif provider_id == "auth-conditional-otp-form":
            warnings.append(
                f"{path}: 'auth-conditional-otp-form' (Conditional OTP Form) has no "
                "authenticationConfig — unlike the plain 'auth-otp-form', its skip/force logic is "
                "entirely self-contained (by user attribute, role, header, or default) and does NOT "
                "inherit a wrapping Condition-* check; with no config it defaults to always showing "
                "the OTP form."
            )


def main() -> None:
    module = AnsibleModule(
        argument_spec=dict(
            realm=dict(type="str", required=True),
            alias=dict(type="str", required=True),
            providerId=dict(type="str", default="basic-flow", choices=["basic-flow", "client-flow"]),
            authenticationExecutions=dict(type="list", elements="dict", required=True),
        ),
        supports_check_mode=True,
    )

    warnings: list[str] = []
    seen_subflow_names: set[str] = set()
    seen_provider_ids: set[str] = set()

    try:
        for index, execution in enumerate(module.params["authenticationExecutions"]):
            validate_node(
                execution,
                f"authenticationExecutions[{index}]",
                depth=0,
                seen_subflow_names=seen_subflow_names,
                seen_provider_ids=seen_provider_ids,
                warnings=warnings,
            )
    except FlowValidationError as e:
        module.fail_json(
            msg=f"{module.params['realm']}/{module.params['alias']}: {e}",
            warnings=warnings,
        )

    # Gotcha: with native passkey support, 'auth-username-password-form' can itself complete a
    # full passkey login, so a separately-added 'webauthn-authenticator-passwordless' execution
    # elsewhere in the flow may end up dead/unreached with no indication in the admin console.
    if "auth-username-password-form" in seen_provider_ids and "webauthn-authenticator-passwordless" in seen_provider_ids:
        warnings.append(
            "flow has both 'auth-username-password-form' and 'webauthn-authenticator-passwordless' "
            "— with native passkey support, the username/password form can itself complete a "
            "passkey login, so it may be ambiguous (and unverifiable from the admin console) which "
            "execution actually authenticates a given passkey login. Verify via the 'execution=<id>' "
            "query parameter during a live test, or DEBUG logging on 'org.keycloak.authentication'."
        )

    for warning in warnings:
        module.warn(f"{module.params['realm']}/{module.params['alias']}: {warning}")

    module.exit_json(
        changed=False,
        msg=f"{module.params['realm']}/{module.params['alias']}: flow definition is structurally valid",
        warnings=warnings,
    )


if __name__ == "__main__":
    main()