fix(security): harden A2A runtime trust boundaries, scoping, and validation

Security fixes:
- Add authentication to the /invoke endpoint (HMAC trust header gate)
- Add authentication to the A2A proxy catch-all before forwarding
- Require non-empty A2A_RUST_AUTH_SECRET at startup to prevent
  predictable trust headers
- Use server-derived owner_email/team_id in agent registration instead
  of trusting client-provided values
- Add visibility scoping to the flush_events internal endpoint
- Close visibility bypass on events/replay when task row is absent
- Close visibility bypass on push config get/list when agent_id omitted
- Validate webhook URL through Pydantic schema (SSRF protection) in
  internal push config creation endpoint
- Add webhook URL validation via validate_core_url on
  A2APushNotificationConfigCreate schema
- Exclude auth_token from A2APushNotificationConfigRead (Field exclude)
- Change _check_agent_access_by_id to fail-closed for deleted agents
- Make encode_auth_context a hard error instead of silent empty fallback
- Sanitize all error responses to external callers — strip internal
  backend URLs, Python response bodies, and reqwest error details
- Redact Redis URL credentials in logs
- Use constant-time comparison for session fingerprint validation
- Warn at startup if HTTP listener is non-loopback

Correctness fixes:
- Pass actual JSON-RPC method name through full_authenticate
- Add explicit authz match arms for cancel, delete, create, and stream
- Remove x-forwarded-for from default session fingerprint headers
- Fix cargo fmt conflicts with pragma allowlist comments
- Cap retry backoff at 60 seconds to prevent runaway delays

Error handling improvements:
- Replace expect() in queue worker with graceful error handling for
  semaphore closure, panicked JoinSet tasks, and missing results
- Add logger.exception + db.invalidate fallback to all 11 internal
  A2A endpoint exception handlers
- Upgrade query param decryption failure log from debug to warning
- Log warning on auth decoding failure during agent update
- Return None from event_store.store_event on serialization failure
- Replace .ok() with logged match on proxy response JSON parsing
- Replace mutex expect("poisoned") with unwrap_or_else recovery
- Upgrade shadow mode payload mismatch log to warning with traceback
- Upgrade Redis cache invalidation failure log to warning
- Bound resolve_inflight DashMap to prevent unbounded memory growth

Comment and code cleanup:
- Fix handle_a2a_proxy doc (does not inject trust headers)
- Fix coalesce_jobs doc (does not match on timeout)
- Fix proxy_task_method doc (also handles cancel, not just reads)
- Clarify trust header as keyed SHA-256, not formal HMAC
- Renumber handle_a2a_invoke steps sequentially (1-2-3-3a-4-5)
- Document _visible_agent_ids admin-bypass divergence
- Remove leftover logger.info debug statements from register_agent
- Remove commented-out dead code from register_agent
- Remove duplicate TOOLS_MANAGE_PLUGINS constant
- Remove constant-comparison test (test_version_endpoint_redis_conditions)

Test coverage:
- Add visibility tests for cancel_task (wrong team, admin, public-only)
- Add deny-path tests for events/replay (inaccessible agent, missing task)
- Add 32 tests for _check_agent_access_by_id, _visible_agent_ids,
  get_task, list_tasks, and _check_server_access
- Add tests for auth_secret startup rejection (unit + binary)
- Update integration tests for trust-gated /invoke and authenticated
  proxy endpoints

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Jonathan Springer <jps@s390x.com>

Author: jonpspri

.secrets.baseline

@@ -135,13 +135,19 @@ def upgrade() -> None:
 
     # --- Backfill a2a_agent_auth from existing a2a_agents auth columns -------
     if "a2a_agent_auth" in inspector.get_table_names():
+        # Standard
+        import json
+        import uuid
+
         conn = op.get_bind()
         agents_with_auth = conn.execute(
             sa.text("SELECT id, auth_type, auth_value, auth_query_params " "FROM a2a_agents " "WHERE auth_type IS NOT NULL " "AND id NOT IN (SELECT a2a_agent_id FROM a2a_agent_auth)")
         ).fetchall()
         for agent in agents_with_auth:
-            # Standard
-            import uuid
+            # auth_query_params may be a Python dict (from a JSON column);
+            # serialize it so the untyped text bind works on all drivers.
+            raw_params = agent[3]
+            params_str = json.dumps(raw_params) if isinstance(raw_params, (dict, list)) else raw_params
 
             conn.execute(
                 sa.text(
@@ -153,7 +159,7 @@ def upgrade() -> None:
                     "agent_id": agent[0],
                     "auth_type": agent[1],
                     "auth_value": agent[2],
-                    "auth_query_params": agent[3],
+                    "auth_query_params": params_str,
                 },
             )
 

mcpgateway/alembic/versions/a2a_v1_domain_models_3f7e9d1a2b4c.py

@@ -29,6 +29,7 @@ def upgrade() -> None:
     op.create_table(
         "a2a_task_events",
         sa.Column("id", sa.String(36), primary_key=True, nullable=False),
+        sa.Column("a2a_agent_id", sa.String(36), sa.ForeignKey("a2a_agents.id", ondelete="CASCADE"), nullable=True),
         sa.Column("task_id", sa.String(255), nullable=False),
         sa.Column("event_id", sa.String(36), nullable=False),
         sa.Column("sequence", sa.BigInteger(), nullable=False),
@@ -38,6 +39,7 @@ def upgrade() -> None:
     )
     op.create_index("ix_a2a_task_events_task_id", "a2a_task_events", ["task_id"])
     op.create_index("ix_a2a_task_events_task_seq", "a2a_task_events", ["task_id", "sequence"])
+    op.create_index("ix_a2a_task_events_agent_id", "a2a_task_events", ["a2a_agent_id"])
 
 
 def downgrade() -> None:

mcpgateway/alembic/versions/ffe4494639d3_add_a2a_task_events_table.py

@@ -1258,8 +1258,6 @@ class Permissions:
     TOOLS_EXECUTE = "tools.execute"
     TOOLS_MANAGE_PLUGINS = "tools.manage_plugins"
 
-    TOOLS_MANAGE_PLUGINS = "tools.manage_plugins"
-
     # Resource permissions
     RESOURCES_CREATE = "resources.create"
     RESOURCES_READ = "resources.read"
@@ -5066,6 +5064,7 @@ class A2ATaskEvent(Base):
     __tablename__ = "a2a_task_events"
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    a2a_agent_id: Mapped[Optional[str]] = mapped_column(String(36), ForeignKey("a2a_agents.id", ondelete="CASCADE"), nullable=True, index=True)
     task_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
     event_id: Mapped[str] = mapped_column(String(36), nullable=False)
     sequence: Mapped[int] = mapped_column(BigInteger, nullable=False)

mcpgateway/db.py

@@ -8997,10 +8997,14 @@ async def handle_internal_a2a_agent_resolve(request: Request, agent_name: str):
 
         return ORJSONResponse(status_code=200, content=result)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9044,10 +9048,14 @@ async def handle_internal_a2a_agent_card(request: Request, agent_name: str):
             return ORJSONResponse(status_code=404, content={"error": f"agent '{agent_name}' not found"})
         return ORJSONResponse(status_code=200, content=card)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9080,10 +9088,14 @@ async def handle_internal_a2a_tasks_get(request: Request):
             return ORJSONResponse(status_code=404, content={"error": f"task '{task_id}' not found"})
         return ORJSONResponse(status_code=200, content=task)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9116,10 +9128,14 @@ async def handle_internal_a2a_tasks_list(request: Request):
         tasks = service.list_tasks(db, agent_id=agent_id, state=state, limit=limit, offset=offset, user_email=user_email, token_teams=token_teams)
         return ORJSONResponse(status_code=200, content={"tasks": tasks})
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9152,10 +9168,14 @@ async def handle_internal_a2a_tasks_cancel(request: Request):
             return ORJSONResponse(status_code=404, content={"error": f"task '{task_id}' not found"})
         return ORJSONResponse(status_code=200, content=task)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9174,20 +9194,31 @@ async def handle_internal_a2a_push_create(request: Request):
         if not body.get("a2a_agent_id") or not body.get("task_id") or not body.get("webhook_url"):
             return ORJSONResponse(status_code=400, content={"error": "a2a_agent_id, task_id, and webhook_url are required"})
 
+        # Validate webhook URL through the schema to enforce SSRF protection.
         # First-Party
+        from mcpgateway.schemas import A2APushNotificationConfigCreate  # pylint: disable=import-outside-toplevel
         from mcpgateway.services.a2a_service import A2AAgentService  # pylint: disable=import-outside-toplevel
 
+        try:
+            validated = A2APushNotificationConfigCreate(**body)
+        except Exception as validation_err:
+            return ORJSONResponse(status_code=400, content={"error": f"invalid push config: {validation_err}"})
+
         user_email, token_teams = _get_internal_a2a_scope_context(request)
         service = A2AAgentService()
         if not service._check_agent_access_by_id(db, body["a2a_agent_id"], user_email, token_teams):  # pylint: disable=protected-access
             return ORJSONResponse(status_code=404, content={"error": "agent not found"})
-        cfg = service.create_push_config(db, body)
+        cfg = service.create_push_config(db, validated.model_dump())
         return ORJSONResponse(status_code=200, content=cfg)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9213,17 +9244,21 @@ async def handle_internal_a2a_push_get(request: Request):
 
         user_email, token_teams = _get_internal_a2a_scope_context(request)
         service = A2AAgentService()
-        if agent_id and not service._check_agent_access_by_id(db, agent_id, user_email, token_teams):  # pylint: disable=protected-access
-            return ORJSONResponse(status_code=404, content={"error": f"push config for task '{task_id}' not found"})
         cfg = service.get_push_config(db, task_id, agent_id=agent_id)
         if cfg is None:
             return ORJSONResponse(status_code=404, content={"error": f"push config for task '{task_id}' not found"})
+        if not service._check_agent_access_by_id(db, cfg["a2a_agent_id"], user_email, token_teams):  # pylint: disable=protected-access
+            return ORJSONResponse(status_code=404, content={"error": f"push config for task '{task_id}' not found"})
         return ORJSONResponse(status_code=200, content=cfg)
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9247,15 +9282,19 @@ async def handle_internal_a2a_push_list(request: Request):
 
         user_email, token_teams = _get_internal_a2a_scope_context(request)
         service = A2AAgentService()
-        if agent_id and not service._check_agent_access_by_id(db, agent_id, user_email, token_teams):  # pylint: disable=protected-access
-            return ORJSONResponse(status_code=200, content={"configs": []})
         configs = service.list_push_configs(db, agent_id=agent_id, task_id=task_id)
-        return ORJSONResponse(status_code=200, content={"configs": configs})
+        # Filter configs to only those whose owning agent is visible to the caller.
+        visible = [c for c in configs if service._check_agent_access_by_id(db, c["a2a_agent_id"], user_email, token_teams)]  # pylint: disable=protected-access
+        return ORJSONResponse(status_code=200, content={"configs": visible})
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9289,10 +9328,14 @@ async def handle_internal_a2a_push_delete(request: Request):
             return ORJSONResponse(status_code=404, content={"error": f"push config '{config_id}' not found"})
         return ORJSONResponse(status_code=200, content={"deleted": True})
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9313,16 +9356,32 @@ async def handle_internal_a2a_events_flush(request: Request):
             return ORJSONResponse(status_code=200, content={"count": 0})
 
         # First-Party
+        from mcpgateway.db import A2ATask as DbA2ATask  # pylint: disable=import-outside-toplevel
         from mcpgateway.services.a2a_service import A2AAgentService  # pylint: disable=import-outside-toplevel
 
+        user_email, token_teams = _get_internal_a2a_scope_context(request)
         service = A2AAgentService()
+
+        # Verify the caller has access to the agents that own the referenced tasks.
+        task_ids = {e["task_id"] for e in events if "task_id" in e}
+        if task_ids:
+            tasks = db.query(DbA2ATask).filter(DbA2ATask.task_id.in_(task_ids)).all()
+            agent_ids = {t.a2a_agent_id for t in tasks}
+            for agent_id in agent_ids:
+                if not service._check_agent_access_by_id(db, agent_id, user_email, token_teams):  # pylint: disable=protected-access
+                    return ORJSONResponse(status_code=403, content={"error": "access denied for one or more referenced tasks"})
+
         count = service.flush_events(db, events)
         return ORJSONResponse(status_code=200, content={"count": count})
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()
@@ -9351,15 +9410,21 @@ async def handle_internal_a2a_events_replay(request: Request):
         user_email, token_teams = _get_internal_a2a_scope_context(request)
         service = A2AAgentService()
         task_row = db.query(DbA2ATask).filter(DbA2ATask.task_id == task_id).first()
-        if task_row and not service._check_agent_access_by_id(db, task_row.a2a_agent_id, user_email, token_teams):  # pylint: disable=protected-access
+        if task_row is None:
+            return ORJSONResponse(status_code=404, content={"error": "task not found"})
+        if not service._check_agent_access_by_id(db, task_row.a2a_agent_id, user_email, token_teams):  # pylint: disable=protected-access
             return ORJSONResponse(status_code=404, content={"error": "task not found"})
         events = service.replay_events(db, task_id, after_sequence, limit=limit)
         return ORJSONResponse(status_code=200, content={"events": events})
     except Exception:
+        logger.exception("Internal A2A endpoint error")
         try:
             db.rollback()
         except Exception:
-            pass
+            try:
+                db.invalidate()
+            except Exception:
+                pass  # nosec B110
         raise
     finally:
         db.close()

mcpgateway/main.py

@@ -5575,6 +5575,12 @@ class A2APushNotificationConfigCreate(BaseModel):
     events: Optional[List[str]] = None
     enabled: bool = True
 
+    @field_validator("webhook_url")
+    @classmethod
+    def validate_webhook_url(cls, v: str) -> str:
+        """Validate webhook URL for scheme, SSRF, and dangerous patterns."""
+        return validate_core_url(v, "Webhook URL")
+
 
 class A2APushNotificationConfigRead(BaseModel):
     """Schema for reading a push notification webhook configuration."""
@@ -5585,7 +5591,7 @@ class A2APushNotificationConfigRead(BaseModel):
     a2a_agent_id: str
     task_id: str
     webhook_url: str
-    auth_token: Optional[str] = None
+    auth_token: Optional[str] = Field(default=None, exclude=True)
     events: Optional[List[str]] = None
     enabled: bool
     created_at: datetime
@@ -5595,6 +5601,7 @@ class A2APushNotificationConfigRead(BaseModel):
 class A2ATaskEventCreate(BaseModel):
     """Schema for creating a task event log entry."""
 
+    a2a_agent_id: Optional[str] = None
     task_id: str
     event_id: str
     sequence: int
@@ -5608,6 +5615,7 @@ class A2ATaskEventRead(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: str
+    a2a_agent_id: Optional[str] = None
     task_id: str
     event_id: str
     sequence: int

mcpgateway/schemas.py

@@ -393,7 +393,7 @@ def prepare_a2a_invocation(
                 decrypted = decode_auth(encrypted_value)
                 auth_query_params_decrypted[str(param_key)] = str(decrypted.get(param_key, ""))
             except Exception:  # nosec B112
-                logger.debug("Failed to decrypt query param %r for A2A agent invocation", param_key, exc_info=True)
+                logger.warning("Failed to decrypt query param %r for A2A agent invocation — invocation proceeds without this credential", param_key, exc_info=True)
                 continue
         if auth_query_params_decrypted:
             target_endpoint_url = apply_query_param_auth(target_endpoint_url, auth_query_params_decrypted)

mcpgateway/services/a2a_protocol.py

@@ -39,6 +39,7 @@ def _check_server_access(server: DbServer, user_email: Optional[str], token_team
         return True
 
     if token_teams is None and user_email is None:
+        logger.debug("admin bypass: granting access to %s server %s", server.visibility, getattr(server, "name", "?"))
         return True
 
     if not user_email:
@@ -313,8 +314,11 @@ def resolve_task_mapping(self, db: Session, server_id: str, server_task_id: str)
     def _find_a2a_interface(self, db: Session, server_id: str) -> Optional[DbServerInterface]:
         """Return the first enabled ServerInterface with an A2A protocol for *server_id*.
 
-        The match is case-insensitive on the protocol field (e.g. "a2a", "A2A",
-        "a2a/v1" all qualify).
+        Matches any protocol whose lower-cased value starts with ``a2a``
+        (e.g. ``a2a``, ``a2a/v1``, ``a2a-jsonrpc``).  When a server exposes
+        multiple A2A interfaces, the first one (by DB insertion order) is
+        returned — this avoids ``MultipleResultsFound`` when both v0.3 and
+        v1 bindings exist.
 
         Args:
             db: Database session.
@@ -323,9 +327,14 @@ def _find_a2a_interface(self, db: Session, server_id: str) -> Optional[DbServerI
         Returns:
             Matching ServerInterface ORM instance, or None.
         """
-        query = select(DbServerInterface).where(
-            DbServerInterface.server_id == server_id,
-            DbServerInterface.enabled == True,  # noqa: E712
-            func.lower(DbServerInterface.protocol).in_(["a2a", "a2a/v1", "a2a/v0.3"]),
+        query = (
+            select(DbServerInterface)
+            .where(
+                DbServerInterface.server_id == server_id,
+                DbServerInterface.enabled == True,  # noqa: E712
+                func.lower(DbServerInterface.protocol).like("a2a%"),
+            )
+            .order_by(DbServerInterface.created_at.desc())
+            .limit(1)
         )
         return db.execute(query).scalar_one_or_none()