a2aproject / a2aproject/a2a-python

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

Đang mở
#1,023 2 bình luận 0 reaction 0 người được giao Xem trên GitHub
component: core
Ngôn ngữ chính
Python
Star
2.1k
Fork
496
Merge trung bình
4 ngày 17 giờ
Pull request đã merge (30 ngày)
12

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.