a2aproject / a2aproject/a2a-python

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

Abierto
#1,023 2 comentarios 0 reacciones 0 asignados Ver en GitHub
component: core
Lenguaje dominante
Python
Estrellas
2.1k
Forks
496
Merge medio
4 d 17 h
PR fusionados (30 d)
12

Descripción

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

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.