# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2026)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Server related utility functions."""

from __future__ import annotations

import os
from typing import TYPE_CHECKING, Final, cast

from streamlit import config, net_util, url_util
from streamlit.runtime.secrets import secrets_singleton

if TYPE_CHECKING:
    from collections.abc import Callable

# The port used for internal development.
DEVELOPMENT_PORT: Final = 3000
VITE_PORT_ENV_VAR: Final = "VITE_PORT"
PORT_ENV_VAR: Final = "PORT"

AUTH_COOKIE_NAME: Final = "_streamlit_user"
TOKENS_COOKIE_NAME: Final = "_streamlit_user_tokens"


def allowlisted_origins() -> set[str]:
    return {origin.strip() for origin in config.get_option("server.corsAllowedOrigins")}


def allow_all_cross_origin_requests() -> bool:
    """True if cross-origin requests from any origin are allowed.

    We allow ALL cross-origin requests when CORS protection has been disabled
    with server.enableCORS=False or when in dev mode (where the Vite dev server
    and backend use different ports, counting as two origins).
    """
    return not config.get_option("server.enableCORS") or config.get_option(
        "global.developmentMode"
    )


def is_allowed_origin(origin: str) -> bool:
    """Check if origin is in the allowlisted origins."""
    return origin in allowlisted_origins()


def is_url_from_allowed_origins(url: str) -> bool:
    """Return True if URL is from allowed origins (for CORS purpose).

    Allowed origins:
    1. localhost
    2. The internal and external IP addresses of the machine where this
       function was called from.

    If `server.enableCORS` is False, this allows all origins.
    """
    if not config.get_option("server.enableCORS"):
        # Allow everything when CORS is disabled.
        return True

    hostname = url_util.get_hostname(url)

    allowlisted_domains = [
        url_util.get_hostname(origin) for origin in allowlisted_origins()
    ]

    allowed_domains: list[str | Callable[[], str | None] | None] = [
        # Check localhost first.
        "localhost",
        "0.0.0.0",  # noqa: S104
        "127.0.0.1",
        # Try to avoid making unnecessary HTTP requests by checking if the user
        # manually specified a server address.
        _get_server_address_if_manually_set,
        # Then try the options that depend on HTTP requests or opening sockets.
        net_util.get_internal_ip,
        net_util.get_external_ip,
        *allowlisted_domains,
    ]

    for allowed_domain in allowed_domains:
        if isinstance(allowed_domain, str) or allowed_domain is None:
            allowed_domain_str = allowed_domain
        else:
            allowed_domain_str = allowed_domain()

        if allowed_domain_str is None:
            continue

        if hostname == allowed_domain_str:
            return True

    return False


def get_cookie_secret() -> str:
    """Get the cookie secret.

    If the user has not set a cookie secret, we generate a random one.
    """
    cookie_secret: str = config.get_option("server.cookieSecret")
    if secrets_singleton.load_if_toml_exists():
        auth_section = secrets_singleton.get("auth")
        if auth_section:
            cookie_secret = auth_section.get("cookie_secret", cookie_secret)
    return cookie_secret


def is_xsrf_enabled() -> bool:
    csrf_enabled = config.get_option("server.enableXsrfProtection")
    if not csrf_enabled and secrets_singleton.load_if_toml_exists():
        auth_section = secrets_singleton.get("auth", None)
        csrf_enabled = csrf_enabled or auth_section is not None
    return cast("bool", csrf_enabled)


def _get_server_address_if_manually_set() -> str | None:
    if config.is_manually_set("browser.serverAddress"):
        return url_util.get_hostname(config.get_option("browser.serverAddress"))
    return None


def get_display_address(address: str) -> str:
    """Get a display-friendly address for URLs shown to users.

    Wildcard addresses like "0.0.0.0" (all IPv4) or "::" (all interfaces)
    are not valid browser addresses on all platforms. This translates
    them to "localhost" for display purposes.

    Parameters
    ----------
    address
        The server address (IP or hostname).

    Returns
    -------
    str
        Address suitable for display. Wildcards become "localhost".
    """
    if address in {"0.0.0.0", "::"}:  # noqa: S104
        return "localhost"
    return address


def get_url(host_ip: str) -> str:
    """Get the URL for any app served at the given host_ip.

    Parameters
    ----------
    host_ip : str
        The IP address of the machine that is running the Streamlit Server.

    Returns
    -------
    str
        The URL.
    """
    protocol = "https" if config.get_option("server.sslCertFile") else "http"

    port = _get_browser_address_bar_port()
    base_path = config.get_option("server.baseUrlPath").strip("/")

    if base_path:
        base_path = "/" + base_path

    host_ip = host_ip.strip("/")
    return f"{protocol}://{host_ip}:{port}{base_path}"


def _get_browser_address_bar_port() -> int:
    """Get the app URL that will be shown in the browser's address bar.

    That is, this is the port where static assets will be served from. In dev,
    this is different from the URL that will be used to connect to the
    server-browser websocket.

    """
    if config.get_option("global.developmentMode"):
        for env_var in (VITE_PORT_ENV_VAR, PORT_ENV_VAR):
            port_str = os.environ.get(env_var)
            if not port_str:
                continue

            try:
                port = int(port_str)
            except ValueError:
                continue

            if 1 <= port <= 65535:
                return port

        return DEVELOPMENT_PORT
    return int(config.get_option("browser.serverPort"))
