Skip to content

[Task]: Prepare URL validation infrastructure for agent card and webhook SSRF protection #1023

@ishymko

Description

@ishymko

Context

We have two places which can benefit from SSRF protection:

  1. Client: fetching agent cards and using URLs to make A2A calls (brought in fix(security): SSRF via AgentCard URL and context ID Injection (A2A-SSRF-01, A2A-INJ-01) #895, filed a separate issue [Feat]: Agent card validation hook #975).
  2. Push notification webhooks URLs (Security Improvement: Add SSRF protection for Push Notification webhooks and authorization checks for Task operations #786): there is a spec statement about SSRF protection there.

Description

Both cases above can share domain agnostic machinery for URL validation.

The following requirements apply:

  1. Composable: local addresses are perfectly fine for some deployments.
  2. Validation should happen before the actual invocation and invocation should use "pinned" IP to protect from DNS rebinding.
  3. Built-in rules: RequireScheme (to restrict HTTP(S)-only or HTTPS-only), BlockPrivateNetworks (with allowlist for deployments where using private networks is a legitimate use-case).

This issue scopes to UrlValidator only and should be a foundation for #975 and #786 where appropriate domain validators are going to be implemented.

flowchart TB
    subgraph Wrappers["Domain wrappers (own which fields to validate + domain error type)"]
        direction LR
        PN["PushNotificationUrlValidator"]
        AC["AgentCardUrlValidator"]
    end
    V["UrlValidator<br/>parse → resolve → run rules in order<br/>(rule raises = reject, returns = continue)"]
    subgraph Rules["Composable rules"]
        direction LR
        RS["RequireScheme"]
        BPN["BlockPrivateNetworks<br/>(allow_hosts, allow_cidrs)"]
        CR["…custom…"]
    end
    PN --> V
    AC --> V
    V --> Rules
    class PN,AC wrap
    class V core
    class RS,BPN,HD,CR rule
Loading

Sketch:

class InvalidUrlError(ValueError):
    """Raised by UrlValidator.validate when a URL is rejected."""

class UrlValidationRule(ABC):
    @abstractmethod
    async def check(self, url: ResolvedUrl) -> None:
        """Raise InvalidUrlError to reject url."""

class ResolvedUrl:
    raw: str
    parsed: SplitResult
    addresses: tuple[IpAddress, ...]


class UrlValidator:
    def __init__(
        self,
        rules: Sequence[UrlValidationRule] = (),
        resolve: bool = True,
    ) -> None:    
    . . .

    async def validate(self, url: str) -> ResolvedUrl:
        # Conditionally resolves IPs via getaddrinfo
        resolved = await self._build(url)

        # Just runs all rules one by one, they throw to stop
        for rule in self._rules:
            await rule.check(resolved)

        # Returns URL so that pinned address can be used
        return resolved

Metadata

Metadata

Assignees

No one assigned

    Labels

    component: coreIssues related to base data models, auth, gRPC interfaces, observability, and fundamental utilities.

    Type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions