smart-org/apps/api/app/domain/platform/rules.py

"""Platform-wide validation rules carried over from the existing infrastructure.

These are almost all pure functions with no I/O, so both the application layer
*and* the adapters can enforce them. Defence in depth: a rule that is only
checked in one place is a rule that will eventually be bypassed.

The one exception is :func:`assert_remote_host_allowed`, which resolves a
hostname. It has to: a *name* tells you nothing about where it points, and the
whole point of the check is to refuse a name that resolves into the private
network. The address classification itself is pure
(:func:`is_blocked_address`), so the policy is testable without DNS.

Reference: ``docs/current-infra-notes.md`` §1.
"""

from __future__ import annotations

import ipaddress
import re
import socket
from urllib.parse import urlsplit

DEFAULT_ROOT_DOMAIN = "msa.artacloud.ir"

#: Ports that are never allocatable (system services + data stores).
SYSTEM_RESERVED_PORTS: frozenset[int] = frozenset(
    {22, 53, 80, 443, 2019, 3000, 5432, 6379, 8008}
)

#: Ports of internal platform services. A tenant preview must never proxy to
#: these, otherwise a public hostname would expose another component.
SYSTEM_PROTECTED_PORTS: frozenset[int] = (
    frozenset({3000, 35555, 35556, 35557}) | SYSTEM_RESERVED_PORTS
)

DEFAULT_PORT_RANGE_START = 8080
DEFAULT_PORT_RANGE_END = 8999

LABEL_MIN = 3
LABEL_MAX = 40
_LABEL_RE = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$")
_PROXY_TARGET_RE = re.compile(r"^(https?://)?(localhost|127\.0\.0\.1):(\d{1,5})$")


class RuleViolation(ValueError):
    """Raised when a request violates a platform safety rule."""


def validate_subdomain_label(label: str) -> str:
    """Normalise and validate a single DNS label.

    Only ``a-z0-9-``, must start and end alphanumeric, 3-40 characters.
    """
    clean = (label or "").strip().lower()
    if not (LABEL_MIN <= len(clean) <= LABEL_MAX):
        raise RuleViolation(
            f"نام زیردامنه باید بین {LABEL_MIN} تا {LABEL_MAX} کاراکتر باشد."
        )
    if not _LABEL_RE.match(clean):
        raise RuleViolation("نام زیردامنه فقط می‌تواند شامل حروف کوچک، عدد و خط تیره باشد.")
    return clean


def build_host(label: str, root_domain: str = DEFAULT_ROOT_DOMAIN) -> str:
    return f"{validate_subdomain_label(label)}.{root_domain.strip().lower()}"


def validate_managed_host(host: str, root_domain: str = DEFAULT_ROOT_DOMAIN) -> str:
    """Validate a fully-qualified host against the managed wildcard domain.

    Rejects anything that is not a single label under the root domain — this is
    the anti-injection check that keeps a crafted host from breaking out of a
    Caddy site block or pointing at an arbitrary host.
    """
    clean = (host or "").strip().lower()
    suffix = f".{root_domain.strip().lower()}"
    if not clean.endswith(suffix):
        raise RuleViolation(f"دامنه باید زیرمجموعهٔ «{root_domain}» باشد.")
    label = clean[: -len(suffix)]
    if "." in label:
        raise RuleViolation("فقط یک برچسب (single label) مجاز است.")
    validate_subdomain_label(label)
    return clean


def validate_proxy_target(target: str) -> tuple[str, int]:
    """Ensure the reverse-proxy target is a loopback address with a numeric port.

    Blocks SSRF: a tenant must not be able to proxy a public hostname to an
    internal service.
    """
    clean = (target or "").strip()
    match = _PROXY_TARGET_RE.match(clean)
    if not match:
        raise RuleViolation(
            "مقصد reverse proxy باید localhost یا 127.0.0.1 با پورت عددی باشد."
        )
    port = int(match.group(3))
    if not (1 <= port <= 65535):
        raise RuleViolation("شمارهٔ پورت نامعتبر است.")
    return match.group(2), port


def is_port_allocatable(
    port: int,
    *,
    start: int = DEFAULT_PORT_RANGE_START,
    end: int = DEFAULT_PORT_RANGE_END,
) -> bool:
    if port in SYSTEM_RESERVED_PORTS:
        return False
    return start <= port <= end


def assert_port_allocatable(
    port: int,
    *,
    start: int = DEFAULT_PORT_RANGE_START,
    end: int = DEFAULT_PORT_RANGE_END,
) -> int:
    if not is_port_allocatable(port, start=start, end=end):
        raise RuleViolation(f"پورت {port} قابل تخصیص نیست.")
    return port


def assert_not_protected_port(port: int) -> int:
    """A preview domain must never point at an internal platform port."""
    if port in SYSTEM_PROTECTED_PORTS:
        raise RuleViolation(f"اتصال به پورت محافظت‌شدهٔ {port} ممنوع است.")
    return port


# ---------------------------------------------------------------------------
# SSRF: user-supplied remote hosts (git remotes, deploy targets)
# ---------------------------------------------------------------------------

#: Ports that are always refused on a user-supplied host, whatever the address.
#: A remote git or deploy target has no business talking to a database, a cache
#: or the Docker daemon. Port 22 is deliberately *not* here: git-over-ssh and
#: SFTP deploys are legitimate.
BLOCKED_REMOTE_PORTS: frozenset[int] = frozenset(
    {2375, 2376, 3306, 5432, 6379, 9200, 11211, 27017}
)

#: Hostnames that are never a valid user-supplied remote. `localhost` is a name,
#: not an address, so it needs its own rule — DNS would resolve it to loopback,
#: but only if we remembered to resolve.
_BLOCKED_HOSTNAMES: frozenset[str] = frozenset(
    {
        "localhost",
        "localhost.localdomain",
        "ip6-localhost",
        "ip6-loopback",
        "metadata",
        "metadata.google.internal",
        "instance-data",
    }
)

_ALLOWED_REMOTE_SCHEMES: frozenset[str] = frozenset(
    {"http", "https", "ssh", "git", "ftp", "ftps", "sftp"}
)


def is_blocked_address(address: str) -> bool:
    """True when an IP literal must not be the target of a user-supplied remote.

    Covers every way to reach something that is not the public internet:
    loopback, RFC1918 private, link-local (including the cloud metadata address
    ``169.254.169.254``), unique-local IPv6, multicast, reserved, unspecified,
    and IPv4-mapped IPv6 forms of all of the above.
    """
    try:
        ip = ipaddress.ip_address(address)
    except ValueError:
        return False  # not an IP literal; the DNS path handles it

    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        # ::ffff:127.0.0.1 is loopback in disguise.
        ip = ip.ipv4_mapped

    return bool(
        ip.is_loopback
        or ip.is_private
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
    )


def assert_remote_host_allowed(
    url: str, *, allow_private: bool = False, resolver=None
) -> tuple[str, int | None]:
    """Validate a user-supplied remote URL and return ``(host, port)``.

    Runs before *any* outbound connection — a git remote test, a deploy
    connection — because a user-supplied URL is the classic SSRF vector:
    ``http://169.254.169.254/`` reaches cloud instance credentials, and
    ``http://127.0.0.1:6379/`` reaches Redis.

    ``resolver`` is injectable so the DNS step can be faked in tests; it defaults
    to ``socket.getaddrinfo``. Every address the name resolves to must pass, not
    just the first: a name can legitimately have both a public and a private
    address, and returning either would defeat the check.
    """
    clean = (url or "").strip()
    if not clean:
        raise RuleViolation("آدرس مقصد خالی است.")
    if any(ch.isspace() for ch in clean):
        raise RuleViolation("آدرس مقصد نباید فاصله داشته باشد.")

    # A bare ``git@host:path`` scp-style remote has no scheme; normalise it so
    # urlsplit can do its job.
    if "://" not in clean and re.match(r"^[^/@\s]+@[^/:\s]+:", clean):
        clean = "ssh://" + clean.replace(":", "/", 1)

    parts = urlsplit(clean)
    if parts.scheme.lower() not in _ALLOWED_REMOTE_SCHEMES:
        raise RuleViolation(
            "پروتکل آدرس مقصد پشتیبانی نمی‌شود."
        )
    host = parts.hostname or ""
    if not host:
        raise RuleViolation("میزبان آدرس مقصد مشخص نیست.")
    if parts.password is not None:
        # Credentials in a URL end up in logs, in `ps` and in the git remote
        # config. They must be supplied as credentials, not as part of the URL.
        raise RuleViolation("آدرس مقصد نباید شامل نام کاربری و رمز باشد.")

    port = parts.port
    if port is not None:
        if not (1 <= port <= 65535):
            raise RuleViolation("شمارهٔ پورت مقصد نامعتبر است.")
        if port in BLOCKED_REMOTE_PORTS:
            raise RuleViolation(f"پورت {port} برای مقصد مجاز نیست.")

    if allow_private:
        return host, port

    lowered = host.lower().strip(".")
    if lowered in _BLOCKED_HOSTNAMES or lowered.endswith(".localhost"):
        raise RuleViolation("اتصال به میزبان محلی مجاز نیست.")

    # An IP literal can be judged directly, without DNS.
    try:
        ipaddress.ip_address(lowered.strip("[]"))
    except ValueError:
        pass
    else:
        if is_blocked_address(lowered.strip("[]")):
            raise RuleViolation("اتصال به آدرس‌های داخلی مجاز نیست.")
        return host, port

    lookup = resolver or socket.getaddrinfo
    try:
        infos = lookup(host, port or 0, type=socket.SOCK_STREAM)
    except (socket.gaierror, UnicodeError, OSError) as exc:
        raise RuleViolation("میزبان مقصد پیدا نشد.") from exc

    # ``getaddrinfo`` returns a family-dependent sockaddr, so the address is
    # coerced to str explicitly rather than relying on the overload's inference.
    addresses: set[str] = {str(info[4][0]) for info in infos}
    if not addresses:
        raise RuleViolation("میزبان مقصد پیدا نشد.")
    for address in addresses:
        if is_blocked_address(address):
            raise RuleViolation("اتصال به آدرس‌های داخلی مجاز نیست.")

    return host, port