decorators.py

"""CLI - Decorators."""

import functools

import click

from cloudsmith_cli.cli import validators

from ..core.api.init import initialise_api as _initialise_api
from ..core.credentials.chain import CredentialProviderChain
from ..core.credentials.models import CredentialContext
from ..core.mcp import server
from ..core.rest import create_requests_session as _create_session
from . import config, utils


def report_retry(seconds, context=None):
    if context == "retry-after":
        click.echo()
        click.echo(
            "Request was throttled (429): Retrying after %(seconds)s second(s) ... "
            % {"seconds": click.style(str(seconds), bold=True)}
        )


def _pop_boolean_flag(kwargs, name, invert=False):
    """Pop a boolean flag from kwargs, optionally inverting it."""
    value = kwargs.pop(name)
    if value is not None and invert:
        value = not value
    return value


def common_package_action_options(f):
    """Add common options for package actions."""

    @click.option(
        "-s",
        "--skip-errors",
        default=False,
        is_flag=True,
        help="Skip/ignore errors when copying packages.",
    )
    @click.option(
        "-W",
        "--no-wait-for-sync",
        default=False,
        is_flag=True,
        help="Don't wait for package synchronisation to complete before exiting.",
    )
    @click.option(
        "-I",
        "--wait-interval",
        default=5.0,
        type=float,
        show_default=True,
        help=(
            "The minimum time in seconds to wait between checking sync status after "
            "uploading. This is cumulative, so that status checks happen with less "
            "frequency over time, upto a maximum of 5 minutes of waiting."
        ),
    )
    @click.option(
        "--sync-attempts",
        default=3,
        type=int,
        help="Number of times to attempt package synchronisation. If the "
        "package fails the first time, the client will attempt to "
        "automatically resynchronise it.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_config_options(f):
    """Add common CLI config options to commands."""

    @click.option(
        "-C",
        "--config-file",
        envvar="CLOUDSMITH_CONFIG_FILE",
        type=click.Path(dir_okay=True, exists=True, writable=False, resolve_path=True),
        help="The path to your config.ini file.",
    )
    @click.option(
        "--credentials-file",
        envvar="CLOUDSMITH_CREDENTIALS_FILE",
        type=click.Path(dir_okay=True, exists=True, writable=False, resolve_path=True),
        help="The path to your credentials.ini file.",
    )
    @click.option(
        "-P",
        "--profile",
        default=None,
        envvar="CLOUDSMITH_PROFILE",
        help="The name of the profile to use for configuration.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        profile = kwargs.pop("profile") or ctx.meta.get("profile")
        config_file = kwargs.pop("config_file") or ctx.meta.get("config_file")
        creds_file = kwargs.pop("credentials_file") or ctx.meta.get("creds_file")

        # Store in context for subcommands to inherit
        ctx.meta["profile"] = profile
        ctx.meta["config_file"] = config_file
        ctx.meta["creds_file"] = creds_file

        opts.load_config_file(path=config_file, profile=profile)
        opts.load_creds_file(path=creds_file, profile=profile)
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_output_options(f):
    """Add common CLI output options to commands."""

    @click.option(
        "-d",
        "--debug",
        default=False,
        is_flag=True,
        help="Produce debug output during processing.",
    )
    @click.option(
        "-F",
        "--output-format",
        default="pretty",
        type=click.Choice(["pretty", "json", "pretty_json"]),
        help="Determines how output is formatted. This is only supported by a "
        "subset of the commands at the moment (e.g. list).",
    )
    @click.option(
        "-v",
        "--verbose",
        is_flag=True,
        default=False,
        help="Produce more output during processing.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        opts.debug = kwargs.pop("debug")
        opts.output = kwargs.pop("output_format")
        opts.verbose = kwargs.pop("verbose")
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_list_options(f):
    """Add common list options to commands."""

    @click.option(
        "-l",
        "--page-size",
        default=30,
        type=validators.IntOrWildcard(),
        help="The amount of items to view per page for lists. Use '*' to show all results.",
        callback=validators.validate_page_size,
    )
    @click.option(
        "-p",
        "--page",
        default=1,
        type=int,
        help="The page to view for lists, where 1 is the first page",
        callback=validators.validate_page,
    )
    @click.option(
        "--page-all",
        "--show-all",
        default=False,
        is_flag=True,
        help="Show all results. Cannot be used with --page (-p) or --page-size (-l).",
    )  # Validation performed post-parse for order-independence.
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)

        # Handle wildcard page_size (converts to page_all behavior)
        page_size = kwargs.get("page_size")
        if page_size == -1:
            kwargs["page_all"] = True
            # Validate that wildcard isn't used with explicit --page
            validators.enforce_page_all_exclusive(ctx, wildcard_used=True)

        # Order-independent validation: perform after all options parsed.
        page_all = kwargs.get("page_all")
        if page_all:
            validators.enforce_page_all_exclusive(ctx)
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_api_auth_options(f):
    """Add common API authentication options to commands."""

    @click.option(
        "-k",
        "--api-key",
        hide_input=True,
        envvar="CLOUDSMITH_API_KEY",
        help="The API key for authenticating with the API.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        api_key = kwargs.pop("api_key")
        if api_key:
            opts.api_key = api_key
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def initialise_session(f):
    """Create a shared HTTP session with proxy/SSL/user-agent settings."""

    @click.option(
        "--api-host", envvar="CLOUDSMITH_API_HOST", help="The API host to connect to."
    )
    @click.option(
        "--api-proxy",
        envvar="CLOUDSMITH_API_PROXY",
        help="The API proxy to connect through.",
    )
    @click.option(
        "-S",
        "--without-api-ssl-verify",
        default=None,
        is_flag=True,
        envvar="CLOUDSMITH_WITHOUT_API_SSL_VERIFY",
        help="Don't verify the SSL connection for the API. This is dangerous and "
        "should only be used in an old/broken environment that doesn't support the "
        "same secure ciphers as Cloudsmith.",
    )
    @click.option(
        "--api-user-agent",
        envvar="CLOUDSMITH_API_USER_AGENT",
        help="The user agent to use for requests.",
    )
    @click.option(
        "--api-headers",
        envvar="CLOUDSMITH_API_HEADERS",
        help="A CSV list of extra headers (key=value) to send to the API.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        opts.api_host = kwargs.pop("api_host")
        opts.api_proxy = kwargs.pop("api_proxy")
        opts.api_ssl_verify = _pop_boolean_flag(
            kwargs, "without_api_ssl_verify", invert=True
        )
        opts.api_user_agent = kwargs.pop("api_user_agent")
        opts.api_headers = kwargs.pop("api_headers")

        opts.session = _create_session(
            proxy=opts.api_proxy,
            ssl_verify=opts.api_ssl_verify,
            user_agent=opts.api_user_agent,
            headers=opts.api_headers,
        )

        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def resolve_credentials(f):
    """Resolve credentials via the provider chain. Depends on initialise_session."""

    @click.option(
        "--oidc-audience",
        envvar="CLOUDSMITH_OIDC_AUDIENCE",
        help="The OIDC audience for token requests.",
    )
    @click.option(
        "--oidc-org",
        envvar="CLOUDSMITH_ORG",
        help="The Cloudsmith organisation slug for OIDC token exchange.",
    )
    @click.option(
        "--oidc-service-slug",
        envvar="CLOUDSMITH_SERVICE_SLUG",
        help="The Cloudsmith service slug for OIDC token exchange.",
    )
    @click.option(
        "--oidc-discovery-disabled",
        default=None,
        is_flag=True,
        envvar="CLOUDSMITH_OIDC_DISCOVERY_DISABLED",
        help="Disable OIDC auto-discovery.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)

        oidc_audience = kwargs.pop("oidc_audience")
        oidc_org = kwargs.pop("oidc_org")
        oidc_service_slug = kwargs.pop("oidc_service_slug")
        oidc_discovery_disabled = _pop_boolean_flag(kwargs, "oidc_discovery_disabled")

        if oidc_audience:
            opts.oidc_audience = oidc_audience
        if oidc_org:
            opts.oidc_org = oidc_org
        if oidc_service_slug:
            opts.oidc_service_slug = oidc_service_slug
        if oidc_discovery_disabled:
            opts.oidc_discovery_disabled = oidc_discovery_disabled

        context = CredentialContext(
            session=opts.session,
            api_key=opts.api_key,
            api_host=opts.api_host or "https://api.cloudsmith.io",
            creds_file_path=ctx.meta.get("creds_file"),
            profile=ctx.meta.get("profile"),
            debug=opts.debug,
            oidc_audience=opts.oidc_audience,
            oidc_org=opts.oidc_org,
            oidc_service_slug=opts.oidc_service_slug,
            oidc_discovery_disabled=opts.oidc_discovery_disabled,
        )

        chain = CredentialProviderChain()
        credential = chain.resolve(context)

        if context.keyring_refresh_failed:
            click.secho(
                "An error occurred when attempting to refresh your SSO access token. "
                "To refresh this session, run 'cloudsmith auth'",
                fg="yellow",
                err=True,
            )
            if credential:
                click.secho(
                    "Falling back to API key authentication.",
                    fg="yellow",
                    err=True,
                )

        opts.credential = credential

        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return initialise_session(wrapper)


def initialise_api(f):
    """Initialise the Cloudsmith API for use. Depends on resolve_credentials."""

    @click.option(
        "-R",
        "--without-rate-limit",
        default=None,
        is_flag=True,
        help="Don't obey the suggested rate limit interval. The CLI will "
        "otherwise automatically sleep between commands to ensure that you do "
        "not hit the server-side rate limit.",
    )
    @click.option(
        "--rate-limit-warning",
        default=30,
        help="When rate limiting, display information that it is happening "
        "if wait interval is higher than this setting. By default no "
        "information will be printed. Set to zero to always see it.",
    )
    @click.option(
        "--error-retry-max",
        default=6,
        help="The maximum amount of times to retry on errors received from "
        "the API, as determined by the --error-retry-codes parameter.",
    )
    @click.option(
        "--error-retry-backoff",
        default=1.0,
        type=float,
        help="The backoff factor determines how long to wait in seconds "
        "between error retries. The backoff factor is multiplied by the "
        "amount of retries so far. So if 1.0, then the wait is 1.0s then "
        "2.0s, then 4.0s, and so forth.",
    )
    @click.option(
        "--error-retry-codes",
        default="429,500,502,503,504",
        help="The status codes that when received from the API will cause "
        "a retry (if --error-retry-max is > 0). By default this will be for "
        "429, 500, 502, 503 and 504 error codes.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        opts.rate_limit = _pop_boolean_flag(kwargs, "without_rate_limit", invert=True)
        opts.rate_limit_warning = kwargs.pop("rate_limit_warning")
        opts.error_retry_max = kwargs.pop("error_retry_max")
        opts.error_retry_backoff = kwargs.pop("error_retry_backoff")
        opts.error_retry_codes = kwargs.pop("error_retry_codes")
        opts.error_retry_cb = report_retry

        def call_print_rate_limit_info_with_opts(rate_info):
            utils.print_rate_limit_info(opts, rate_info)

        opts.api_config = _initialise_api(
            debug=opts.debug,
            host=opts.api_host,
            credential=opts.credential,
            proxy=opts.api_proxy,
            ssl_verify=opts.api_ssl_verify,
            user_agent=opts.api_user_agent,
            headers=opts.api_headers,
            rate_limit=opts.rate_limit,
            rate_limit_callback=call_print_rate_limit_info_with_opts,
            error_retry_max=opts.error_retry_max,
            error_retry_backoff=opts.error_retry_backoff,
            error_retry_codes=opts.error_retry_codes,
            error_retry_cb=opts.error_retry_cb,
        )

        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return resolve_credentials(wrapper)


def initialise_mcp(f):
    @click.option(
        "-a",
        "--all-tools",
        default=False,
        is_flag=True,
        help="Show all tools",
    )
    @click.option(
        "-D",
        "--allow-destructive-tools",
        default=False,
        is_flag=True,
        help="Allow destructive tools to be used",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        opts = kwargs.get("opts")

        all_tools = kwargs.pop("all_tools")
        allow_destructive_tools = kwargs.pop("allow_destructive_tools")

        mcp_server = server.DynamicMCPServer(
            api_config=opts.api_config,
            debug_mode=opts.debug,
            allow_destructive_tools=allow_destructive_tools,
            allowed_tool_groups=opts.mcp_allowed_tool_groups,
            allowed_tools=opts.mcp_allowed_tools,
            force_all_tools=all_tools,
        )
        kwargs["mcp_server"] = mcp_server
        return ctx.invoke(f, *args, **kwargs)

    return wrapper