diff --git a/openhands/automation/capabilities_router.py b/openhands/automation/capabilities_router.py index f03f391..36881f1 100644 --- a/openhands/automation/capabilities_router.py +++ b/openhands/automation/capabilities_router.py @@ -6,17 +6,29 @@ Neither endpoint writes. """ +import asyncio import fnmatch import logging import uuid +from dataclasses import dataclass +from typing import Any, Literal +from urllib.parse import quote, urlparse from zoneinfo import available_timezones -from fastapi import APIRouter, Depends, HTTPException +import httpx +from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import ValidationError from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from openhands.automation.auth import AuthenticatedUser, authenticate_request +from openhands.automation.auth import ( + MAX_SESSION_COOKIE_CHUNKS, + SESSION_COOKIE_NAME, + AuthenticatedUser, + AuthMethod, + authenticate_request, + get_http_client, +) from openhands.automation.config import get_config from openhands.automation.db import get_session from openhands.automation.event_schemas import parse_event @@ -38,6 +50,8 @@ DraftValidationError, EventCapabilities, EventTrigger, + PreflightIntegrationAlternative, + PreflightIntegrationRequirement, TriggerCapabilities, ValidateDraftRequest, ValidateDraftResponse, @@ -74,6 +88,46 @@ # Tags Pydantic inserts into an error location for the trigger union. _TRIGGER_TAGS = frozenset({"cron", "event"}) +_MCP_PROBE_TIMEOUT_SECONDS = 15.0 +_MCP_PROBE_REQUEST_TIMEOUT_SECONDS = _MCP_PROBE_TIMEOUT_SECONDS + 5.0 +_PREFLIGHT_TOTAL_TIMEOUT_SECONDS = 60.0 +_MAX_PREFLIGHT_SEARCH_PAGES = 10 +# The Cloud git route requests one look-ahead item from providers to decide +# whether to return ``next_page_id``. Keeping this below provider max 100 lets +# that look-ahead survive instead of being clamped away. +_CLOUD_REPOSITORY_SEARCH_PAGE_SIZE = 99 +_LOCAL_REPOSITORY_SECRET_NAMES = { + "github": "github_token", + "gitlab": "gitlab_token", + "bitbucket": "bitbucket_token", +} +_REPOSITORY_PROVIDER_HOSTS = { + "github": "github.com", + "gitlab": "gitlab.com", + "bitbucket": "bitbucket.org", +} + + +class _DependencyUnavailable(Exception): + """A trusted validation dependency did not return a usable answer.""" + + +@dataclass(frozen=True) +class _PreflightTarget: + base_url: str + headers: dict[str, str] + local: bool + + +@dataclass(frozen=True) +class _StoredMCPServer: + name: str + raw: dict[str, Any] + transport: Literal["stdio", "shttp", "sse"] + locator: str + auth_strategy: str + enabled: bool + @router.get("/capabilities", response_model_exclude_none=True) async def get_capabilities( @@ -140,8 +194,10 @@ async def get_capabilities( @router.post("/validate") async def validate_draft( body: ValidateDraftRequest, + request: Request, user: AuthenticatedUser = Depends(authenticate_request), session: AsyncSession = Depends(get_session), + client: httpx.AsyncClient = Depends(get_http_client), ) -> ValidateDraftResponse: """Validate a draft automation without creating it. @@ -158,6 +214,33 @@ async def validate_draft( except ValidationError as e: return ValidateDraftResponse(valid=False, errors=_schema_errors(e)) + try: + async with asyncio.timeout(_PREFLIGHT_TOTAL_TIMEOUT_SECONDS): + return await _validated_draft_response( + body=body, + draft=draft, + request=request, + user=user, + session=session, + client=client, + ) + except (TimeoutError, _DependencyUnavailable): + logger.warning("A preflight validation dependency is unavailable") + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Preflight validation is temporarily unavailable.", + ) + + +async def _validated_draft_response( + *, + body: ValidateDraftRequest, + draft: DraftModel, + request: Request, + user: AuthenticatedUser, + session: AsyncSession, + client: httpx.AsyncClient, +) -> ValidateDraftResponse: errors: list[DraftValidationError] = [] sample_event_matched: bool | None = None @@ -210,6 +293,17 @@ async def validate_draft( trigger, trigger.source, event.event_key, body.sample_event ) + if body.requirements is not None: + errors.extend( + await _deployment_preflight_errors( + body=body, + draft=draft, + user=user, + request=request, + client=client, + ) + ) + return ValidateDraftResponse( valid=not errors, errors=errors, @@ -217,6 +311,733 @@ async def validate_draft( ) +async def _deployment_preflight_errors( + *, + body: ValidateDraftRequest, + draft: DraftModel, + user: AuthenticatedUser, + request: Request, + client: httpx.AsyncClient, +) -> list[DraftValidationError]: + requirements = body.requirements + if requirements is None: + return [] + + target = _preflight_target(request, user) + requested_secret_names = { + name + for requirement in requirements.integrations + for alternative in requirement.alternatives + for name in alternative.secret_names + } + if target.local and draft.repos: + requested_secret_names.update(_LOCAL_REPOSITORY_SECRET_NAMES.values()) + + available_secret_names = await _available_secret_names( + target, + requested_secret_names, + client, + ) + errors: list[DraftValidationError] = [] + + if requirements.integrations: + servers = await _stored_mcp_servers(target, client) + for requirement in requirements.integrations: + errors.extend( + await _integration_errors( + requirement, + servers, + available_secret_names, + target, + client, + ) + ) + + if draft.repos: + for index, repository in enumerate(draft.repos): + errors.extend( + await _repository_errors( + repository=repository, + index=index, + requirements=requirements.integrations, + available_secret_names=available_secret_names, + target=target, + client=client, + ) + ) + + return errors + + +def _preflight_target(request: Request, user: AuthenticatedUser) -> _PreflightTarget: + settings = get_config().service + if settings.is_local_mode: + headers = {} + if settings.agent_server_api_key: + headers["X-Session-API-Key"] = settings.agent_server_api_key + return _PreflightTarget( + base_url=settings.agent_server_url.rstrip("/"), + headers=headers, + local=True, + ) + + if user.auth_method == AuthMethod.API_KEY and user.api_key: + headers = {"Authorization": f"Bearer {user.api_key}"} + elif user.auth_method == AuthMethod.COOKIE: + cookies: list[str] = [] + for index in range(MAX_SESSION_COOKIE_CHUNKS): + name = ( + SESSION_COOKIE_NAME if index == 0 else f"{SESSION_COOKIE_NAME}_{index}" + ) + value = request.cookies.get(name) + if value is None: + break + cookies.append(f"{name}={value}") + if not cookies: + raise _DependencyUnavailable + headers = {"Cookie": "; ".join(cookies)} + else: + raise _DependencyUnavailable + + return _PreflightTarget( + base_url=settings.openhands_api_base_url.rstrip("/"), + headers=headers, + local=False, + ) + + +async def _send_preflight_request( + client: httpx.AsyncClient, + method: str, + url: str, + *, + headers: dict[str, str], + **kwargs: Any, +) -> httpx.Response: + try: + response = await client.request(method, url, headers=headers, **kwargs) + except httpx.RequestError: + raise _DependencyUnavailable from None + if response.status_code == status.HTTP_429_TOO_MANY_REQUESTS: + raise _DependencyUnavailable + if response.status_code >= status.HTTP_500_INTERNAL_SERVER_ERROR: + raise _DependencyUnavailable + return response + + +def _json_object(response: httpx.Response) -> dict[str, Any]: + try: + value = response.json() + except ValueError: + raise _DependencyUnavailable from None + if not isinstance(value, dict): + raise _DependencyUnavailable + return value + + +async def _available_secret_names( + target: _PreflightTarget, + requested_names: set[str], + client: httpx.AsyncClient, +) -> set[str]: + if not requested_names: + return set() + + if target.local: + response = await _send_preflight_request( + client, + "GET", + f"{target.base_url}/api/settings/secrets", + headers=target.headers, + ) + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + items = _json_object(response).get("secrets") + if not isinstance(items, list): + raise _DependencyUnavailable + names: set[str] = set() + for item in items: + if not isinstance(item, dict) or not isinstance(item.get("name"), str): + raise _DependencyUnavailable + names.add(item["name"]) + return names.intersection(requested_names) + + found: set[str] = set() + for name in sorted(requested_names): + page_id: str | None = None + seen_page_ids: set[str] = set() + for _ in range(_MAX_PREFLIGHT_SEARCH_PAGES): + params: dict[str, str | int] = { + "name__contains": name, + "limit": 100, + } + if page_id is not None: + params["page_id"] = page_id + response = await _send_preflight_request( + client, + "GET", + f"{target.base_url}/api/v1/secrets/search", + headers=target.headers, + params=params, + ) + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + data = _json_object(response) + items = data.get("items") + if not isinstance(items, list): + raise _DependencyUnavailable + item_names: list[str] = [] + for item in items: + if not isinstance(item, dict) or not isinstance(item.get("name"), str): + raise _DependencyUnavailable + item_names.append(item["name"]) + if name in item_names: + found.add(name) + break + next_page_id = data.get("next_page_id") + if next_page_id is None: + break + if ( + not isinstance(next_page_id, str) + or not next_page_id + or next_page_id in seen_page_ids + ): + raise _DependencyUnavailable + seen_page_ids.add(next_page_id) + page_id = next_page_id + else: + raise _DependencyUnavailable + return found + + +async def _stored_mcp_servers( + target: _PreflightTarget, + client: httpx.AsyncClient, +) -> list[_StoredMCPServer]: + headers = dict(target.headers) + path = "/api/settings" if target.local else "/api/v1/settings" + if target.local: + headers["X-Expose-Secrets"] = "encrypted" + response = await _send_preflight_request( + client, + "GET", + f"{target.base_url}{path}", + headers=headers, + ) + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + agent_settings = _json_object(response).get("agent_settings") + if not isinstance(agent_settings, dict): + raise _DependencyUnavailable + raw_config = agent_settings.get("mcp_config", {}) + if isinstance(raw_config, dict) and "mcpServers" in raw_config: + raw_config = raw_config["mcpServers"] + if not isinstance(raw_config, dict): + raise _DependencyUnavailable + + servers: list[_StoredMCPServer] = [] + for name, raw in raw_config.items(): + if not isinstance(name, str) or not isinstance(raw, dict): + raise _DependencyUnavailable + servers.append(_parse_stored_mcp_server(name, raw)) + return servers + + +def _parse_stored_mcp_server(name: str, raw: dict[str, Any]) -> _StoredMCPServer: + enabled = raw.get("enabled", True) + if not isinstance(enabled, bool): + raise _DependencyUnavailable + + raw_transport = raw.get("transport", raw.get("type")) + if raw_transport is None: + raw_transport = "stdio" if isinstance(raw.get("command"), str) else "http" + if raw_transport == "stdio": + transport: Literal["stdio", "shttp", "sse"] = "stdio" + locator = name + elif raw_transport in {"http", "streamable-http", "shttp"}: + transport = "shttp" + locator = raw.get("url") + elif raw_transport == "sse": + transport = "sse" + locator = raw.get("url") + else: + raise _DependencyUnavailable + if not isinstance(locator, str) or not locator: + raise _DependencyUnavailable + + raw_auth = raw.get("auth") + if raw_auth is None: + auth_strategy = "none" + elif isinstance(raw_auth, dict) and isinstance(raw_auth.get("strategy"), str): + auth_strategy = raw_auth["strategy"] + else: + raise _DependencyUnavailable + return _StoredMCPServer( + name=name, + raw=raw, + transport=transport, + locator=locator, + auth_strategy=auth_strategy, + enabled=enabled, + ) + + +def _normalized_remote_locator(value: str) -> tuple[str, str, int | None, str] | None: + try: + parsed = urlparse(value) + port = parsed.port + except ValueError: + return None + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + return None + if (parsed.scheme == "http" and port == 80) or ( + parsed.scheme == "https" and port == 443 + ): + port = None + return ( + parsed.scheme.lower(), + parsed.hostname.lower(), + port, + parsed.path.rstrip("/"), + ) + + +def _alternative_matches_server( + alternative: PreflightIntegrationAlternative, + server: _StoredMCPServer, +) -> bool: + if alternative.transport != server.transport: + return False + if alternative.transport == "stdio": + if alternative.locator != server.locator: + return False + elif _normalized_remote_locator(alternative.locator) != _normalized_remote_locator( + server.locator + ): + return False + + expected_auth = alternative.auth_strategy + if expected_auth is None: + return True + if server.auth_strategy == "header" and expected_auth == "none": + return True + return server.auth_strategy == expected_auth + + +async def _integration_errors( + requirement: PreflightIntegrationRequirement, + servers: list[_StoredMCPServer], + available_secret_names: set[str], + target: _PreflightTarget, + client: httpx.AsyncClient, +) -> list[DraftValidationError]: + matches = [ + (alternative, server) + for alternative in requirement.alternatives + for server in servers + if _alternative_matches_server(alternative, server) + ] + enabled_matches = [(alt, server) for alt, server in matches if server.enabled] + if not enabled_matches: + code = "integration_disabled" if matches else "integration_not_configured" + message = ( + f"Enable the configured {requirement.id} connection before continuing." + if matches + else f"Connect {requirement.id} before continuing." + ) + return [ + DraftValidationError( + field=None, + code=code, + message=message, + step="prerequisites", + ) + ] + + missing_names: list[str] = [] + probe_candidates: list[_StoredMCPServer] = [] + for alternative, server in enabled_matches: + missing = [ + name + for name in alternative.secret_names + if name not in available_secret_names + ] + if missing: + for name in missing: + if name not in missing_names: + missing_names.append(name) + continue + if all(candidate.name != server.name for candidate in probe_candidates): + probe_candidates.append(server) + + if not probe_candidates: + return [ + DraftValidationError( + field=None, + code="credential_missing", + message=f"Add the required credential '{name}' before continuing.", + step="prerequisites", + ) + for name in missing_names + ] + + dependency_failed = False + for server in probe_candidates: + try: + if await _probe_mcp_server(target, server, client): + return [] + except _DependencyUnavailable: + dependency_failed = True + if dependency_failed: + raise _DependencyUnavailable + return [ + DraftValidationError( + field=None, + code="integration_unavailable", + message=( + f"The {requirement.id} connection could not be reached. " + "Reconnect it and try again." + ), + step="prerequisites", + ) + ] + + +async def _probe_mcp_server( + target: _PreflightTarget, + server: _StoredMCPServer, + client: httpx.AsyncClient, +) -> bool: + if target.local: + path = "/api/mcp/test" + payload = { + "name": server.name, + "server": server.raw, + "timeout": _MCP_PROBE_TIMEOUT_SECONDS, + } + else: + path = f"/api/v1/settings/mcp/{quote(server.name, safe='')}/test" + payload = {"timeout": _MCP_PROBE_TIMEOUT_SECONDS} + response = await _send_preflight_request( + client, + "POST", + f"{target.base_url}{path}", + headers=target.headers, + json=payload, + timeout=_MCP_PROBE_REQUEST_TIMEOUT_SECONDS, + ) + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + ok = _json_object(response).get("ok") + if not isinstance(ok, bool): + raise _DependencyUnavailable + return ok + + +def _repository_parts(repository: Any) -> tuple[str, str] | None: + try: + provider = repository.get_provider().value + except ValueError: + return None + + raw_url = repository.url + if raw_url.startswith("git@"): + user_host, separator, identifier = raw_url.partition(":") + if not separator or "@" not in user_host: + return None + host = user_host.rsplit("@", 1)[1].lower() + if host != _REPOSITORY_PROVIDER_HOSTS[provider]: + return None + elif "://" in raw_url: + try: + parsed = urlparse(raw_url) + host = parsed.hostname + port = parsed.port + except ValueError: + return None + if ( + parsed.scheme != "https" + or host is None + or host.lower() != _REPOSITORY_PROVIDER_HOSTS[provider] + or port is not None + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + return None + identifier = parsed.path + else: + identifier = raw_url + identifier = identifier.strip("/") + if identifier.endswith(".git"): + identifier = identifier[:-4] + if not identifier or "/" not in identifier: + return None + return provider, identifier + + +async def _repository_errors( + *, + repository: Any, + index: int, + requirements: list[PreflightIntegrationRequirement], + available_secret_names: set[str], + target: _PreflightTarget, + client: httpx.AsyncClient, +) -> list[DraftValidationError]: + parts = _repository_parts(repository) + if parts is None: + return [ + DraftValidationError( + field=f"repos[{index}].url", + code="repository_provider_unsupported", + message="Choose a repository from a supported Git provider.", + ) + ] + provider, identifier = parts + if target.local: + return await _local_repository_errors( + provider=provider, + identifier=identifier, + ref=repository.ref, + index=index, + requirements=requirements, + available_secret_names=available_secret_names, + target=target, + client=client, + ) + return await _cloud_repository_errors( + provider=provider, + identifier=identifier, + ref=repository.ref, + index=index, + target=target, + client=client, + ) + + +async def _local_repository_errors( + *, + provider: str, + identifier: str, + ref: str | None, + index: int, + requirements: list[PreflightIntegrationRequirement], + available_secret_names: set[str], + target: _PreflightTarget, + client: httpx.AsyncClient, +) -> list[DraftValidationError]: + candidate_names: list[str] = [] + for requirement in requirements: + if requirement.id != provider: + continue + for alternative in requirement.alternatives: + for name in alternative.secret_names: + if name in available_secret_names and name not in candidate_names: + candidate_names.append(name) + canonical_name = _LOCAL_REPOSITORY_SECRET_NAMES[provider] + if ( + canonical_name in available_secret_names + and canonical_name not in candidate_names + ): + candidate_names.append(canonical_name) + + response = await _send_preflight_request( + client, + "POST", + f"{target.base_url}/api/git/validate-repository", + headers=target.headers, + json={ + "provider": provider, + "repository": identifier, + "ref": ref, + "credential_names": candidate_names[:5], + }, + ) + if response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT: + return [ + DraftValidationError( + field=f"repos[{index}].ref" if ref else f"repos[{index}].url", + code="repository_reference_invalid", + message="Use a valid repository and Git reference.", + ) + ] + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + verdict = _json_object(response).get("status") + if verdict == "accessible": + return [] + if verdict == "unavailable": + raise _DependencyUnavailable + if verdict == "missing_credentials": + return [ + DraftValidationError( + field=f"repos[{index}].url", + code="repository_credentials_missing", + message=f"Add credentials that can access {identifier}.", + ) + ] + if verdict == "denied": + code = "repository_access_denied" + message = f"Your {provider} connection cannot access {identifier}." + elif verdict == "not_found": + code = "repository_not_accessible" + message = f"The repository {identifier} could not be accessed." + else: + raise _DependencyUnavailable + return [ + DraftValidationError( + field=f"repos[{index}].url", + code=code, + message=message, + ) + ] + + +async def _cloud_repository_errors( + *, + provider: str, + identifier: str, + ref: str | None, + index: int, + target: _PreflightTarget, + client: httpx.AsyncClient, +) -> list[DraftValidationError]: + repository_found = False + page_id: str | None = None + seen_page_ids: set[str] = set() + for _ in range(_MAX_PREFLIGHT_SEARCH_PAGES): + params: dict[str, str | int] = { + "provider": provider, + "query": identifier, + "limit": _CLOUD_REPOSITORY_SEARCH_PAGE_SIZE, + } + if page_id is not None: + params["page_id"] = page_id + response = await _send_preflight_request( + client, + "GET", + f"{target.base_url}/api/v1/git/repositories/search", + headers=target.headers, + params=params, + ) + if response.status_code == status.HTTP_403_FORBIDDEN: + return [ + DraftValidationError( + field=f"repos[{index}].url", + code="repository_provider_not_connected", + message=f"Connect {provider} before choosing a repository.", + ) + ] + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + data = _json_object(response) + items = data.get("items") + if not isinstance(items, list): + raise _DependencyUnavailable + names: list[str] = [] + for item in items: + if not isinstance(item, dict) or not isinstance(item.get("full_name"), str): + raise _DependencyUnavailable + names.append(item["full_name"]) + if identifier.casefold() in {name.casefold() for name in names}: + repository_found = True + break + next_page_id = data.get("next_page_id") + if next_page_id is None: + break + if ( + not isinstance(next_page_id, str) + or not next_page_id + or next_page_id in seen_page_ids + ): + raise _DependencyUnavailable + seen_page_ids.add(next_page_id) + page_id = next_page_id + else: + raise _DependencyUnavailable + + if not repository_found: + return [ + DraftValidationError( + field=f"repos[{index}].url", + code="repository_not_accessible", + message=f"The repository {identifier} could not be accessed.", + ) + ] + if ref is None: + return [] + + ref_matches = False + page_id = None + seen_page_ids = set() + for _ in range(_MAX_PREFLIGHT_SEARCH_PAGES): + params = { + "provider": provider, + "repository": identifier, + "query": ref, + "limit": 100, + } + if page_id is not None: + params["page_id"] = page_id + response = await _send_preflight_request( + client, + "GET", + f"{target.base_url}/api/v1/git/branches/search", + headers=target.headers, + params=params, + ) + if response.status_code == status.HTTP_403_FORBIDDEN: + return [ + DraftValidationError( + field=f"repos[{index}].url", + code="repository_provider_not_connected", + message=f"Reconnect {provider} to check this repository.", + ) + ] + if response.status_code != status.HTTP_200_OK: + raise _DependencyUnavailable + data = _json_object(response) + branch_items = data.get("items") + if not isinstance(branch_items, list): + raise _DependencyUnavailable + for item in branch_items: + if not isinstance(item, dict): + raise _DependencyUnavailable + name = item.get("name") + commit_sha = item.get("commit_sha") + if not isinstance(name, str) or not isinstance(commit_sha, str): + raise _DependencyUnavailable + if name == ref or commit_sha.casefold().startswith(ref.casefold()): + ref_matches = True + break + if ref_matches: + break + next_page_id = data.get("next_page_id") + if next_page_id is None: + break + if ( + not isinstance(next_page_id, str) + or not next_page_id + or next_page_id in seen_page_ids + ): + raise _DependencyUnavailable + seen_page_ids.add(next_page_id) + page_id = next_page_id + else: + raise _DependencyUnavailable + if ref_matches: + return [] + return [ + DraftValidationError( + field=f"repos[{index}].ref", + code="repository_ref_not_accessible", + message=f"The Git reference '{ref}' could not be accessed.", + ) + ] + + async def _custom_sources(org_id: uuid.UUID, session: AsyncSession) -> list[str]: """Custom webhook sources this organization has enabled. diff --git a/openhands/automation/schemas.py b/openhands/automation/schemas.py index c7ef1c5..e892837 100644 --- a/openhands/automation/schemas.py +++ b/openhands/automation/schemas.py @@ -3,9 +3,18 @@ import re import uuid from enum import StrEnum -from typing import Annotated, Any, Literal - -from pydantic import BaseModel, ConfigDict, Discriminator, Field, Tag, field_validator +from typing import Annotated, Any, Literal, Self +from urllib.parse import urlparse + +from pydantic import ( + BaseModel, + ConfigDict, + Discriminator, + Field, + Tag, + field_validator, + model_validator, +) from pydantic.alias_generators import to_camel from openhands.automation.constants import MODEL_PROFILE_PATTERN @@ -744,6 +753,109 @@ class CapabilitiesResponse(_SetupContractModel): features: list[str] +PreflightIntegrationTransport = Literal["stdio", "shttp", "sse"] +PreflightIntegrationAuthStrategy = Literal[ + "none", "api_key", "bearer", "basic", "oauth2" +] + +_PREFLIGHT_IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") +_PREFLIGHT_SECRET_NAME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,63}$") +_MAX_PREFLIGHT_SECRET_REFERENCES = 32 + + +class PreflightIntegrationAlternative(_SetupContractModel): + """One deployment-checkable way to satisfy an integration requirement.""" + + model_config = ConfigDict(extra="forbid") + + transport: PreflightIntegrationTransport + locator: str = Field( + ..., + min_length=1, + max_length=2048, + description="Stored MCP server name or remote URL; never a secret value", + ) + auth_strategy: PreflightIntegrationAuthStrategy | None = None + secret_names: list[str] = Field(default_factory=list, max_length=8) + + @field_validator("secret_names") + @classmethod + def validate_secret_names(cls, names: list[str]) -> list[str]: + if len(names) != len(set(names)): + raise ValueError("secretNames must not contain duplicates") + if not all(_PREFLIGHT_SECRET_NAME_PATTERN.fullmatch(name) for name in names): + raise ValueError("secretNames must contain valid secret reference names") + return names + + @model_validator(mode="after") + def validate_locator(self) -> Self: + if self.transport == "stdio": + if len(self.locator) > 128 or not _PREFLIGHT_IDENTIFIER_PATTERN.fullmatch( + self.locator + ): + raise ValueError("stdio locator must be a valid MCP server name") + return self + + parsed = urlparse(self.locator) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + ): + raise ValueError( + "remote locator must be an HTTP(S) URL without credentials" + ) + return self + + +class PreflightIntegrationRequirement(_SetupContractModel): + """A required integration and the catalog alternatives that can provide it.""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., min_length=1, max_length=128) + alternatives: list[PreflightIntegrationAlternative] = Field( + default_factory=list, + max_length=8, + ) + + @field_validator("id") + @classmethod + def validate_id(cls, integration_id: str) -> str: + if not _PREFLIGHT_IDENTIFIER_PATTERN.fullmatch(integration_id): + raise ValueError("integration id must be a bounded identifier") + return integration_id + + +class PreflightRequirements(_SetupContractModel): + """Concrete selected-automation requirements sent by the setup client.""" + + model_config = ConfigDict(extra="forbid") + + integrations: list[PreflightIntegrationRequirement] = Field( + default_factory=list, + max_length=16, + ) + + @model_validator(mode="after") + def validate_bounds(self) -> Self: + ids = [requirement.id for requirement in self.integrations] + if len(ids) != len(set(ids)): + raise ValueError("integration requirements must not contain duplicates") + names = { + name + for requirement in self.integrations + for alternative in requirement.alternatives + for name in alternative.secret_names + } + if len(names) > _MAX_PREFLIGHT_SECRET_REFERENCES: + raise ValueError( + "requirements reference too many distinct credential names" + ) + return self + + class DraftValidationError(_SetupContractModel): """A single problem with a draft, addressed to the field that caused it.""" @@ -756,6 +868,7 @@ class DraftValidationError(_SetupContractModel): ) code: str message: str + step: Literal["prerequisites", "form"] = "form" class ValidateDraftRequest(_SetupContractModel): @@ -781,6 +894,13 @@ class ValidateDraftRequest(_SetupContractModel): "webhook endpoint would receive it." ), ) + requirements: PreflightRequirements | None = Field( + default=None, + description=( + "Selected automation requirements. Omitted by legacy clients; " + "contains public locators and secret names only, never values." + ), + ) class ValidateDraftResponse(_SetupContractModel): diff --git a/tests/test_capabilities_router.py b/tests/test_capabilities_router.py index 60b57bc..a1d1e77 100644 --- a/tests/test_capabilities_router.py +++ b/tests/test_capabilities_router.py @@ -1,14 +1,19 @@ """Tests for the capabilities and preflight validation endpoints.""" +import asyncio import dataclasses +import json import uuid +import httpx import pytest +from sqlalchemy.ext.asyncio import create_async_engine +from openhands.automation import capabilities_router from openhands.automation.app import app from openhands.automation.auth import authenticate_request from openhands.automation.config import clear_config_cache -from openhands.automation.models import CustomWebhook +from openhands.automation.models import Base, CustomWebhook # Test UUID matching mock_authenticated_user fixture @@ -55,6 +60,33 @@ def preflight(draft: dict, **extra: object) -> dict: } +def integration_requirement( + integration_id: str, + *, + transport: str, + locator: str, + auth_strategy: str = "none", + secret_names: list[str] | None = None, +) -> dict: + """Build one integration requirement in the public setup contract.""" + alternative: dict[str, object] = { + "transport": transport, + "locator": locator, + "authStrategy": auth_strategy, + } + if secret_names: + alternative["secretNames"] = secret_names + return {"id": integration_id, "alternatives": [alternative]} + + +async def install_outbound_transport(handler) -> None: + """Replace the app's dependency client with a deterministic mock transport.""" + await app.state.http_client.aclose() + app.state.http_client = httpx.AsyncClient( + transport=httpx.MockTransport(handler), + ) + + def comment_event(body: str) -> dict: """A GitHub issue_comment payload carrying the given comment text.""" return { @@ -84,6 +116,16 @@ def reset_config_cache(): clear_config_cache() +@pytest.fixture +async def async_engine(): + """Capabilities tests need SQLAlchemy, not an external Postgres service.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + yield engine + await engine.dispose() + + @pytest.fixture def ready_deployment(monkeypatch): """A deployment that can mint the API key every run needs. @@ -376,3 +418,643 @@ async def test_unknown_creation_endpoint_is_rejected(self, async_client): ) assert response.status_code == 422 + + async def test_required_secret_and_integration_failures_are_step_addressable( + self, async_client + ): + """Every unsatisfied prerequisite points back to the setup step.""" + slack_url = "https://mcp.example.test/slack" + + def outbound(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/secrets/search": + return httpx.Response(200, json={"items": [], "next_page_id": None}) + if request.url.path == "/api/v1/settings": + return httpx.Response( + 200, + json={ + "agent_settings": { + "mcp_config": { + "slack": { + "transport": "http", + "url": slack_url, + "auth": { + "strategy": "api_key", + "value": "**********", + }, + } + } + } + }, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + draft = {**CRON_DRAFT, "repos": None} + response = await async_client.post( + VALIDATE_URL, + json=preflight( + draft, + requirements={ + "integrations": [ + integration_requirement( + "slack", + transport="shttp", + locator=slack_url, + auth_strategy="api_key", + secret_names=["SLACK_BOT_TOKEN"], + ), + integration_requirement( + "postgres", + transport="stdio", + locator="postgres", + ), + ] + }, + ), + ) + + assert response.status_code == 200 + body = response.json() + assert body["valid"] is False + assert [(error["code"], error["step"]) for error in body["errors"]] == [ + ("credential_missing", "prerequisites"), + ("integration_not_configured", "prerequisites"), + ] + assert "SLACK_BOT_TOKEN" in body["errors"][0]["message"] + + async def test_cloud_preflight_probes_matching_integration_without_secret_leakage( + self, async_client + ): + """Cloud probes the stored server while only returning a coarse verdict.""" + mcp_url = "https://mcp.example.test/github" + sentinel = "provider-secret-should-never-leak" + seen_probe = False + + def outbound(request: httpx.Request) -> httpx.Response: + nonlocal seen_probe + if request.url.path == "/api/v1/secrets/search": + return httpx.Response( + 200, + json={ + "items": [{"name": "GITHUB_TOKEN", "description": sentinel}], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/settings": + return httpx.Response( + 200, + json={ + "agent_settings": { + "mcp_config": { + "github": { + "transport": "http", + "url": mcp_url, + "auth": { + "strategy": "api_key", + "value": "**********", + }, + } + } + } + }, + ) + if request.url.path == "/api/v1/settings/mcp/github/test": + seen_probe = True + assert request.extensions["timeout"]["read"] > 15 + return httpx.Response( + 200, + json={"ok": False, "error": sentinel, "error_kind": "connection"}, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight( + {**CRON_DRAFT, "repos": None}, + requirements={ + "integrations": [ + integration_requirement( + "github", + transport="shttp", + locator=mcp_url, + auth_strategy="api_key", + secret_names=["GITHUB_TOKEN"], + ) + ] + }, + ), + ) + + assert seen_probe is True + assert response.status_code == 200 + body = response.json() + assert [(error["code"], error["step"]) for error in body["errors"]] == [ + ("integration_unavailable", "prerequisites") + ] + assert sentinel not in response.text + + async def test_cloud_preflight_accepts_usable_integration_and_repository( + self, async_client + ): + """A usable credential, MCP server, repository, and ref pass together.""" + mcp_url = "https://mcp.example.test/github" + calls: list[str] = [] + + def outbound(request: httpx.Request) -> httpx.Response: + calls.append(request.url.path) + if request.url.path == "/api/v1/secrets/search": + return httpx.Response( + 200, + json={ + "items": [{"name": "GITHUB_TOKEN", "description": None}], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/settings": + return httpx.Response( + 200, + json={ + "agent_settings": { + "mcp_config": { + "github": { + "transport": "http", + "url": mcp_url, + "auth": { + "strategy": "api_key", + "value": "**********", + }, + } + } + } + }, + ) + if request.url.path == "/api/v1/settings/mcp/github/test": + return httpx.Response(200, json={"ok": True}) + if request.url.path == "/api/v1/git/repositories/search": + assert request.url.params["provider"] == "github" + assert request.url.params["query"] == "OpenHands/agent-server-gui" + return httpx.Response( + 200, + json={ + "items": [ + { + "id": "1", + "full_name": "OpenHands/agent-server-gui", + "git_provider": "github", + "is_public": True, + } + ], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/git/branches/search": + assert request.url.params["repository"] == ( + "OpenHands/agent-server-gui" + ) + assert request.url.params["query"] == "main" + return httpx.Response( + 200, + json={ + "items": [ + { + "name": "main", + "commit_sha": "0123456789abcdef", + "protected": True, + } + ], + "next_page_id": None, + }, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight( + CRON_DRAFT, + requirements={ + "integrations": [ + integration_requirement( + "github", + transport="shttp", + locator=mcp_url, + auth_strategy="api_key", + secret_names=["GITHUB_TOKEN"], + ) + ] + }, + ), + ) + + assert response.status_code == 200 + assert response.json()["valid"] is True + assert response.json()["errors"] == [] + assert "/api/v1/settings/mcp/github/test" in calls + assert "/api/v1/git/branches/search" in calls + + async def test_cloud_repository_and_ref_search_follow_pagination( + self, async_client + ): + """An exact repository or ref on a later page is still accessible.""" + calls: list[tuple[str, str | None]] = [] + + def outbound(request: httpx.Request) -> httpx.Response: + page_id = request.url.params.get("page_id") + calls.append((request.url.path, page_id)) + if request.url.path == "/api/v1/git/repositories/search": + assert request.url.params["limit"] == "99" + if page_id is None: + return httpx.Response( + 200, + json={ + "items": [], + "next_page_id": "repository-page-2", + }, + ) + assert page_id == "repository-page-2" + return httpx.Response( + 200, + json={ + "items": [ + { + "id": "1", + "full_name": "OpenHands/agent-server-gui", + "git_provider": "github", + "is_public": False, + } + ], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/git/branches/search": + if page_id is None: + return httpx.Response( + 200, + json={"items": [], "next_page_id": "branch-page-2"}, + ) + assert page_id == "branch-page-2" + return httpx.Response( + 200, + json={ + "items": [ + { + "name": "main", + "commit_sha": "0123456789abcdef", + "protected": True, + } + ], + "next_page_id": None, + }, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight(CRON_DRAFT, requirements={"integrations": []}), + ) + + assert response.status_code == 200 + assert response.json()["valid"] is True + assert calls == [ + ("/api/v1/git/repositories/search", None), + ("/api/v1/git/repositories/search", "repository-page-2"), + ("/api/v1/git/branches/search", None), + ("/api/v1/git/branches/search", "branch-page-2"), + ] + + async def test_repeated_repository_page_token_fails_closed(self, async_client): + """A broken dependency cannot make preflight loop or claim success.""" + calls = 0 + + def outbound(request: httpx.Request) -> httpx.Response: + nonlocal calls + assert request.url.path == "/api/v1/git/repositories/search" + calls += 1 + return httpx.Response( + 200, + json={"items": [], "next_page_id": "repeated-page"}, + ) + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight(CRON_DRAFT, requirements={"integrations": []}), + ) + + assert calls == 2 + assert response.status_code == 503 + assert response.json() == { + "detail": "Preflight validation is temporarily unavailable." + } + + @pytest.mark.parametrize( + "url", + [ + "https://attacker.example/OpenHands/agent-server-gui", + "git@attacker.example:OpenHands/agent-server-gui", + ], + ) + async def test_repository_provider_must_match_the_selected_host( + self, async_client, url + ): + """A declared provider cannot validate a different host's same path.""" + + def outbound(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/git/repositories/search": + return httpx.Response( + 200, + json={ + "items": [ + { + "full_name": "OpenHands/agent-server-gui", + "git_provider": "github", + } + ], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/git/branches/search": + return httpx.Response( + 200, + json={ + "items": [{"name": "main", "commit_sha": "abc123"}], + "next_page_id": None, + }, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + draft = { + **CRON_DRAFT, + "repos": [{"url": url, "provider": "github"}], + } + response = await async_client.post( + VALIDATE_URL, + json=preflight(draft, requirements={"integrations": []}), + ) + + assert response.status_code == 200 + assert addressed_errors(response.json()) == [ + ("repos[0].url", "repository_provider_unsupported") + ] + + @pytest.mark.parametrize( + ("repository_items", "branch_items", "expected_error"), + [ + pytest.param( + [], + None, + ("repos[0].url", "repository_not_accessible"), + id="repository", + ), + pytest.param( + [ + { + "id": "1", + "full_name": "OpenHands/agent-server-gui", + "git_provider": "github", + "is_public": False, + } + ], + [], + ("repos[0].ref", "repository_ref_not_accessible"), + id="ref", + ), + ], + ) + async def test_cloud_repository_failures_are_field_addressable( + self, + async_client, + repository_items, + branch_items, + expected_error, + ): + """Repository and ref failures identify the exact form field to fix.""" + + def outbound(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/git/repositories/search": + return httpx.Response( + 200, + json={"items": repository_items, "next_page_id": None}, + ) + if request.url.path == "/api/v1/git/branches/search": + assert branch_items is not None + return httpx.Response( + 200, + json={"items": branch_items, "next_page_id": None}, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight(CRON_DRAFT, requirements={"integrations": []}), + ) + + assert response.status_code == 200 + assert addressed_errors(response.json()) == [expected_error] + + async def test_local_preflight_uses_names_and_encrypted_mcp_configuration( + self, async_client, monkeypatch + ): + """Local secrets stay in agent-server; automation forwards encrypted config.""" + monkeypatch.setenv("AUTOMATION_AGENT_SERVER_URL", "http://agent-server.test") + monkeypatch.setenv("AUTOMATION_AGENT_SERVER_API_KEY", "session-key") + clear_config_cache() + mcp_url = "https://mcp.example.test/github" + encrypted = "enc:v1:not-a-plaintext-secret" + + def outbound(request: httpx.Request) -> httpx.Response: + assert request.headers["x-session-api-key"] == "session-key" + if request.url.path == "/api/settings/secrets": + return httpx.Response( + 200, + json={ + "secrets": [ + {"name": "GITHUB_TOKEN", "description": None}, + {"name": "github_token", "description": None}, + ] + }, + ) + if request.url.path == "/api/settings": + assert request.headers["x-expose-secrets"] == "encrypted" + return httpx.Response( + 200, + json={ + "agent_settings": { + "mcp_config": { + "github": { + "transport": "http", + "url": mcp_url, + "auth": { + "strategy": "api_key", + "value": encrypted, + }, + } + } + } + }, + ) + if request.url.path == "/api/mcp/test": + payload = json.loads(request.content) + assert payload["server"]["auth"]["value"] == encrypted + return httpx.Response(200, json={"ok": True, "tools": []}) + if request.url.path == "/api/git/validate-repository": + payload = json.loads(request.content) + assert payload["credential_names"] == [ + "GITHUB_TOKEN", + "github_token", + ] + return httpx.Response(200, json={"status": "accessible"}) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight( + CRON_DRAFT, + requirements={ + "integrations": [ + integration_requirement( + "github", + transport="shttp", + locator=mcp_url, + auth_strategy="api_key", + secret_names=["GITHUB_TOKEN"], + ) + ] + }, + ), + ) + + assert response.status_code == 200 + assert response.json()["valid"] is True + assert encrypted not in response.text + + @pytest.mark.parametrize("failure_kind", ["status", "transport"]) + async def test_dependency_failures_return_sanitized_503( + self, async_client, failure_kind + ): + """A dependency outage blocks creation and never surfaces its body.""" + sentinel = "provider-internal-stack-and-secret" + + def outbound(request: httpx.Request) -> httpx.Response: + if failure_kind == "transport": + raise httpx.ConnectError(sentinel, request=request) + return httpx.Response(500, text=sentinel) + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight(CRON_DRAFT, requirements={"integrations": []}), + ) + + assert response.status_code == 503 + assert response.json() == { + "detail": "Preflight validation is temporarily unavailable." + } + assert sentinel not in response.text + + async def test_preflight_total_timeout_fails_closed( + self, async_client, monkeypatch + ): + """A slow dependency cannot retain an unbounded validation request.""" + monkeypatch.setattr( + capabilities_router, + "_PREFLIGHT_TOTAL_TIMEOUT_SECONDS", + 0.01, + raising=False, + ) + + async def outbound(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(0.05) + if request.url.path == "/api/v1/git/repositories/search": + return httpx.Response( + 200, + json={ + "items": [{"full_name": "OpenHands/agent-server-gui"}], + "next_page_id": None, + }, + ) + if request.url.path == "/api/v1/git/branches/search": + return httpx.Response( + 200, + json={ + "items": [{"name": "main", "commit_sha": "abc123"}], + "next_page_id": None, + }, + ) + raise AssertionError(f"Unexpected outbound request: {request.url}") + + await install_outbound_transport(outbound) + response = await async_client.post( + VALIDATE_URL, + json=preflight(CRON_DRAFT, requirements={"integrations": []}), + ) + + assert response.status_code == 503 + assert response.json() == { + "detail": "Preflight validation is temporarily unavailable." + } + + async def test_preflight_total_timeout_covers_event_source_lookup( + self, async_client, monkeypatch + ): + """The request deadline also covers event checks before requirements.""" + monkeypatch.setattr( + capabilities_router, + "_PREFLIGHT_TOTAL_TIMEOUT_SECONDS", + 0.01, + raising=False, + ) + + async def slow_webhook_lookup(*_args, **_kwargs): + await asyncio.sleep(0.05) + return None + + monkeypatch.setattr( + capabilities_router, + "get_webhook_config", + slow_webhook_lookup, + ) + + response = await async_client.post( + VALIDATE_URL, + json=preflight(EVENT_DRAFT), + ) + + assert response.status_code == 503 + assert response.json() == { + "detail": "Preflight validation is temporarily unavailable." + } + + async def test_malformed_requirements_are_rejected_before_validation( + self, async_client + ): + """The public requirements envelope is bounded and rejects secret values.""" + response = await async_client.post( + VALIDATE_URL, + json=preflight( + CRON_DRAFT, + requirements={ + "integrations": [ + { + "id": "github", + "alternatives": [ + { + "transport": "shttp", + "locator": "https://mcp.example.test/github", + "authStrategy": "api_key", + "secretNames": ["GITHUB_TOKEN"], + "secretValue": "must-not-be-accepted", + } + ], + } + ] + }, + ), + ) + + assert response.status_code == 422