64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""State-changing requests may originate only from explicitly configured sites.
|
|
|
|
The public Hosting & proxy URL can be stored in the database, while the CORS
|
|
environment setting still has its localhost default on an upgraded install.
|
|
Never infer a trusted origin from request Host or forwarded headers.
|
|
"""
|
|
|
|
from urllib.parse import urlsplit
|
|
from starlette.middleware.cors import CORSMiddleware
|
|
|
|
from ..config import settings
|
|
from ..installation_origin import managed_runtime
|
|
from .public_urls import magent_public_url, valid_public_url
|
|
|
|
|
|
def _origin(value: str, *, configured_url: bool = False) -> tuple[str, str, int] | None:
|
|
value = str(value or "")
|
|
if any(character.isspace() or ord(character) < 33 or ord(character) == 127 for character in value):
|
|
return None
|
|
if "?" in value or "#" in value:
|
|
return None
|
|
validated = valid_public_url(value)
|
|
if not validated:
|
|
return None
|
|
parsed = urlsplit(value)
|
|
if parsed.username is not None or parsed.password is not None:
|
|
return None
|
|
if not configured_url and parsed.path:
|
|
return None
|
|
return (
|
|
parsed.scheme.lower(),
|
|
parsed.hostname.lower(),
|
|
parsed.port or (443 if parsed.scheme == "https" else 80),
|
|
)
|
|
|
|
|
|
def is_allowed_request_origin(origin: str) -> bool:
|
|
candidate = _origin(origin)
|
|
if candidate is None:
|
|
return False
|
|
if managed_runtime():
|
|
# The operator confirms this address using the first-install token.
|
|
# No localhost fallback remains trusted after a managed installation.
|
|
return candidate == _origin(magent_public_url(), configured_url=True)
|
|
if candidate == _origin(str(settings.cors_allow_origin or "").rstrip("/")):
|
|
return True
|
|
return candidate == _origin(magent_public_url(), configured_url=True)
|
|
|
|
|
|
def can_claim_initial_origin() -> bool:
|
|
if not managed_runtime() or magent_public_url():
|
|
return False
|
|
from .setup import get_public_setup_status
|
|
return get_public_setup_status()["needs_admin"]
|
|
|
|
|
|
class ConfiguredOriginCORSMiddleware(CORSMiddleware):
|
|
"""Keep CORS response/preflight policy aligned with managed origin checks."""
|
|
|
|
def is_allowed_origin(self, origin: str) -> bool:
|
|
if managed_runtime():
|
|
return is_allowed_request_origin(origin)
|
|
return super().is_allowed_origin(origin)
|