Coverage for src/local_deep_research/security/url_validator.py: 94%

176 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-22 21:20 +0000

1""" 

2Centralized URL validation utilities for security. 

3 

4This module provides secure URL validation to prevent XSS attacks, 

5data exfiltration, and other URL-based security vulnerabilities. 

6""" 

7 

8import re 

9from typing import Optional, List 

10from urllib.parse import urlparse, urljoin, unquote 

11from loguru import logger 

12 

13from .log_sanitizer import redact_and_bound_for_log 

14 

15 

16class URLValidationError(ValueError): 

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

18 

19 pass 

20 

21 

22class URLValidator: 

23 """Centralized URL validation for security.""" 

24 

25 # Unsafe URL schemes that could lead to XSS or data exfiltration 

26 UNSAFE_SCHEMES = ( 

27 "javascript", 

28 "data", 

29 "vbscript", 

30 "about", 

31 "blob", 

32 "file", 

33 ) 

34 

35 # Safe schemes for external links 

36 SAFE_SCHEMES = ("http", "https", "ftp", "ftps") 

37 

38 # Email scheme 

39 EMAIL_SCHEME = "mailto" 

40 

41 # Common academic/research domains that should be allowed 

42 TRUSTED_ACADEMIC_DOMAINS = ( 

43 "arxiv.org", 

44 "pubmed.ncbi.nlm.nih.gov", 

45 "ncbi.nlm.nih.gov", 

46 "biorxiv.org", 

47 "medrxiv.org", 

48 "doi.org", 

49 "nature.com", 

50 "science.org", 

51 "sciencedirect.com", 

52 "springer.com", 

53 "wiley.com", 

54 "plos.org", 

55 "pnas.org", 

56 "ieee.org", 

57 "acm.org", 

58 ) 

59 

60 @staticmethod 

61 def is_unsafe_scheme(url: str) -> bool: 

62 """ 

63 Check if a URL uses an unsafe scheme. 

64 

65 Args: 

66 url: The URL to check 

67 

68 Returns: 

69 True if the URL uses an unsafe scheme, False otherwise 

70 """ 

71 if not url: 

72 return False 

73 

74 # Normalize the URL - trim whitespace and convert to lowercase 

75 normalized_url = url.strip().lower() 

76 

77 # Check for unsafe schemes 

78 for scheme in URLValidator.UNSAFE_SCHEMES: 

79 if normalized_url.startswith(f"{scheme}:"): 

80 logger.warning( 

81 "Unsafe URL scheme detected: {} in URL: {}", 

82 scheme, 

83 # (#6938: redacted before it is cut, so the cut cannot 

84 # drop the ``@`` after URL userinfo.) 

85 redact_and_bound_for_log(url, 100), 

86 ) 

87 return True 

88 

89 return False 

90 

91 @staticmethod 

92 def is_safe_url( 

93 url: str, 

94 require_scheme: bool = True, 

95 allow_fragments: bool = True, 

96 allow_mailto: bool = False, 

97 trusted_domains: Optional[List[str]] = None, 

98 ) -> bool: 

99 """ 

100 Validate if a URL is safe to use. 

101 

102 Args: 

103 url: The URL to validate 

104 require_scheme: Whether to require an explicit scheme 

105 allow_fragments: Whether to allow fragment identifiers (#) 

106 allow_mailto: Whether to allow mailto: links 

107 trusted_domains: Optional list of trusted domains 

108 

109 Returns: 

110 True if the URL is safe, False otherwise 

111 """ 

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

113 return False 

114 

115 # Check for unsafe schemes first 

116 if URLValidator.is_unsafe_scheme(url): 

117 return False 

118 

119 # Handle fragment-only URLs 

120 if url.startswith("#"): 

121 return allow_fragments 

122 

123 # Parse the URL 

124 try: 

125 parsed = urlparse(url) 

126 except Exception: 

127 logger.warning( 

128 "Failed to parse URL '{}'", 

129 redact_and_bound_for_log(str(url), 100), 

130 ) 

131 return False 

132 

133 # Check scheme 

134 if not parsed.scheme: 

135 if require_scheme: 

136 return False 

137 # If no scheme is required, assume http/https for URL parsing 

138 parsed = urlparse(f"http://{url}") # DevSkim: ignore DS137138 

139 

140 scheme_lower = parsed.scheme.lower() 

141 

142 # Check if it's a mailto link 

143 if scheme_lower == URLValidator.EMAIL_SCHEME: 

144 return allow_mailto 

145 

146 # Check if it's a safe scheme 

147 if scheme_lower not in URLValidator.SAFE_SCHEMES: 

148 logger.warning(f"Unsafe URL scheme: {scheme_lower}") 

149 return False 

150 

151 # Validate domain if trusted domains are specified 

152 if trusted_domains and parsed.hostname: 

153 hostname_lower = parsed.hostname.lower() 

154 if not any( 

155 hostname_lower == domain.lower() 

156 or hostname_lower.endswith(f".{domain.lower()}") 

157 for domain in trusted_domains 

158 ): 

159 logger.warning( 

160 f"URL domain not in trusted list: {parsed.hostname}" 

161 ) 

162 return False 

163 

164 # Check for suspicious patterns in the URL 

165 if URLValidator._has_suspicious_patterns(url): 

166 return False 

167 

168 return True 

169 

170 @staticmethod 

171 def _has_suspicious_patterns(url: str) -> bool: 

172 """ 

173 Check for suspicious patterns in URLs that might indicate attacks. 

174 

175 Args: 

176 url: The URL to check 

177 

178 Returns: 

179 True if suspicious patterns are found, False otherwise 

180 """ 

181 suspicious_patterns = [ 

182 # Double encoding 

183 r"%25[0-9a-fA-F]{2}", 

184 # Null bytes 

185 r"%00", 

186 # Unicode encoding bypass attempts 

187 r"\\u[0-9a-fA-F]{4}", 

188 # HTML entity encoding 

189 r"&(#x?[0-9a-fA-F]+|[a-zA-Z]+);", 

190 ] 

191 

192 for pattern in suspicious_patterns: 

193 if re.search(pattern, url, re.IGNORECASE): 

194 logger.warning(f"Suspicious pattern found in URL: {pattern}") 

195 return True 

196 

197 return False 

198 

199 @staticmethod 

200 def sanitize_url(url: str, default_scheme: str = "https") -> Optional[str]: 

201 """ 

202 Sanitize a URL by adding a scheme if missing and validating it. 

203 

204 Args: 

205 url: The URL to sanitize 

206 default_scheme: The default scheme to add if missing 

207 

208 Returns: 

209 Sanitized URL or None if the URL is unsafe 

210 """ 

211 if not url: 

212 return None 

213 

214 # Check for unsafe schemes 

215 if URLValidator.is_unsafe_scheme(url): 

216 return None 

217 

218 # Strip whitespace 

219 url = url.strip() 

220 

221 # Parse the URL 

222 try: 

223 parsed = urlparse(url) 

224 

225 # Add scheme if missing 

226 if not parsed.scheme: 

227 url = f"{default_scheme}://{url}" 

228 parsed = urlparse(url) 

229 

230 # Validate the final URL 

231 if URLValidator.is_safe_url(url, require_scheme=True): 

232 return url 

233 

234 except Exception: 

235 logger.warning( 

236 "Failed to sanitize URL '{}'", 

237 redact_and_bound_for_log(str(url), 100), 

238 ) 

239 

240 return None 

241 

242 @staticmethod 

243 def is_academic_url(url: str) -> bool: 

244 """ 

245 Check if a URL is from a known academic/research domain. 

246 

247 Args: 

248 url: The URL to check 

249 

250 Returns: 

251 True if the URL is from an academic domain, False otherwise 

252 """ 

253 try: 

254 parsed = urlparse(url) 

255 if parsed.hostname: 

256 hostname_lower = parsed.hostname.lower() 

257 return any( 

258 hostname_lower == domain 

259 or hostname_lower.endswith(f".{domain}") 

260 for domain in URLValidator.TRUSTED_ACADEMIC_DOMAINS 

261 ) 

262 except Exception: 

263 logger.debug( 

264 "URL parsing may fail on malformed input; treat as non-academic", 

265 exc_info=True, 

266 ) 

267 

268 return False 

269 

270 @staticmethod 

271 def extract_doi(url: str) -> Optional[str]: 

272 """ 

273 Extract DOI from a URL if present. 

274 

275 Args: 

276 url: The URL to extract DOI from 

277 

278 Returns: 

279 The DOI if found, None otherwise 

280 """ 

281 # Common DOI patterns with explicit pattern identification 

282 doi_patterns = [ 

283 ( 

284 r"10\.\d{4,}(?:\.\d+)*\/[-._;()\/:a-zA-Z0-9]+", 

285 0, 

286 ), # Direct DOI, group 0 

287 (r"doi\.org\/(10\.\d{4,}[^\s]*)", 1), # doi.org URL, group 1 

288 ] 

289 

290 for pattern, group_index in doi_patterns: 

291 match = re.search(pattern, url, re.IGNORECASE) 

292 if match: 

293 return match.group(group_index) 

294 

295 return None 

296 

297 @staticmethod 

298 def is_safe_http_link(url: object) -> bool: 

299 """Allow absolute HTTP(S) links without interpreting path/query data. 

300 

301 Literal percent signs, entity-like query values and trailing-dot hosts 

302 are valid in browser links. Callback-specific suspicious-pattern checks 

303 would reject them. Callers must still HTML-escape the URL in a quoted 

304 attribute; this policy does not authorize server-side requests. 

305 """ 

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

307 return False 

308 

309 # Browsers and URL parsers can strip raw control characters. Reject 

310 # those ambiguous inputs while preserving percent-encoded URL data. 

311 if any(ord(char) < 0x20 or ord(char) == 0x7F for char in url): 

312 return False 

313 

314 try: 

315 parsed = urlparse(url) 

316 # Accessing port also validates its numeric form and range. 

317 port = parsed.port 

318 return ( 

319 parsed.scheme in ("http", "https") 

320 and bool(parsed.hostname) 

321 and (port is None or 0 <= port <= 65535) 

322 ) 

323 except ValueError: 

324 return False 

325 

326 @staticmethod 

327 def validate_http_url(url: str) -> bool: 

328 """ 

329 Validate that a callback URL is well-formed and safe for HTTP/HTTPS use. 

330 

331 This is stricter than is_safe_url() and specifically validates HTTP/HTTPS 

332 URLs for use as application callbacks (e.g., in notifications, redirects). 

333 It does NOT validate Apprise service URLs which use other protocols. 

334 

335 Args: 

336 url: HTTP/HTTPS callback URL to validate 

337 

338 Returns: 

339 True if valid 

340 

341 Raises: 

342 URLValidationError: If URL is invalid 

343 """ 

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

345 raise URLValidationError("URL must be a non-empty string") 

346 

347 try: 

348 parsed = urlparse(url) 

349 except URLValidationError: 

350 raise 

351 except Exception as e: 

352 raise URLValidationError(f"Failed to validate URL: {e}") from e 

353 

354 # Must have a scheme 

355 if not parsed.scheme: 

356 raise URLValidationError("URL must have a scheme (http or https)") 

357 

358 # Must be http or https (callback URLs only) 

359 if parsed.scheme not in ("http", "https"): 

360 raise URLValidationError( 

361 f"URL scheme must be http or https, got: {parsed.scheme}" 

362 ) 

363 

364 # Use the general security validator for additional safety 

365 if not URLValidator.is_safe_url(url, require_scheme=True): 

366 raise URLValidationError(f"URL failed security validation: {url}") 

367 

368 # Must have a netloc (hostname) 

369 if not parsed.netloc: 

370 raise URLValidationError("URL must have a hostname") 

371 

372 # Check for obvious hostname issues 

373 if parsed.netloc.startswith(".") or parsed.netloc.endswith("."): 

374 raise URLValidationError(f"Invalid hostname: {parsed.netloc}") 

375 

376 # Path should be valid if present 

377 if parsed.path and not parsed.path.startswith("/"): 377 ↛ 378line 377 didn't jump to line 378 because the condition on line 377 was never true

378 raise URLValidationError( 

379 f"URL path must start with /: {parsed.path}" 

380 ) 

381 

382 return True 

383 

384 @staticmethod 

385 def is_safe_redirect_url(target: str, host_url: str) -> bool: 

386 """ 

387 Validate that a redirect target is safe (same host). 

388 

389 Prevents open redirect attacks by ensuring the target URL 

390 is either relative or points to the same host as the application. 

391 Also prevents CRLF injection attacks by rejecting URLs containing 

392 carriage return or line feed characters. 

393 

394 Uses the standard Flask pattern from: 

395 https://github.com/fengsp/flask-snippets/blob/master/security/redirect_back.py 

396 

397 Security protections implemented: 

398 - CRLF injection prevention 

399 - Protocol-relative URL bypass (//evil.com) 

400 - Triple-slash bypass (///evil.com) 

401 - URL-encoded bypass attempts 

402 - Backslash bypass (treated as forward slash by some browsers) 

403 - Path traversal detection 

404 

405 Args: 

406 target: The redirect URL to validate (can be relative or absolute) 

407 host_url: The application's host URL (e.g., request.host_url) 

408 

409 Returns: 

410 True if the URL is safe to redirect to, False otherwise 

411 """ 

412 if not target: 

413 return False 

414 

415 # Prevent CRLF injection by rejecting URLs with CR or LF characters 

416 if "\r" in target or "\n" in target: 

417 logger.warning("CRLF injection attempt detected in redirect URL") 

418 return False 

419 

420 # Normalize and decode URL to detect encoded bypass attempts 

421 # This catches %2f%2f (encoded //) and similar tricks 

422 decoded_target = unquote(target) 

423 

424 # Check for encoded CRLF injection (e.g. %0d%0a) 

425 if "\r" in decoded_target or "\n" in decoded_target: 

426 logger.warning( 

427 "Encoded CRLF injection attempt detected in redirect URL" 

428 ) 

429 return False 

430 

431 # Check for backslash (treated as forward slash by some browsers) 

432 # Prevents \evil.com and \\evil.com bypasses 

433 if "\\" in decoded_target: 

434 logger.warning("Backslash detected in redirect URL") 

435 return False 

436 

437 # Check for protocol-relative URLs before stripping slashes 

438 # //evil.com would be treated as absolute URL by urljoin() 

439 if decoded_target.startswith("//"): 

440 logger.warning("Protocol-relative URL detected in redirect") 

441 return False 

442 

443 # Block null bytes (truncation attacks in some URL processors) 

444 if "\x00" in decoded_target: 

445 logger.warning("Null byte detected in redirect URL") 

446 return False 

447 

448 # Block path traversal patterns (/../, /.. at end, ../ at start) 

449 if re.search(r"(^|/)\.\.(/|$|\?|#)", decoded_target): 

450 logger.warning( 

451 "Path traversal detected in redirect URL: {}", 

452 # (#6938: the ``next`` target is a raw query param.) 

453 redact_and_bound_for_log(target, 200), 

454 ) 

455 return False 

456 

457 ref_url = urlparse(host_url) 

458 test_url = urlparse(urljoin(host_url, target)) 

459 

460 if test_url.path.startswith("//"): 

461 logger.warning( 

462 "Double-slash path in redirect URL: {}", 

463 redact_and_bound_for_log(target, 200), 

464 ) 

465 return False 

466 

467 return ( 

468 test_url.scheme in ("http", "https") 

469 and ref_url.netloc == test_url.netloc 

470 ) 

471 

472 @staticmethod 

473 def get_safe_redirect_path(target: str, host_url: str) -> Optional[str]: 

474 """ 

475 Validate a redirect target and return its path-only form. 

476 

477 Combines is_safe_redirect_url validation with defense-in-depth path 

478 extraction. The target is resolved against host_url via urljoin, then 

479 only the path, query, and fragment are returned. This ensures that 

480 even a hypothetical validator bypass cannot cause an external redirect, 

481 since path-only URLs always resolve to the same host. 

482 

483 Args: 

484 target: The redirect URL to validate (can be relative or absolute) 

485 host_url: The application's host URL (e.g., request.host_url) 

486 

487 Returns: 

488 The safe path string (e.g., "/dashboard?tab=settings") if the 

489 target is valid, or None if the target is unsafe. 

490 """ 

491 if not URLValidator.is_safe_redirect_url(target, host_url): 

492 return None 

493 

494 resolved = urljoin(host_url, target) 

495 parsed = urlparse(resolved) 

496 safe_path = parsed.path or "/" 

497 if parsed.query: 

498 safe_path += "?" + parsed.query 

499 if parsed.fragment: 

500 safe_path += "#" + parsed.fragment 

501 

502 # Final assertion: the constructed path must be relative (no netloc). 

503 # This is always true by construction but guards against future bugs. 

504 if urlparse(safe_path).netloc: 504 ↛ 505line 504 didn't jump to line 505 because the condition on line 504 was never true

505 return None 

506 return safe_path 

507 

508 

509def get_javascript_url_validator() -> str: 

510 """ 

511 Get JavaScript code for URL validation that matches the Python implementation. 

512 

513 Returns: 

514 JavaScript code as a string that can be embedded in web pages 

515 """ 

516 return r""" 

517 // URL validation utilities matching Python URLValidator 

518 const URLValidator = { 

519 UNSAFE_SCHEMES: ['javascript', 'data', 'vbscript', 'about', 'blob', 'file'], 

520 SAFE_SCHEMES: ['http', 'https', 'ftp', 'ftps'], 

521 EMAIL_SCHEME: 'mailto', 

522 

523 isUnsafeScheme: function(url) { 

524 if (!url) return false; 

525 

526 const normalizedUrl = url.trim().toLowerCase(); 

527 

528 for (const scheme of this.UNSAFE_SCHEMES) { 

529 if (normalizedUrl.startsWith(scheme + ':')) { 

530 console.warn(`Unsafe URL scheme detected: ${scheme}`); 

531 return true; 

532 } 

533 } 

534 

535 return false; 

536 }, 

537 

538 isSafeUrl: function(url, options = {}) { 

539 const { 

540 requireScheme = true, 

541 allowFragments = true, 

542 allowMailto = false, 

543 trustedDomains = [] 

544 } = options; 

545 

546 if (!url || typeof url !== 'string') { 

547 return false; 

548 } 

549 

550 // Check for unsafe schemes first 

551 if (this.isUnsafeScheme(url)) { 

552 return false; 

553 } 

554 

555 // Handle fragment-only URLs 

556 if (url.startsWith('#')) { 

557 return allowFragments; 

558 } 

559 

560 // Parse the URL 

561 try { 

562 const parsed = new URL(url, window.location.href); 

563 const scheme = parsed.protocol.slice(0, -1).toLowerCase(); // Remove trailing ':' 

564 

565 // Check if it's a mailto link 

566 if (scheme === this.EMAIL_SCHEME) { 

567 return allowMailto; 

568 } 

569 

570 // Check if it's a safe scheme 

571 if (!this.SAFE_SCHEMES.includes(scheme)) { 

572 console.warn(`Unsafe URL scheme: ${scheme}`); 

573 return false; 

574 } 

575 

576 // Validate domain if trusted domains are specified 

577 if (trustedDomains.length > 0 && parsed.hostname) { 

578 const hostname = parsed.hostname.toLowerCase(); 

579 const isTrusted = trustedDomains.some(domain => 

580 hostname === domain.toLowerCase() || 

581 hostname.endsWith('.' + domain.toLowerCase()) 

582 ); 

583 

584 if (!isTrusted) { 

585 console.warn(`URL domain not in trusted list: ${parsed.hostname}`); 

586 return false; 

587 } 

588 } 

589 

590 return true; 

591 } catch (e) { 

592 console.warn(`Failed to parse URL: ${e.message}`); 

593 return false; 

594 } 

595 }, 

596 

597 sanitizeUrl: function(url, defaultScheme = 'https') { 

598 if (!url) return null; 

599 

600 // Check for unsafe schemes 

601 if (this.isUnsafeScheme(url)) { 

602 return null; 

603 } 

604 

605 // Strip whitespace 

606 url = url.trim(); 

607 

608 // Add scheme if missing 

609 if (!url.match(/^[a-zA-Z][a-zA-Z\d+\-.]*:/)) { 

610 url = `${defaultScheme}://${url}`; 

611 } 

612 

613 // Validate the final URL 

614 if (this.isSafeUrl(url, { requireScheme: true })) { 

615 return url; 

616 } 

617 

618 return null; 

619 } 

620 }; 

621 """