Coverage for src/local_deep_research/security/egress/fetch.py: 88%

1"""Egress-aware URL fetch validation. 

2 

3A thin wrapper over the general SSRF validator that threads the egress 

4*scope* through, so this layer (egress) depends on the SSRF validator rather 

5than the SSRF validator depending on egress. 

6""" 

7 

8from loguru import logger 

9 

10from ..ssrf_validator import validate_url 

11from .policy import EgressScope 

12 

13 

14def policy_aware_validate_url(url: str, egress_context=None) -> bool: 

15 """Validate ``url`` for SSRF, taking the egress policy scope into account. 

16 

17 Under ``EgressScope.PRIVATE_ONLY`` the policy lets the user reach 

18 private hosts (local lab deployments — Ollama on 127.0.0.1, SearXNG 

19 on 192.168.x). Plain ``validate_url(url)`` would reject those at 

20 the SSRF layer even though policy explicitly permits them. This 

21 wrapper threads the scope through so a PRIVATE_ONLY run can reach 

22 private hosts WITHOUT requiring the operator to set 

23 ``SSRF_ALLOW_PRIVATE_IPS=1`` globally. 

24 

25 Cloud-metadata IPs in ``ALWAYS_BLOCKED_METADATA_IPS`` remain blocked 

26 regardless of scope (handled inside ``is_ip_blocked``). 

27 

28 When ``egress_context`` is ``None`` the call is equivalent to 

29 ``validate_url(url)`` (strict defaults). 

30 """ 

31 if egress_context is None: 

32 return validate_url(url) 

33 try: 

34 if egress_context.scope == EgressScope.PRIVATE_ONLY: 

35 return validate_url(url, allow_private_ips=True) 

36 except Exception: # noqa: silent-exception 

37 # Defensive: any attribute error here means the context is 

38 # malformed; treat as if no policy was supplied and fall back to 

39 # strict defaults. Logging is intentionally suppressed to avoid 

40 # leaking policy state via timing. 

41 logger.debug("policy_aware_validate_url defensive fallback") 

42 return validate_url(url)