a2aproject / a2aproject/a2a-python
[Task]: Prepare URL validation infrastructure for agent card and webhook SSRF protection
- 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
Đánh giá
Issue này chưa được đánh giá.