Authenticate A2A requests and fix jwt-validation pipeline reload

exgentic_a2a_runner/authbridge/apply-pipeline.sh

@@ -60,6 +60,25 @@ TOKEN_BROKER_URL="${TOKEN_BROKER_URL:-}"
 TOKEN_BROKER_AUDIENCE="${TOKEN_BROKER_AUDIENCE:-}"
 export TOKEN_BROKER_URL TOKEN_BROKER_AUDIENCE
 
+# --- jwt-validation config (consumed only when jwt-validation is selected,
+# which every preset does). The plugin needs issuer + keycloak_url/realm that
+# the operator ships as flat keys in the generic `authbridge-config` ConfigMap —
+# NOT in the per-agent config.yaml the merge reads. Source them from there so
+# the operator stays the source of truth; an explicit JWT_VALIDATION_* env
+# override wins. Keys match authbridge's jwtValidationConfig struct
+# (issuer + keycloak_url + keycloak_realm); Configure() DisallowUnknownFields,
+# so do NOT introduce provider* keys here — those belong to token-exchange.
+AUTHBRIDGE_BASE_CM="${AUTHBRIDGE_BASE_CM:-authbridge-config}"
+_ab_base_key() {
+    # Echo a single flat key from the operator base ConfigMap, or empty.
+    kubectl -n "$NAMESPACE" get configmap "$AUTHBRIDGE_BASE_CM" \
+        -o jsonpath="{.data.$1}" 2>/dev/null || true
+}
+JWT_VALIDATION_ISSUER="${JWT_VALIDATION_ISSUER:-$(_ab_base_key ISSUER)}"
+JWT_VALIDATION_KEYCLOAK_URL="${JWT_VALIDATION_KEYCLOAK_URL:-$(_ab_base_key KEYCLOAK_URL)}"
+JWT_VALIDATION_KEYCLOAK_REALM="${JWT_VALIDATION_KEYCLOAK_REALM:-$(_ab_base_key KEYCLOAK_REALM)}"
+export JWT_VALIDATION_ISSUER JWT_VALIDATION_KEYCLOAK_URL JWT_VALIDATION_KEYCLOAK_REALM
+
 echo "Applying AuthBridge pipeline to $NAMESPACE/$AGENT_NAME"
 echo "  Plugins: $PIPELINE_PLUGINS"
 if [ -n "${PIPELINE_OVERLAY_FILE:-}" ]; then
@@ -129,6 +148,26 @@ if $ibac_active && [ ! -s "$PROMPT_FILE" ]; then
     exit 1
 fi
 
+# --- Pre-flight: confirm the issuer is resolved when jwt-validation is
+# active. Without it the fragment renders an empty issuer and the sidecar
+# rejects the reload ("jwt-validation config: issuer is required"), leaving
+# the pod serving stale config — the exact failure this guard prevents.
+jwt_validation_active=false
+for tok in $PIPELINE_PLUGINS; do
+    name=${tok%%:*}
+    policy=${tok#*:}; [[ "$policy" == "$tok" ]] && policy=enforce
+    if [[ "$name" == "jwt-validation" && "$policy" != "off" ]]; then
+        jwt_validation_active=true
+        break
+    fi
+done
+if $jwt_validation_active && [ -z "${JWT_VALIDATION_ISSUER:-}" ]; then
+    echo "ERROR: jwt-validation is active but no issuer could be resolved." >&2
+    echo "       Expected key ISSUER in ConfigMap $NAMESPACE/$AUTHBRIDGE_BASE_CM," >&2
+    echo "       or set JWT_VALIDATION_ISSUER explicitly." >&2
+    exit 1
+fi
+
 # --- Render every plugin fragment with envsubst into a temp dir. The
 # merge script reads these as the rendered defaults; --plugin-config-file
 # overrides are deep-merged on top inside pipeline-merge.py.

exgentic_a2a_runner/authbridge/pipeline-merge.py

@@ -29,6 +29,7 @@
 """
 
 import argparse
+import copy
 import sys
 from pathlib import Path
 from typing import Any
@@ -118,22 +119,63 @@ def build_entry(
     policy: str,
     fragment: dict[str, Any],
     overrides: dict[str, Any] | None,
+    operator_config: dict[str, Any] | None,
 ) -> dict[str, Any]:
-    """Build the final plugin entry: fragment defaults + overrides + on_error."""
+    """Build the final plugin entry.
+
+    Config precedence (later wins on leaf conflicts):
+        operator base config  <  fragment defaults  <  --config-file overrides
+
+    The operator base is the config block the operator already rendered for
+    this plugin in the ConfigMap we read from stdin. Preserving it is the
+    whole point: several plugins (jwt-validation's issuer, token-exchange's
+    token_url/identity/routes) are configured ONLY by the operator, and a
+    prior version of this script replaced the plugin list wholesale — which
+    dropped that config and made the sidecar reject the reload
+    ("issuer is required", "token_url is required"). We now start from the
+    operator's block and layer our fragment + overrides on top, matching how
+    authbridge's own `abctl edit` splices rather than rebuilds.
+    """
     entry: dict[str, Any] = {"name": name}
-    cfg = fragment.get("config")
-    if isinstance(cfg, dict) and cfg:
-        entry["config"] = dict(cfg)
+    config: dict[str, Any] = {}
+    if isinstance(operator_config, dict) and operator_config:
+        config = copy.deepcopy(operator_config)
+    frag_cfg = fragment.get("config")
+    if isinstance(frag_cfg, dict) and frag_cfg:
+        deep_merge(config, copy.deepcopy(frag_cfg))
     if overrides:
-        entry.setdefault("config", {})
-        deep_merge(entry["config"], overrides)
+        deep_merge(config, overrides)
+    if config:
+        entry["config"] = config
     # `enforce` is the framework default — omit on_error to keep diffs
     # minimal. observe/off are explicit.
     if policy != "enforce":
         entry["on_error"] = policy
     return entry
 
 
+def index_operator_config(operator: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Map plugin name -> its `config` block as rendered by the operator in
+    the ConfigMap read from stdin. Scans both chains; missing/blank config
+    yields no entry."""
+    by_name: dict[str, dict[str, Any]] = {}
+    pipeline = operator.get("pipeline")
+    if not isinstance(pipeline, dict):
+        return by_name
+    for chain in ("inbound", "outbound"):
+        section = pipeline.get(chain)
+        if not isinstance(section, dict):
+            continue
+        for entry in section.get("plugins", []) or []:
+            if not isinstance(entry, dict):
+                continue
+            n = entry.get("name")
+            cfg = entry.get("config")
+            if isinstance(n, str) and isinstance(cfg, dict) and cfg:
+                by_name[n] = cfg
+    return by_name
+
+
 def validate_mutex(resolved: dict[str, str]) -> None:
     """Reject mutually-exclusive plugin pairs both active (not off) in the
     same chain."""
@@ -207,31 +249,35 @@ def main() -> int:
         if text.strip():
             ibac_prompt = text
 
-    # Build inbound and outbound chains in canonical order.
+    # Index the config blocks the operator already rendered, so build_entry
+    # can preserve them (issuer, token_url/identity, …) instead of dropping
+    # them. Done before we overwrite the plugin lists below.
+    operator_config = index_operator_config(operator)
+
+    # Build inbound and outbound chains in canonical order. We're
+    # authoritative for the chain composition (which plugins, in what order,
+    # with what on_error) by design (see §4.3), but each plugin's config is
+    # seeded from the operator's rendered block so operator-supplied settings
+    # survive.
     inbound_entries: list[dict[str, Any]] = []
     for name in INBOUND_ORDER:
-        # Skip plugins not in the operator base when our policy is `off`
-        # — no need to emit a no-op entry for something that doesn't
-        # exist downstream. The operator base today enables every
-        # supported plugin, so this branch effectively never fires; it's
-        # here for the case where the operator drops one in the future.
-        # (Conservative behavior: always emit, since we don't read the
-        # operator base here. We *always* emit.)
         fragment = load_fragment(args.plugins_dir, name)
-        entry = build_entry(name, resolved[name], fragment, overrides.get(name))
+        entry = build_entry(name, resolved[name], fragment,
+                            overrides.get(name), operator_config.get(name))
         inbound_entries.append(entry)
 
     outbound_entries: list[dict[str, Any]] = []
     for name in OUTBOUND_ORDER:
         fragment = load_fragment(args.plugins_dir, name)
-        entry = build_entry(name, resolved[name], fragment, overrides.get(name))
+        entry = build_entry(name, resolved[name], fragment,
+                            overrides.get(name), operator_config.get(name))
         if name == "ibac" and ibac_prompt is not None and resolved[name] != "off":
             cfg = entry.setdefault("config", {})
             cfg["system_prompt"] = ibac_prompt
         outbound_entries.append(entry)
 
-    # Replace operator's plugin lists with our resolved ones. We're
-    # authoritative for the chain composition by design (see §4.3).
+    # Replace operator's plugin lists with our resolved ones (composition is
+    # ours; per-plugin config was preserved via operator_config above).
     pipeline = operator.setdefault("pipeline", {})
     pipeline.setdefault("inbound", {})["plugins"] = inbound_entries
     pipeline.setdefault("outbound", {})["plugins"] = outbound_entries

exgentic_a2a_runner/authbridge/plugins/jwt-validation.yaml

@@ -1,8 +1,30 @@
 # jwt-validation plugin fragment.
 #
-# The operator base config supplies issuer/keycloak_url/keycloak_realm.
-# This fragment exists so the merge can keep the plugin in the resolved
-# selection (or flip its on_error policy) without re-rendering the
-# operator's heavy config block. Per-plugin overrides may be supplied
-# via --plugin-config-file (see AUTHBRIDGE_PIPELINE_SPEC.md §4.5).
+# Placeholders below are expanded by apply-pipeline.sh (envsubst) before
+# the merge runs. apply-pipeline.sh sources the issuer / keycloak values
+# from the operator's `authbridge-config` ConfigMap (flat keys ISSUER,
+# KEYCLOAK_URL, KEYCLOAK_REALM) and exports them as the JWT_VALIDATION_*
+# vars below; a JWT_VALIDATION_* override in the environment wins over the
+# operator value.
+#
+# Historically this fragment was config-less on the assumption that the
+# operator base config already carried a rendered jwt-validation config
+# block. It does not — the operator ships flat env keys in a separate
+# ConfigMap — so pipeline-merge.py (which replaces the inbound plugin
+# list wholesale) emitted a bare `name: jwt-validation`, and the sidecar
+# rejected the reload with "issuer is required". Rendering the config
+# here, like plugins/ibac.yaml, keeps the operator as the source of truth
+# while giving the plugin the config it needs.
+#
+# Schema matches the authbridge jwtValidationConfig struct
+# (authbridge/authlib/plugins/jwtvalidation/plugin.go @ v0.6.0-alpha.12):
+# issuer (required), keycloak_url + keycloak_realm (derive jwks_url).
+# Configure() calls DisallowUnknownFields(), so ONLY these keys are
+# accepted — note `provider`/`provider_url`/`provider_realm` belong to the
+# token-exchange plugin, NOT jwt-validation. Per-plugin overrides may still
+# be supplied via --plugin-config-file (see AUTHBRIDGE_PIPELINE_SPEC.md §4.5).
 name: jwt-validation
+config:
+  issuer: "${JWT_VALIDATION_ISSUER}"
+  keycloak_url: "${JWT_VALIDATION_KEYCLOAK_URL}"
+  keycloak_realm: "${JWT_VALIDATION_KEYCLOAK_REALM}"

exgentic_a2a_runner/deploy-agent.sh

@@ -180,6 +180,30 @@ if [ -z "$BENCHMARK_NAME" ] || [ -z "$AGENT_NAME_INPUT" ]; then
     exit 1
 fi
 
+# Prerequisite check: AuthBridge plugin resolution (Step 8) and the skip_hosts
+# merge (Step 7.5) both shell out to `python3` with PyYAML. That happens deep
+# into the deploy after minutes of network work, so a missing venv/module fails
+# late (ModuleNotFoundError: No module named 'yaml'). Detect it here, up front,
+# and only when a plugin selector was actually supplied (the sole trigger for
+# those Python steps).
+if [ -n "$PIPELINE_PRESET" ] || [ ${#PIPELINE_SELECTORS[@]} -gt 0 ] || [ -n "$PIPELINE_OVERLAY_FILE" ]; then
+    if ! command -v python3 >/dev/null 2>&1; then
+        echo "Error: python3 is required for AuthBridge plugin resolution but was not found on PATH." >&2
+        echo "  You passed a plugin selector (--plugin/--no-plugin/--plugin-preset/--plugin-config-file)," >&2
+        echo "  which needs python3 + PyYAML. Install python3 and re-run." >&2
+        exit 1
+    fi
+    if ! python3 -c 'import yaml' 2>/dev/null; then
+        echo "Error: python3 is on PATH but the PyYAML module is missing." >&2
+        echo "  $(command -v python3) cannot 'import yaml'. This usually means the wrong" >&2
+        echo "  Python/venv is active. Activate the venv with PyYAML, or install it:" >&2
+        echo "    pip3 install --user pyyaml" >&2
+        echo "    brew install libyaml && pip3 install pyyaml      # macOS" >&2
+        echo "    sudo apt install python3-yaml                    # Debian/Ubuntu" >&2
+        exit 1
+    fi
+fi
+
 # Determine agent name and image
 # Automatically add exgentic-a2a- prefix if not already present
 if [[ "$AGENT_NAME_INPUT" == exgentic-a2a-* ]]; then

exgentic_a2a_runner/evaluate-benchmark.sh

@@ -378,6 +378,46 @@ source .venv/bin/activate
 export EXGENTIC_MCP_SERVER_URL="${MCP_BASE_URL}/mcp"
 export A2A_BASE_URL="$AGENT_BASE_URL"
 
+# Obtain a Keycloak bearer token for the A2A agent endpoint. The agent runs
+# behind an authbridge sidecar that validates a JWT from the "rossoctl" realm,
+# so unauthenticated requests get HTTP 401. This mirrors the direct-access-grant
+# flow used by deploy-benchmark.sh / deploy-agent.sh (client_id=rossoctl).
+# Skip if A2A_AUTH_TOKEN is already provided in the environment/.env.
+if [ -z "${A2A_AUTH_TOKEN:-}" ]; then
+    echo "Obtaining Keycloak token for A2A agent..."
+    KEYCLOAK_API="$(keycloak_api_url)"
+    KEYCLOAK_USERNAME="${KEYCLOAK_USERNAME:-admin}"
+
+    # Resolve the password the same way deploy-agent.sh does: prefer an explicit
+    # KEYCLOAK_PASSWORD, else the rossoctl-test-user cluster secret, else "admin".
+    if [ -z "${KEYCLOAK_PASSWORD:-}" ] || [ "${KEYCLOAK_PASSWORD}" = "unknown" ]; then
+        KC_SECRET_PASSWORD=$("$KUBECTL_BIN" get secret rossoctl-test-user -n keycloak \
+            -o jsonpath='{.data.password}' 2>/dev/null | base64 -d 2>/dev/null || echo "")
+        KEYCLOAK_PASSWORD="${KC_SECRET_PASSWORD:-admin}"
+    fi
+
+    A2A_TOKEN_RESPONSE=$(curl -s -X POST "$KEYCLOAK_API/realms/rossoctl/protocol/openid-connect/token" \
+        -H "Content-Type: application/x-www-form-urlencoded" \
+        -d "username=${KEYCLOAK_USERNAME}" \
+        -d "password=${KEYCLOAK_PASSWORD}" \
+        -d "grant_type=password" \
+        -d "client_id=rossoctl" 2>/dev/null || echo "")
+
+    A2A_AUTH_TOKEN=$(echo "$A2A_TOKEN_RESPONSE" | grep -o '"access_token":"[^"]*"' | sed 's/"access_token":"\([^"]*\)"/\1/')
+    if [ -z "$A2A_AUTH_TOKEN" ]; then
+        echo "Error: Could not obtain Keycloak token for the A2A agent."
+        echo "  Keycloak: $KEYCLOAK_API (realm rossoctl, client_id rossoctl, user $KEYCLOAK_USERNAME)"
+        echo "  Response: $A2A_TOKEN_RESPONSE"
+        echo "  Ensure Direct Access Grants are enabled for the rossoctl client"
+        echo "  (deploy-agent.sh enables them) or set A2A_AUTH_TOKEN explicitly."
+        exit 1
+    fi
+    export A2A_AUTH_TOKEN
+    echo "✓ Obtained A2A bearer token"
+else
+    echo "Using A2A_AUTH_TOKEN from environment"
+fi
+
 # Set tool prefix when using MCP gateway (gateway namespaces tools with a prefix)
 if [ "$USE_MCP_GATEWAY" = "true" ]; then
     export EXGENTIC_MCP_TOOL_PREFIX="${EXGENTIC_MCP_TOOL_PREFIX:-exgentic_${BENCHMARK_NAME}_}"

exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py

@@ -96,9 +96,16 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N
         # keys intent off authbridge's own session store (populated by a2a-parser),
         # not this header — but emitting it is harmless and useful for tracing.
         x_session_id = uuid.uuid4().hex
+        headers = {"x-session-id": x_session_id}
+        # Attach the Keycloak-issued bearer token so authbridge (the agent's
+        # sidecar) accepts the request; without it the endpoint returns 401.
+        # The same header is applied to the agent-card fetch and every
+        # send_message call because the client is reused for both.
+        if self.config.auth_token:
+            headers["Authorization"] = f"Bearer {self.config.auth_token}"
         httpx_client = httpx.AsyncClient(
             timeout=timeout_s,
-            headers={"x-session-id": x_session_id},
+            headers=headers,
         )
         if self.otel_enabled:
             try: