131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
"""Refuse to fetch a URL that points back inside the network.
|
|
|
|
An identity provider is configured by a workspace administrator, who supplies an
|
|
issuer URL that the **server** then fetches — discovery documents, JWKS, token
|
|
exchange. That is a request the platform makes on a customer's instruction, to
|
|
wherever the customer says, from inside the network. Unchecked, it is server-side
|
|
request forgery: `http://169.254.169.254/` reaches the cloud metadata service,
|
|
`http://localhost:5432` reaches the database, and an internal admin panel is one
|
|
hostname away.
|
|
|
|
Two checks, and the second is the one usually missed:
|
|
|
|
1. **The host must resolve to a public address.** A literal is checked directly;
|
|
a name is resolved and every address it returns must be global, because a
|
|
name with one public and one private answer is a name that will eventually
|
|
return the private one.
|
|
|
|
2. **The connection is pinned to the address that was checked.** Resolving and
|
|
then handing the *hostname* to the HTTP client leaves a gap: the name can
|
|
resolve again, to something else, between the check and the request. That is
|
|
DNS rebinding, and it defeats a check that only looks at the name.
|
|
|
|
Ported from the platform, which needed it for the same reason.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import socket
|
|
from ipaddress import IPv4Address, IPv6Address
|
|
from typing import Optional
|
|
from urllib.parse import SplitResult, urlsplit, urlunsplit
|
|
|
|
ALLOWED_SCHEMES = ("http", "https")
|
|
|
|
_PRIVATE_ADDRESS_MESSAGE = (
|
|
"URL host resolves to a private, loopback or otherwise reserved address, "
|
|
"which is not allowed"
|
|
)
|
|
|
|
|
|
class PrivateAddressError(Exception):
|
|
"""The target is inside the network, or cannot be shown not to be."""
|
|
|
|
def __init__(self, message: str = _PRIVATE_ADDRESS_MESSAGE) -> None:
|
|
super().__init__(message)
|
|
|
|
|
|
def _as_ip_literal(host: str) -> Optional[IPv4Address | IPv6Address]:
|
|
try:
|
|
return ipaddress.ip_address(host.strip("[]"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def resolve_public_address(host: str, port: Optional[int] = None) -> str:
|
|
"""The address to connect to, or a refusal.
|
|
|
|
Every address the name resolves to has to be global. Accepting a name
|
|
because *one* of its answers is public would accept a name that alternates,
|
|
which is the whole trick.
|
|
"""
|
|
literal = _as_ip_literal(host)
|
|
if literal is not None:
|
|
if not literal.is_global:
|
|
raise PrivateAddressError()
|
|
return str(literal)
|
|
|
|
try:
|
|
infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
|
|
except socket.gaierror as e:
|
|
raise PrivateAddressError(f"URL host could not be resolved: {e}")
|
|
|
|
if not infos:
|
|
raise PrivateAddressError("URL host could not be resolved")
|
|
|
|
addresses = [ipaddress.ip_address(info[4][0]) for info in infos]
|
|
if any(not address.is_global for address in addresses):
|
|
raise PrivateAddressError()
|
|
return str(addresses[0])
|
|
|
|
|
|
def pin_url_to_address(url: str, address: str) -> str:
|
|
"""Rewrite the URL to connect to a specific address.
|
|
|
|
Closes the window between checking a name and using it. The `Host` header
|
|
still has to carry the original name for TLS and virtual hosting, which is
|
|
the caller's job.
|
|
"""
|
|
parts = urlsplit(url)
|
|
literal = f"[{address}]" if ":" in address else address
|
|
netloc = f"{literal}:{parts.port}" if parts.port else literal
|
|
return urlunsplit(
|
|
(parts.scheme, netloc, parts.path or "/", parts.query, parts.fragment)
|
|
)
|
|
|
|
|
|
def _url_shape_error(parts: SplitResult, require_https: bool) -> Optional[str]:
|
|
if parts.scheme not in ALLOWED_SCHEMES:
|
|
return f"URL scheme must be one of {', '.join(ALLOWED_SCHEMES)}"
|
|
if require_https and parts.scheme != "https":
|
|
return "URL must use https"
|
|
if not parts.hostname:
|
|
return "URL has no host"
|
|
if parts.username or parts.password:
|
|
return "URL must not contain credentials"
|
|
return None
|
|
|
|
|
|
def url_destination_error(url: str, *, require_https: bool = True) -> Optional[str]:
|
|
"""Why this URL cannot be fetched, or None if it can.
|
|
|
|
Returns a message rather than raising so a configuration form can show it
|
|
next to the field, which is where somebody can act on it.
|
|
"""
|
|
try:
|
|
parts = urlsplit(url)
|
|
except ValueError:
|
|
return "URL could not be parsed"
|
|
|
|
shape = _url_shape_error(parts, require_https)
|
|
if shape:
|
|
return shape
|
|
|
|
try:
|
|
resolve_public_address(parts.hostname, parts.port)
|
|
except PrivateAddressError as e:
|
|
return str(e)
|
|
|
|
return None
|