a2aproject / a2aproject/a2a-python

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

オープン
#1,023 コメント 2 件 リアクション 0 件 担当者 0 名 GitHub で見る
component: core
主要言語
Python
スター
2.1k
フォーク
496
平均マージ
4日 17時間
マージ済み PR(30日)
12

説明

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

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。