pixano_inference.utils.media_security

Security policy for media ingestion (SSRF guard, size/time limits, path containment).

Client requests reference images and video by URL, local path, or base64. Dereferencing those references is a classic SSRF and arbitrary-file-read surface. Every fetch and every local-path resolution goes through :class:MediaPolicy here, which:

• accepts only http/https URLs (never file:///s3://);
• resolves the URL host and rejects private/loopback/link-local/reserved IPs (with an optional host allowlist for trusted internal services);
• re-validates every redirect hop;
• enforces connect/read timeouts and a streamed maximum-byte cap;
• denies local-path media unless the path resolves under a configured media_roots entry.

The active policy is a per-process global built lazily from :class:ServerSettings, so a Ray worker reconstructs the same policy from the environment its driver passed on.

MediaPolicy(allow_url=True, url_host_allowlist=frozenset(), allow_private_ips=False, media_roots=(), connect_timeout_s=5.0, read_timeout_s=30.0, max_redirects=3, max_image_bytes=50 * 1024 * 1024, max_video_bytes=512 * 1024 * 1024) dataclass

Resolved media-ingestion security policy for the current process.

from_env() classmethod

Build a policy from the environment (via :class:ServerSettings).

Source code in pixano_inference/utils/media_security.py

@classmethod
def from_env(cls) -> MediaPolicy:
    """Build a policy from the environment (via :class:`ServerSettings`)."""
    from pixano_inference.server_settings import ServerSettings

    return cls.from_settings(ServerSettings())

from_settings(settings) classmethod

Build a policy from :class:ServerSettings.

Source code in pixano_inference/utils/media_security.py

@classmethod
def from_settings(cls, settings: ServerSettings) -> MediaPolicy:
    """Build a policy from :class:`ServerSettings`."""
    return cls(
        allow_url=settings.media_allow_url,
        url_host_allowlist=frozenset(h.lower() for h in settings.media_url_host_allowlist),
        allow_private_ips=settings.media_allow_private_ips,
        media_roots=tuple(Path(p).expanduser().resolve() for p in settings.media_roots),
        connect_timeout_s=settings.media_connect_timeout_s,
        read_timeout_s=settings.media_read_timeout_s,
        max_redirects=settings.media_max_redirects,
        max_image_bytes=settings.media_max_image_bytes,
        max_video_bytes=settings.media_max_video_bytes,
    )

MediaSecurityError

Bases: ValueError

Raised when a media reference violates the security policy.

fetch_url_bytes(url, *, max_bytes, policy=None)

Fetch a URL as bytes under the media policy (SSRF-guarded, size/time-capped).

Parameters:

Name	Type	Description	Default
url	str	The http/https URL to fetch.	required
max_bytes	int	Maximum number of bytes to read before aborting.	required
policy	MediaPolicy | None	Policy to apply; defaults to the active process policy.	None

Returns:

Type	Description
bytes	The response body bytes.

Raises:

Type	Description
MediaSecurityError	On any policy violation (bad scheme/host, too many redirects, or the response exceeding max_bytes).

Source code in pixano_inference/utils/media_security.py

def fetch_url_bytes(url: str, *, max_bytes: int, policy: MediaPolicy | None = None) -> bytes:
    """Fetch a URL as bytes under the media policy (SSRF-guarded, size/time-capped).

    Args:
        url: The http/https URL to fetch.
        max_bytes: Maximum number of bytes to read before aborting.
        policy: Policy to apply; defaults to the active process policy.

    Returns:
        The response body bytes.

    Raises:
        MediaSecurityError: On any policy violation (bad scheme/host, too many redirects,
            or the response exceeding *max_bytes*).
    """
    import requests

    policy = policy or get_media_policy()
    timeout = (policy.connect_timeout_s, policy.read_timeout_s)

    current = url
    with requests.Session() as session:
        for _ in range(policy.max_redirects + 1):
            _validate_url(current, policy)
            response = session.get(current, stream=True, allow_redirects=False, timeout=timeout)
            try:
                if response.is_redirect or response.is_permanent_redirect:
                    location = response.headers.get("Location")
                    if not location:
                        raise MediaSecurityError("Redirect response without a Location header.")
                    current = urljoin(current, location)
                    continue
                response.raise_for_status()
                return _read_capped(response, max_bytes)
            finally:
                response.close()
    raise MediaSecurityError(f"Too many redirects while fetching media (> {policy.max_redirects}).")

get_media_policy()

Return the active media policy, lazily building a secure default from the env.

Source code in pixano_inference/utils/media_security.py

def get_media_policy() -> MediaPolicy:
    """Return the active media policy, lazily building a secure default from the env."""
    global _ACTIVE_POLICY
    if _ACTIVE_POLICY is None:
        _ACTIVE_POLICY = MediaPolicy.from_env()
    return _ACTIVE_POLICY

is_http_url(value)

Whether value is an http/https URL (the only fetchable schemes).

Source code in pixano_inference/utils/media_security.py

def is_http_url(value: str) -> bool:
    """Whether *value* is an ``http``/``https`` URL (the only fetchable schemes)."""
    scheme = urlsplit(value).scheme.lower()
    return scheme in {"http", "https"}

resolve_local_path(raw_path, policy=None)

Resolve a client-supplied local path, enforcing containment under media_roots.

Parameters:

Name	Type	Description	Default
raw_path	str | Path	The path from the request.	required
policy	MediaPolicy | None	Policy to apply; defaults to the active process policy.	None

Returns:

Type	Description
Path	The resolved, real path.

Raises:

Type	Description
MediaSecurityError	If local-path media is disabled (no roots configured) or the path escapes every configured root.

Source code in pixano_inference/utils/media_security.py

def resolve_local_path(raw_path: str | Path, policy: MediaPolicy | None = None) -> Path:
    """Resolve a client-supplied local path, enforcing containment under ``media_roots``.

    Args:
        raw_path: The path from the request.
        policy: Policy to apply; defaults to the active process policy.

    Returns:
        The resolved, real path.

    Raises:
        MediaSecurityError: If local-path media is disabled (no roots configured) or the
            path escapes every configured root.
    """
    policy = policy or get_media_policy()
    if not policy.media_roots:
        raise MediaSecurityError("Local-path media is disabled. Configure PIXANO_INFERENCE_MEDIA_ROOTS to allow it.")
    resolved = Path(raw_path).expanduser().resolve()
    for root in policy.media_roots:
        try:
            resolved.relative_to(root)
            return resolved
        except ValueError:
            continue
    raise MediaSecurityError(f"Path '{raw_path}' is outside the allowed media roots.")

set_media_policy(policy)

Install the active media policy for this process.

Source code in pixano_inference/utils/media_security.py

def set_media_policy(policy: MediaPolicy) -> None:
    """Install the active media policy for this process."""
    global _ACTIVE_POLICY
    _ACTIVE_POLICY = policy