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 re 

7import socket 

8from typing import TYPE_CHECKING 

9from urllib.parse import ParseResult, urlparse 

10 

11# ada-url is in the optional ``crawler`` extra, so each function that needs it imports it on call. 

12if TYPE_CHECKING: 

13 import ada_url 

14 

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

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

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

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

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

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

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

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

23 ipaddress.ip_network("240.0.0.0/4"), # RFC 1112 reserved; holds broadcast 255.255.255.255 

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

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

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

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

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

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

30) 

31 

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

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

34# The mark that ends a username and password in the host part of an address. 

35_USERINFO_END = "@" 

36# A WHATWG parser escapes a raw "^" in a path; the url crate 2.5.8 crawlberg 1.12.3 uses keeps it. 

37_CARET = "^" 

38_CARET_ESCAPE = "%5E" 

39_CARET_ESCAPE_LOWER = "%5e" 

40_ANY_CARET_ESCAPE = re.compile(_CARET_ESCAPE, re.IGNORECASE) 

41# What is shown for an address with an ``@`` that does not parse: where a password ends is unknown. 

42UNPARSED_ADDRESS = "(an address that does not parse)" 

43# The refusal of a port that is not a number; the port text can hold part of a password. 

44INVALID_PORT_MESSAGE = "URL has an invalid port" 

45 

46 

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

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

49 return _BLOCKED_NETWORKS 

50 

51 

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

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

54 

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

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

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

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

59 """ 

60 if ip.ipv4_mapped is not None: 

61 return ip.ipv4_mapped 

62 if ip.sixtofour is not None: 

63 return ip.sixtofour 

64 low32 = int(ip) & 0xFFFFFFFF 

65 if ip in _NAT64_PREFIX or ip in _IPV4_TRANSLATED: 

66 return ipaddress.IPv4Address(low32) 

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

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

69 return ipaddress.IPv4Address(low32) 

70 return None 

71 

72 

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

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

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

76 

77 

78def parsed_url(url: str) -> ada_url.URL | None: 

79 """*url* parsed by the WHATWG URL standard, or None when it does not parse.""" 

80 import ada_url 

81 

82 try: 

83 return ada_url.URL(url) 

84 except ValueError: 

85 return None 

86 

87 

88def parsed_address(url: str) -> str | None: 

89 """*url* without its fragment as crawlberg 1.12.3's URL parser writes it; None if it fails.""" 

90 parsed = parsed_url(_ANY_CARET_ESCAPE.sub(_CARET_ESCAPE_LOWER, url)) 

91 if parsed is None: 

92 return None 

93 parsed.hash = "" 

94 return parsed.href.replace(_CARET_ESCAPE, _CARET) 

95 

96 

97def without_userinfo(url: str) -> str: 

98 """*url* without the username and password of its host part, as the URL standard reads it. 

99 

100 An address with an ``@`` that does not parse is shown as ``UNPARSED_ADDRESS``. 

101 """ 

102 if _USERINFO_END not in url: 

103 return url 

104 parsed = parsed_url(url) 

105 if parsed is None: 

106 return UNPARSED_ADDRESS 

107 if not parsed.username and not parsed.password: 

108 return url 

109 parsed.username = "" 

110 parsed.password = "" 

111 return parsed.href 

112 

113 

114def _checked_port(parsed: ParseResult) -> int | None: 

115 """The port of *parsed*; a ValueError naming no part of the address when it is no number.""" 

116 try: 

117 return parsed.port 

118 except ValueError as exc: 

119 raise ValueError(INVALID_PORT_MESSAGE) from exc 

120 

121 

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

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

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

125 """ 

126 parsed = urlparse(url) 

127 scheme = parsed.scheme.lower() 

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

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

130 

131 hostname = parsed.hostname 

132 if not hostname: 

133 raise ValueError("URL has no hostname") 

134 _checked_port(parsed) 

135 

136 try: 

137 addr_infos = socket.getaddrinfo(hostname, None) 

138 except socket.gaierror as exc: 

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

140 

141 networks = get_blocked_networks() 

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

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

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

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

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

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

148 if isinstance(ip, ipaddress.IPv6Address): 

149 embedded = _embedded_ipv4(ip) 

150 if embedded is not None: 

151 candidates.append(embedded) 

152 for candidate in candidates: 

153 for network in networks: 

154 if candidate in network: 

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

156 

157 

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

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

160 if not is_url(url): 

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

162 validate_crawl_url(url) 

163 

164 

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

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

167 if not link_host: 

168 return False 

169 if link_host == host: 

170 return True 

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