Coverage for chalicelib/utils/ssrf.py: 68%

1import ipaddress 

2import socket 

3from urllib.parse import urlparse, urlunparse 

4 

5import requests 

6from decouple import config 

7from requests.adapters import HTTPAdapter 

8 

9 

10def get_allowed_hosts(): 

11 return {h.strip().lower() for h in config("WEBHOOK_ALLOWED_HOSTS", default="").split(",") if h.strip()} 

12 

13 

14def resolve_public_ip(endpoint): 

15 """Validates that the endpoint resolves to public IPs only and returns one of them, 

16 so the request can be pinned to it (protects against DNS-rebinding between check and request). 

17 Returns None (no validation, no pinning) for hosts listed in WEBHOOK_ALLOWED_HOSTS.""" 

18 parsed = urlparse(endpoint) 

19 hostname = parsed.hostname 

20 if not hostname: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 raise ValueError(f"endpoint has no valid hostname: {endpoint}") 

22 if hostname.lower() in get_allowed_hosts(): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 return None 

24 try: 

25 resolved = socket.getaddrinfo(hostname, parsed.port or (443 if parsed.scheme == "https" else 80), 

26 proto=socket.IPPROTO_TCP) 

27 except socket.gaierror as e: 

28 raise ValueError(f"endpoint hostname could not be resolved: {hostname}") from e 

29 ips = [] 

30 for family, _, _, _, sockaddr in resolved: 

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

32 if not ip.is_global: 

33 raise ValueError(f"endpoint resolves to a non-public IP address: {hostname} -> {ip}") 

34 ips.append(ip) 

35 return str(ips[0]) 

36 

37 

38class PinnedHostHTTPSAdapter(HTTPAdapter): 

39 """Keeps TLS SNI and certificate validation bound to the original hostname 

40 while the connection itself targets a pre-resolved IP.""" 

41 

42 def __init__(self, hostname, **kwargs): 

43 self._hostname = hostname 

44 super().__init__(**kwargs) 

45 

46 def init_poolmanager(self, *args, **kwargs): 

47 kwargs["server_hostname"] = self._hostname 

48 kwargs["assert_hostname"] = self._hostname 

49 super().init_poolmanager(*args, **kwargs) 

50 

51 

52def post_json(endpoint, json_data, headers=None, timeout=None): 

53 """POSTs JSON to an endpoint after validating that it targets a public IP (SSRF guard). 

54 The connection is pinned to the validated IP and redirects are not followed. 

55 Raises ValueError if the endpoint is non-public and not in WEBHOOK_ALLOWED_HOSTS.""" 

56 endpoint = str(endpoint) 

57 headers = headers or {} 

58 pinned_ip = resolve_public_ip(endpoint) 

59 if pinned_ip is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 return requests.post(url=endpoint, json=json_data, headers=headers, timeout=timeout, 

61 allow_redirects=False) 

62 parsed = urlparse(endpoint) 

63 hostname = parsed.hostname 

64 ip_host = f"[{pinned_ip}]" if ":" in pinned_ip else pinned_ip 

65 netloc = ip_host if parsed.port is None else f"{ip_host}:{parsed.port}" 

66 pinned_url = urlunparse(parsed._replace(netloc=netloc)) 

67 headers = {**headers, "Host": hostname if parsed.port is None else f"{hostname}:{parsed.port}"} 

68 with requests.Session() as s: 

69 if parsed.scheme == "https": 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true

70 s.mount("https://", PinnedHostHTTPSAdapter(hostname)) 

71 return s.post(url=pinned_url, json=json_data, headers=headers, timeout=timeout, 

72 allow_redirects=False) 

73 

74 

75def get_integration_allowed_hosts(): 

76 """Hosts exempted from the public-IP requirement for user-configured 

77 integrations (e.g. an Elasticsearch instance on a private network).""" 

78 return {h.strip().lower() for h in config("INTEGRATION_ALLOWED_HOSTS", default="").split(",") if h.strip()} 

79 

80 

81def is_public_host(host: str, port=None) -> bool: 

82 """Returns True only when the bare hostname/IP resolves exclusively to 

83 public IPs (blocks loopback, RFC1918, link-local incl. cloud metadata 

84 endpoints, CGNAT, ULA, reserved and multicast ranges).""" 

85 if not host: 

86 return False 

87 try: 

88 infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP) 

89 except (socket.gaierror, UnicodeError): 

90 return False 

91 if len(infos) == 0: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 return False 

93 for info in infos: 93 ↛ 100line 93 didn't jump to line 100 because the loop on line 93 didn't complete

94 try: 

95 ip = ipaddress.ip_address(info[4][0]) 

96 except ValueError: 

97 return False 

98 if not ip.is_global or ip.is_multicast: 98 ↛ 93line 98 didn't jump to line 93 because the condition on line 98 was always true

99 return False 

100 return True 

101 

102 

103ALLOWED_SCHEMES = ("http", "https") 

104ALLOWED_PORTS = (80, 443) 

105 

106 

107def is_safe_external_url(url: str) -> bool: 

108 """SSRF guard for URLs fetched server-side on behalf of user-controlled data 

109 (e.g. sourcemap URLs coming from error stack frames): http(s) only, no 

110 embedded credentials, standard ports, and every IP the hostname resolves to 

111 must be public (blocks loopback, RFC1918, link-local incl. cloud metadata 

112 endpoints, CGNAT, ULA, reserved and multicast ranges). 

113 

114 Note: DNS is resolved here for validation only; the actual fetch resolves 

115 again, so consumers must re-validate at connect time to rule out DNS 

116 rebinding (sourcemapreader does this via its lookup hook).""" 

117 try: 

118 p = urlparse(url) 

119 except ValueError: 

120 return False 

121 if p.scheme not in ALLOWED_SCHEMES: 

122 return False 

123 if not p.hostname or p.username or p.password: 

124 return False 

125 try: 

126 port = p.port 

127 except ValueError: 

128 return False 

129 port = port or (443 if p.scheme == "https" else 80) 

130 if port not in ALLOWED_PORTS: 

131 return False 

132 return is_public_host(p.hostname, port)