Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 4 additions & 21 deletions src/previewshield/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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]")

Expand Down Expand Up @@ -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, ...]:
Expand Down
34 changes: 8 additions & 26 deletions src/previewshield/policy.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,24 +2,22 @@

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

import yaml

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",
Expand Down Expand Up @@ -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, ...]:
Expand All @@ -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.")
Expand Down
50 changes: 7 additions & 43 deletions src/previewshield/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__
Expand All @@ -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"
Expand All @@ -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
Expand All @@ -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(
Expand Down Expand Up @@ -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}"
Expand All @@ -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
Expand Down
72 changes: 72 additions & 0 deletions src/previewshield/targets.py
Original file line number Diff line number Diff line change
@@ -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)
Loading