Coverage for src/lilbee/crawler/url_filter.py: 100%

1"""URL validation, blocked-network checks, and host-scope helpers.""" 

2 

3from __future__ import annotations 

4 

5import ipaddress 

6import socket 

7from urllib.parse import urlparse 

8 

9_BLOCKED_NETWORKS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = ( 

10 ipaddress.ip_network("0.0.0.0/8"), # "this host" range; 0.0.0.0 routes to localhost 

11 ipaddress.ip_network("127.0.0.0/8"), 

12 ipaddress.ip_network("10.0.0.0/8"), 

13 ipaddress.ip_network("172.16.0.0/12"), 

14 ipaddress.ip_network("192.168.0.0/16"), 

15 ipaddress.ip_network("169.254.0.0/16"), 

16 ipaddress.ip_network("100.64.0.0/10"), # RFC 6598 shared / CGNAT 

17 ipaddress.ip_network("::/128"), # IPv6 unspecified 

18 ipaddress.ip_network("::1/128"), 

19 ipaddress.ip_network("fe80::/10"), # IPv6 link-local 

20 ipaddress.ip_network("fc00::/7"), # IPv6 unique-local (ULA) 

21 ipaddress.ip_network("ff00::/8"), # IPv6 multicast 

22 ipaddress.ip_network("64:ff9b::/96"), # NAT64 well-known prefix 

23) 

24 

25_NAT64_PREFIX = ipaddress.ip_network("64:ff9b::/96") 

26_IPV4_TRANSLATED = ipaddress.ip_network("::ffff:0:0:0/96") # RFC 6052 SIIT 

27 

28 

29def get_blocked_networks() -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]: 

30 """Return blocked network list. Override in tests via monkeypatch.""" 

31 return _BLOCKED_NETWORKS 

32 

33 

34def _embedded_ipv4(ip: ipaddress.IPv6Address) -> ipaddress.IPv4Address | None: 

35 """Return the IPv4 an IPv6 address embeds, if any. 

36 

37 Covers IPv4-mapped (``::ffff:a.b.c.d``), 6to4 (``2002::``), the NAT64 

38 well-known prefix, the IPv4-translated/SIIT prefix (``::ffff:0:a.b.c.d``), 

39 and the deprecated IPv4-compatible (``::a.b.c.d``) form. Each can reach the 

40 same host as its bare IPv4, so the embedded address must face the blocklist. 

41 """ 

42 if ip.ipv4_mapped is not None: 

43 return ip.ipv4_mapped 

44 if ip.sixtofour is not None: 

45 return ip.sixtofour 

46 low32 = int(ip) & 0xFFFFFFFF 

47 if ip in _NAT64_PREFIX or ip in _IPV4_TRANSLATED: 

48 return ipaddress.IPv4Address(low32) 

49 # IPv4-compatible ::a.b.c.d: top 96 bits zero, excluding :: and ::1. 

50 if int(ip) >> 32 == 0 and low32 > 1: 

51 return ipaddress.IPv4Address(low32) 

52 return None 

53 

54 

55def is_url(value: str) -> bool: 

56 """Check if a string is an HTTP/HTTPS URL.""" 

57 return value.startswith(("http://", "https://")) 

58 

59 

60def validate_crawl_url(url: str) -> None: 

61 """Validate a URL for crawling. Raises ValueError for unsafe URLs. 

62 Rejects private IPs, loopback, link-local, and non-HTTP schemes. 

63 """ 

64 parsed = urlparse(url) 

65 scheme = parsed.scheme.lower() 

66 if scheme not in ("http", "https"): 

67 raise ValueError(f"Only http:// and https:// URLs are allowed, got {scheme}://") 

68 

69 hostname = parsed.hostname 

70 if not hostname: 

71 raise ValueError("URL has no hostname") 

72 

73 try: 

74 addr_infos = socket.getaddrinfo(hostname, None) 

75 except socket.gaierror as exc: 

76 raise ValueError(f"Cannot resolve hostname: {hostname}") from exc 

77 

78 networks = get_blocked_networks() 

79 for _family, _type, _proto, _canonname, sockaddr in addr_infos: 

80 ip = ipaddress.ip_address(sockaddr[0]) 

81 # An IPv6 address can embed an IPv4 (mapped, 6to4, NAT64, compatible) 

82 # that reaches the same host but would slip past the IPv4 checks, so 

83 # test both the address and any embedded IPv4 against the blocklist. 

84 candidates: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [ip] 

85 if isinstance(ip, ipaddress.IPv6Address): 

86 embedded = _embedded_ipv4(ip) 

87 if embedded is not None: 

88 candidates.append(embedded) 

89 for candidate in candidates: 

90 for network in networks: 

91 if candidate in network: 

92 raise ValueError(f"Crawling private/reserved IP {candidate} is not allowed") 

93 

94 

95def require_valid_crawl_url(url: str) -> None: 

96 """Validate URL for crawling. Raises ValueError if invalid.""" 

97 if not is_url(url): 

98 raise ValueError("URL must start with http:// or https://") 

99 validate_crawl_url(url) 

100 

101 

102def host_in_scope(link_host: str, host: str, *, include_subdomains: bool) -> bool: 

103 """Return True when ``link_host`` should be followed during a whole-site crawl.""" 

104 if not link_host: 

105 return False 

106 if link_host == host: 

107 return True 

108 return include_subdomains and link_host.endswith(f".{host}")