Coverage for src/local_deep_research/security/url_builder.py: 96%

1""" 

2URL building utilities for security and application use. 

3 

4Provides centralized URL construction logic that can be reused 

5throughout the application for consistent URL handling. 

6""" 

7 

8import re 

9from typing import Optional, Union 

10from urllib.parse import urlparse 

11from loguru import logger 

12 

13from .ssrf_validator import redact_url_for_log 

14 

15 

16class URLBuilderError(Exception): 

17 """Raised when URL construction fails.""" 

18 

19 pass 

20 

21 

22def normalize_bind_address(host: str) -> str: 

23 """ 

24 Convert bind addresses to URL-friendly hostnames. 

25 

26 Args: 

27 host: Host address from settings (may include bind addresses) 

28 

29 Returns: 

30 URL-friendly hostname 

31 """ 

32 # Convert bind-all addresses to localhost for URLs 

33 if host in ("0.0.0.0", "::"): 

34 return "localhost" 

35 return host 

36 

37 

38def build_base_url_from_settings( 

39 external_url: Optional[str] = None, 

40 host: Optional[str] = None, 

41 port: Optional[Union[str, int]] = None, 

42 fallback_base: str = "http://localhost:5000", 

43) -> str: 

44 """ 

45 Build a base URL from application settings with intelligent fallbacks. 

46 

47 This function handles the common pattern of building application URLs 

48 from various configuration sources with proper normalization. 

49 

50 Args: 

51 external_url: Pre-configured external URL (highest priority) 

52 host: Hostname/IP address (used if external_url not provided) 

53 port: Port number (used with host if external_url not provided) 

54 fallback_base: Final fallback URL if nothing else is available 

55 

56 Returns: 

57 Complete base URL (e.g., "https://myapp.com" or "http://localhost:5000") 

58 

59 Raises: 

60 URLBuilderError: If URL construction fails 

61 """ 

62 try: 

63 # Try external URL first (highest priority) 

64 if external_url and external_url.strip(): 

65 base_url = external_url.strip().rstrip("/") 

66 logger.debug( 

67 f"Using configured external URL: {redact_url_for_log(base_url)}" 

68 ) 

69 return base_url 

70 

71 # Try to construct from host and port 

72 if host and port: 

73 normalized_host = normalize_bind_address(host) 

74 

75 # Use HTTP for host/port combinations (typically internal server addresses) 

76 # For external URLs, users should configure external_url setting instead 

77 base_url = f"http://{normalized_host}:{int(port)}" # DevSkim: ignore DS137138 

78 logger.debug( 

79 f"Constructed URL from host/port: {redact_url_for_log(base_url)}" 

80 ) 

81 return base_url 

82 

83 # Final fallback 

84 base_url = fallback_base.rstrip("/") 

85 logger.debug(f"Using fallback URL: {redact_url_for_log(base_url)}") 

86 return base_url 

87 

88 except Exception as e: 

89 raise URLBuilderError(f"Failed to build base URL: {e}") 

90 

91 

92def build_full_url( 

93 base_url: str, 

94 path: str, 

95 validate: bool = True, 

96 allowed_schemes: Optional[list] = None, 

97) -> str: 

98 """ 

99 Build a complete URL from base URL and path. 

100 

101 Args: 

102 base_url: Base URL (e.g., "https://myapp.com") 

103 path: Path to append (e.g., "/research/123") 

104 validate: Whether to validate the resulting URL 

105 allowed_schemes: List of allowed URL schemes (default: ["http", "https"]) 

106 

107 Returns: 

108 Complete URL (e.g., "https://myapp.com/research/123") 

109 

110 Raises: 

111 URLBuilderError: If URL construction or validation fails 

112 """ 

113 try: 

114 # Ensure path starts with / 

115 if not path.startswith("/"): 

116 path = f"/{path}" 

117 

118 # Ensure base URL doesn't end with / 

119 base_url = base_url.rstrip("/") 

120 

121 # Construct full URL 

122 full_url = f"{base_url}{path}" 

123 

124 if validate: 

125 validate_constructed_url(full_url, allowed_schemes) 

126 

127 return full_url 

128 

129 except Exception as e: 

130 raise URLBuilderError(f"Failed to build full URL: {e}") 

131 

132 

133def validate_constructed_url( 

134 url: str, allowed_schemes: Optional[list] = None 

135) -> bool: 

136 """ 

137 Validate a constructed URL. 

138 

139 Args: 

140 url: URL to validate 

141 allowed_schemes: List of allowed schemes (default: ["http", "https"]) 

142 

143 Returns: 

144 True if valid 

145 

146 Raises: 

147 URLBuilderError: If URL is invalid 

148 """ 

149 if not url or not isinstance(url, str): 

150 raise URLBuilderError("URL must be a non-empty string") 

151 

152 try: 

153 parsed = urlparse(url) 

154 except Exception as e: 

155 raise URLBuilderError(f"Failed to parse URL: {e}") 

156 

157 # Check scheme 

158 if not parsed.scheme: 

159 raise URLBuilderError("URL must have a scheme") 

160 

161 if allowed_schemes and parsed.scheme not in allowed_schemes: 

162 raise URLBuilderError( 

163 f"URL scheme '{parsed.scheme}' not in allowed schemes: {allowed_schemes}" 

164 ) 

165 

166 # Check hostname 

167 if not parsed.netloc: 

168 raise URLBuilderError("URL must have a hostname") 

169 

170 return True 

171 

172 

173def mask_sensitive_url(url: str) -> str: 

174 """ 

175 Mask sensitive parts of a URL for secure logging. 

176 

177 This function masks passwords, webhook tokens, and other sensitive 

178 information in URLs to prevent accidental exposure in logs. 

179 

180 Args: 

181 url: URL to mask 

182 

183 Returns: 

184 URL with sensitive parts replaced with *** 

185 """ 

186 try: 

187 parsed = urlparse(url) 

188 

189 # Mask password if present 

190 if parsed.password: 

191 netloc = parsed.netloc.replace(parsed.password, "***") 

192 else: 

193 netloc = parsed.netloc 

194 

195 # Mask path tokens (common in webhooks) 

196 path = parsed.path 

197 if path: 

198 # Replace long alphanumeric tokens with *** 

199 path = re.sub( 

200 r"/[a-zA-Z0-9_-]{20,}", 

201 "/***", 

202 path, 

203 ) 

204 

205 # Reconstruct URL 

206 masked = f"{parsed.scheme}://{netloc}{path}" 

207 if parsed.query: 

208 masked += "?***" 

209 

210 return masked 

211 

212 except Exception: 

213 # If parsing fails, just return generic mask 

214 return f"{url.split(':')[0]}://***"