a2aproject / a2aproject/a2a-python

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

Open
#1,023 2 comments 0 reactions 0 assignees View on GitHub
component: core
Dominant language
Python
Stars
2.1k
Forks
496
Avg merge
4d 17h
Merged PRs (30d)
12

Description

### 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 #895, filed a separate issue #975).
2. Push notification webhooks URLs (#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.

```mermaid
flowchart TB
subgraph Wrappers["Domain wrappers (own which fields to validate + domain error type)"]
direction LR
PN["PushNotificationUrlValidator"]
AC["AgentCardUrlValidator"]
end
V["UrlValidator
parse → resolve → run rules in order
(rule raises = reject, returns = continue)"]
subgraph Rules["Composable rules"]
direction LR
RS["RequireScheme"]
BPN["BlockPrivateNetworks
(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
```

Sketch:

```python
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
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.