diff --git a/src/previewshield/network.py b/src/previewshield/network.py index 4241688..aca3403 100644 --- a/src/previewshield/network.py +++ b/src/previewshield/network.py @@ -27,12 +27,12 @@ from previewshield.exceptions import NetworkSafetyError, ScanError from previewshield.models import MutableHeaderBag, RedirectHop, ResponseSnapshot, TLSInfo +from previewshield.targets import normalize_hostname as _shared_normalize_hostname _DEFAULT_TIMEOUT = 10.0 _DEFAULT_REDIRECTS = 5 _MAX_REDIRECTS = 20 _MAX_URL_LENGTH = 8192 -_MAX_HOSTNAME_LENGTH = 253 _MAX_PORT = 65_535 _MAX_HEADER_VALUE_LENGTH = 16_384 _MAX_RESPONSE_HEADER_BYTES = 64 * 1024 @@ -52,7 +52,6 @@ ) _HEADER_NAME = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$") _INVALID_PERCENT_ESCAPE = re.compile(r"%(?![0-9A-Fa-f]{2})") -_HOST_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") _CONTROL_CHARACTER = re.compile(r"[\x00-\x08\x0a-\x1f\x7f]") _UNSAFE_URL_CHARACTER = re.compile(r"[\x00-\x1f\x7f]") @@ -401,26 +400,10 @@ def _parse_url(url: str, *, allow_shorthand: bool) -> SplitResult: def _normalize_hostname(hostname: str) -> str: - if "%" in hostname: - raise NetworkSafetyError("IPv6 scope identifiers are not allowed in target URLs.") - try: - return ipaddress.ip_address(hostname).compressed - except ValueError: - pass - - hostname = hostname.rstrip(".") - if not hostname: - raise NetworkSafetyError("Target URL must include a hostname.") - try: - ascii_hostname = hostname.encode("idna").decode("ascii").lower() - except UnicodeError as error: - raise NetworkSafetyError("Target URL contains an invalid hostname.") from error - if len(ascii_hostname) > _MAX_HOSTNAME_LENGTH: - raise NetworkSafetyError("Target hostname exceeds 253 characters.") - if any(not _HOST_LABEL.fullmatch(label) for label in ascii_hostname.split(".")): - raise NetworkSafetyError("Target URL contains an invalid hostname.") - return ascii_hostname + return _shared_normalize_hostname(hostname) + except ValueError as error: + raise NetworkSafetyError(str(error)) from error def _normalize_allowed_hosts(patterns: tuple[str, ...]) -> tuple[str, ...]: diff --git a/src/previewshield/policy.py b/src/previewshield/policy.py index 97c5760..046ae1e 100644 --- a/src/previewshield/policy.py +++ b/src/previewshield/policy.py @@ -2,11 +2,10 @@ from __future__ import annotations -import ipaddress import re from collections.abc import Mapping from dataclasses import dataclass -from pathlib import Path, PurePosixPath +from pathlib import Path from typing import Any from urllib.parse import urlsplit @@ -14,12 +13,11 @@ from previewshield.exceptions import ConfigurationError from previewshield.models import Severity +from previewshield.targets import normalize_hostname, normalize_origin_paths from previewshield.utils import clean_text, parse_threshold, validate_header_name MAX_POLICY_BYTES = 256 * 1024 _RULE_ID = re.compile(r"^(?:PS\d{4}|CUSTOM\.[A-Z0-9_.-]+)$") -_HOST_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") -_MAX_HOSTNAME_LENGTH = 253 _ALLOWED_ROOT_KEYS = { "version", "name", @@ -351,19 +349,10 @@ def _paths(value: Any) -> tuple[str, ...]: paths = _string_list(value, "paths") if not paths: raise ConfigurationError("paths must include at least one route.") - normalized: list[str] = [] - for path in paths: - parsed = urlsplit(path) - if parsed.scheme or parsed.netloc or parsed.fragment: - raise ConfigurationError(f"Path must be origin-relative without a fragment: {path!r}.") - route = parsed.path or "/" - if not route.startswith("/") or ".." in PurePosixPath(route).parts: - raise ConfigurationError(f"Unsafe route path: {path!r}.") - if parsed.query: - route = f"{route}?{parsed.query}" - if route not in normalized: - normalized.append(route) - return tuple(normalized) + try: + return normalize_origin_paths(paths) + except ValueError as error: + raise ConfigurationError(f"Unsafe route path ({error})") from error def _rule_ids(value: Any, field: str) -> tuple[str, ...]: @@ -390,16 +379,9 @@ def _normalize_host(value: str) -> str: if "*" in raw_hostname: raise ConfigurationError(f"Invalid allowed host: {value!r}.") try: - normalized = raw_hostname.encode("idna").decode("ascii") - except UnicodeError as error: + normalized = normalize_hostname(raw_hostname) + except ValueError as error: raise ConfigurationError(f"Invalid allowed host: {value!r}.") from error - if len(normalized) > _MAX_HOSTNAME_LENGTH: - raise ConfigurationError(f"Invalid allowed host: {value!r}.") - try: - ipaddress.ip_address(normalized) - except ValueError: - if any(not _HOST_LABEL.fullmatch(label) for label in normalized.split(".")): - raise ConfigurationError(f"Invalid allowed host: {value!r}.") from None else: if wildcard: raise ConfigurationError("Wildcard allowed hosts cannot target IP addresses.") diff --git a/src/previewshield/scanner.py b/src/previewshield/scanner.py index 519d385..f975197 100644 --- a/src/previewshield/scanner.py +++ b/src/previewshield/scanner.py @@ -2,10 +2,7 @@ from __future__ import annotations -import ipaddress -import re from collections.abc import Mapping, Sequence -from pathlib import PurePosixPath from urllib.parse import urlsplit, urlunsplit from previewshield._version import __version__ @@ -14,6 +11,7 @@ from previewshield.models import RouteReport, ScanReport, Severity from previewshield.network import NetworkOptions, fetch from previewshield.policy import Policy, default_policy +from previewshield.targets import normalize_hostname, normalize_origin_paths from previewshield.utils import utc_now SCHEMA_VERSION = "1.0" @@ -25,8 +23,6 @@ Severity.CRITICAL: 25, } _GRADES = ((95, "A+"), (90, "A"), (80, "B"), (70, "C"), (60, "D")) -_HOST_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") -_MAX_HOSTNAME_LENGTH = 253 def scan( # noqa: PLR0913 - public API uses explicit, optional controls @@ -47,7 +43,10 @@ def scan( # noqa: PLR0913 - public API uses explicit, optional controls active_policy = policy or default_policy() threshold = fail_on or active_policy.fail_on - route_paths = _validate_paths(paths) if paths is not None else active_policy.paths + try: + route_paths = normalize_origin_paths(paths) if paths is not None else active_policy.paths + except ValueError as error: + raise ConfigurationError(str(error)) from error route_urls, normalized_target = _route_urls(target, route_paths, paths is None) try: options = NetworkOptions( @@ -117,10 +116,9 @@ def _route_urls( raise ConfigurationError("Target URL fragments are not allowed.") try: - hostname = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower() - except UnicodeError as error: + hostname = normalize_hostname(parsed.hostname or "") + except ValueError as error: raise ConfigurationError("Target contains an invalid hostname.") from error - _validate_hostname(hostname) display_host = f"[{hostname}]" if ":" in hostname else hostname default_port = 443 if parsed.scheme.lower() == "https" else 80 netloc = display_host if port in {None, default_port} else f"{display_host}:{port}" @@ -132,40 +130,6 @@ def _route_urls( return tuple(f"{origin}{path}" for path in paths), origin -def _validate_hostname(hostname: str) -> None: - if len(hostname) > _MAX_HOSTNAME_LENGTH: - raise ConfigurationError("Target hostname exceeds 253 characters.") - try: - ipaddress.ip_address(hostname) - except ValueError: - if not hostname or any(not _HOST_LABEL.fullmatch(label) for label in hostname.split(".")): - raise ConfigurationError("Target contains an invalid hostname.") from None - - -def _validate_paths(paths: Sequence[str]) -> tuple[str, ...]: - if isinstance(paths, (str, bytes)) or not paths: - raise ConfigurationError("At least one origin-relative path is required.") - result: list[str] = [] - for raw_path in paths: - if not isinstance(raw_path, str) or not raw_path.strip(): - raise ConfigurationError("Scan paths must be non-empty strings.") - parsed = urlsplit(raw_path.strip()) - route = parsed.path or "/" - if ( - parsed.scheme - or parsed.netloc - or parsed.fragment - or not route.startswith("/") - or ".." in PurePosixPath(route).parts - ): - raise ConfigurationError(f"Unsafe origin-relative path: {raw_path!r}.") - if parsed.query: - route = f"{route}?{parsed.query}" - if route not in result: - result.append(route) - return tuple(result) - - def _score(routes: Sequence[RouteReport]) -> int: if not routes: return 0 diff --git a/src/previewshield/targets.py b/src/previewshield/targets.py new file mode 100644 index 0000000..c41395d --- /dev/null +++ b/src/previewshield/targets.py @@ -0,0 +1,72 @@ +"""Shared normalization and validation for scan targets and routes. + +Scanner, network, and policy layers historically repeated hostname and +origin-relative path validation with slightly different messages. These helpers +keep one canonical implementation while letting each caller translate failures +into its own domain error type. +""" + +from __future__ import annotations + +import ipaddress +import re +from collections.abc import Sequence +from pathlib import PurePosixPath +from urllib.parse import urlsplit + +MAX_HOSTNAME_LENGTH = 253 + +_HOST_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") + + +def normalize_hostname(hostname: str) -> str: + """Return a lowercase ASCII hostname, raising ``ValueError`` when unsafe.""" + + if "%" in hostname: + raise ValueError("IPv6 scope identifiers are not allowed in target URLs.") + try: + return ipaddress.ip_address(hostname).compressed + except ValueError: + pass + + stripped = hostname.rstrip(".") + if not stripped: + raise ValueError("Target URL must include a hostname.") + try: + normalized = stripped.encode("idna").decode("ascii").lower() + except UnicodeError as error: + raise ValueError("Target URL contains an invalid hostname.") from error + if len(normalized) > MAX_HOSTNAME_LENGTH: + raise ValueError(f"Target hostname exceeds {MAX_HOSTNAME_LENGTH} characters.") + if any(not _HOST_LABEL.fullmatch(label) for label in normalized.split(".")): + raise ValueError("Target URL contains an invalid hostname.") + return normalized + + +def normalize_origin_paths(paths: Sequence[str]) -> tuple[str, ...]: + """Validate origin-relative paths and return deduplicated normalized routes. + + Raises ``ValueError`` when a path is not a safe origin-relative route. + """ + + if isinstance(paths, (str, bytes)) or not paths: + raise ValueError("At least one origin-relative path is required.") + result: list[str] = [] + for raw_path in paths: + if not isinstance(raw_path, str) or not raw_path.strip(): + raise ValueError("Scan paths must be non-empty strings.") + parsed = urlsplit(raw_path.strip()) + route = parsed.path or "/" + if ( + parsed.scheme + or parsed.netloc + or parsed.fragment + or not route.startswith("/") + or ".." in PurePosixPath(route).parts + ): + raise ValueError(f"Unsafe origin-relative path: {raw_path!r}.") + if parsed.query: + route = f"{route}?{parsed.query}" + if route not in result: + result.append(route) + return tuple(result)