Coverage for src / lexigram / contracts / security / url_safety.py: 52%

1"""SSRF URL-safety primitive shared by every outbound HTTP consumer. 

2 

3Pure standard-library only (``ipaddress``, ``socket``, ``urllib.parse``) so 

4that any extension package — including ``lexigram-webhook``, ``lexigram-ai-rag`` 

5and ``lexigram-ai-mcp``, which may not import ``lexigram.security`` — can 

6consume it directly. Keep this file stdlib-only (import-linter Contract 1). 

7""" 

8 

9from __future__ import annotations 

10 

11from collections.abc import Callable, Sequence 

12import ipaddress 

13import socket 

14import urllib.parse 

15 

16_PRIVATE_IPV4_NETWORKS: tuple[ipaddress.IPv4Network, ...] = ( 

17 ipaddress.IPv4Network("0.0.0.0/8"), 

18 ipaddress.IPv4Network("10.0.0.0/8"), 

19 ipaddress.IPv4Network("100.64.0.0/10"), 

20 ipaddress.IPv4Network("127.0.0.0/8"), 

21 ipaddress.IPv4Network("169.254.0.0/16"), 

22 ipaddress.IPv4Network("172.16.0.0/12"), 

23 ipaddress.IPv4Network("192.168.0.0/16"), 

24 ipaddress.IPv4Network("224.0.0.0/4"), 

25 ipaddress.IPv4Network("240.0.0.0/4"), 

26) 

27 

28_PRIVATE_IPV6_NETWORKS: tuple[ipaddress.IPv6Network, ...] = ( 

29 ipaddress.IPv6Network("::/128"), 

30 ipaddress.IPv6Network("::1/128"), 

31 ipaddress.IPv6Network("fc00::/7"), 

32 ipaddress.IPv6Network("fe80::/10"), 

33) 

34 

35HostResolver = Callable[[str], Sequence[ipaddress.IPv4Address | ipaddress.IPv6Address]] 

36 

37 

38def resolve_hostname( 

39 hostname: str, 

40) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: 

41 """Resolve a hostname to all its address records via the system resolver. 

42 

43 Args: 

44 hostname: DNS name to resolve. 

45 

46 Returns: 

47 List of resolved addresses (A/AAAA records). 

48 

49 Raises: 

50 OSError: If the hostname cannot be resolved. 

51 """ 

52 infos = socket.getaddrinfo( 

53 hostname, None, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP 

54 ) 

55 addresses: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [] 

56 for _family, _stype, _proto, _canonname, sockaddr in infos: 

57 try: 

58 addresses.append(ipaddress.ip_address(sockaddr[0])) 

59 except ValueError: 

60 continue 

61 return addresses 

62 

63 

64def is_safe_url_for_request( 

65 url: str, 

66 *, 

67 resolver: HostResolver | None = None, 

68) -> bool: 

69 """Return False when requesting ``url`` could reach a private/reserved host. 

70 

71 Checks the scheme (http/https only), rejects literal private/reserved IP 

72 hostnames, and — for DNS hostnames — resolves via ``resolver`` (defaults to 

73 the system ``getaddrinfo``) and rejects when ANY resolved address is 

74 private/reserved. Fails closed: an unresolvable or empty resolution is 

75 rejected. 

76 

77 Args: 

78 url: Candidate absolute URL. 

79 resolver: Optional hostname resolver for hermetic tests. Defaults to 

80 the system resolver via :func:`resolve_hostname`. 

81 

82 Returns: 

83 True only when a request to ``url`` cannot reach a private/reserved IP. 

84 """ 

85 try: 

86 parsed = urllib.parse.urlparse(url) 

87 except ValueError: 

88 return False 

89 

90 if parsed.scheme.lower() not in {"http", "https"}: 

91 return False 

92 

93 hostname = parsed.hostname 

94 if not hostname: 

95 return False 

96 

97 hostname = hostname.strip("[]") 

98 try: 

99 addr = ipaddress.ip_address(hostname) 

100 except ValueError: 

101 return _hostname_is_public(hostname, resolver) 

102 

103 return not _is_private(addr) 

104 

105 

106def _hostname_is_public(hostname: str, resolver: HostResolver | None) -> bool: 

107 try: 

108 addresses = ( 

109 resolve_hostname(hostname) if resolver is None else resolver(hostname) 

110 ) 

111 except OSError: 

112 return False 

113 if not addresses: 

114 return False 

115 return all(not _is_private(address) for address in addresses) 

116 

117 

118def _is_private( 

119 addr: ipaddress.IPv4Address | ipaddress.IPv6Address, 

120) -> bool: 

121 if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None: 

122 addr = addr.ipv4_mapped 

123 if isinstance(addr, ipaddress.IPv4Address): 

124 return any(addr in network for network in _PRIVATE_IPV4_NETWORKS) 

125 return any(addr in network for network in _PRIVATE_IPV6_NETWORKS) 

126 

127 

128__all__ = ["HostResolver", "is_safe_url_for_request", "resolve_hostname"]