prefect.cli.cloud.ip_allowlist

add(ctx, ip_address_or_range, description=typer.Option(None, '--description', '-d', help='A short description to annotate the entry with.')) async

Add a new IP entry to your account IP allowlist.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def add(
    ctx: typer.Context,
    ip_address_or_range: IP_ARGUMENT,
    description: Optional[str] = typer.Option(
        None,
        "--description",
        "-d",
        help="A short description to annotate the entry with.",
    ),
):
    """Add a new IP entry to your account IP allowlist."""
    new_entry = IPAllowlistEntry(
        ip_network=ip_address_or_range.parsed, description=description, enabled=True
    )

    async with get_cloud_client(infer_cloud_url=True) as client:
        ip_allowlist = await client.read_account_ip_allowlist()

        existing_entry_with_same_ip = None
        for entry in ip_allowlist.entries:
            if entry.ip_network == ip_address_or_range.parsed:
                existing_entry_with_same_ip = entry
                break

        if existing_entry_with_same_ip:
            if not typer.confirm(
                f"There's already an entry for this IP ({ip_address_or_range.raw}). Do you want to overwrite it?"
            ):
                exit_with_error("Aborted.")
            ip_allowlist.entries.remove(existing_entry_with_same_ip)

        ip_allowlist.entries.append(new_entry)

        try:
            await client.update_account_ip_allowlist(ip_allowlist)
        except PrefectHTTPStatusError as exc:
            _handle_update_error(exc)

        updated_ip_allowlist = await client.read_account_ip_allowlist()
        _print_ip_allowlist_table(
            updated_ip_allowlist, enabled=ctx.meta["enforce_ip_allowlist"]
        )

disable() async

Disable the IP allowlist for your account. When disabled, all IP addresses will be allowed to access your Prefect Cloud account.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def disable():
    """Disable the IP allowlist for your account. When disabled, all IP addresses will be allowed to access your Prefect Cloud account."""
    async with get_cloud_client(infer_cloud_url=True) as client:
        await client.update_account_settings({"enforce_ip_allowlist": False})

    exit_with_success("IP allowlist disabled.")

enable(ctx) async

Enable the IP allowlist for your account. When enabled, if the allowlist is non-empty, then access to your Prefect Cloud account will be restricted to only those IP addresses on the allowlist.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def enable(ctx: typer.Context):
    """Enable the IP allowlist for your account. When enabled, if the allowlist is non-empty, then access to your Prefect Cloud account will be restricted to only those IP addresses on the allowlist."""
    enforcing_ip_allowlist = ctx.meta["enforce_ip_allowlist"]
    if enforcing_ip_allowlist:
        exit_with_success("IP allowlist is already enabled.")

    async with get_cloud_client(infer_cloud_url=True) as client:
        my_access_if_enabled = await client.check_ip_allowlist_access()
        if not my_access_if_enabled.allowed:
            exit_with_error(
                f"Error enabling IP allowlist: {my_access_if_enabled.detail}"
            )

        logger.debug(my_access_if_enabled.detail)

        if not typer.confirm(
            "Enabling the IP allowlist will restrict Prefect Cloud API and UI access to only the IP addresses on the list. "
            "Continue?"
        ):
            exit_with_error("Aborted.")
        await client.update_account_settings({"enforce_ip_allowlist": True})

    exit_with_success("IP allowlist enabled.")

ls(ctx) async

Fetch and list all IP allowlist entries in your account.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def ls(ctx: typer.Context):
    """Fetch and list all IP allowlist entries in your account."""
    async with get_cloud_client(infer_cloud_url=True) as client:
        ip_allowlist = await client.read_account_ip_allowlist()

        _print_ip_allowlist_table(
            ip_allowlist, enabled=ctx.meta["enforce_ip_allowlist"]
        )

remove(ctx, ip_address_or_range) async

Remove an IP entry from your account IP allowlist.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def remove(ctx: typer.Context, ip_address_or_range: IP_ARGUMENT):
    """Remove an IP entry from your account IP allowlist."""
    async with get_cloud_client(infer_cloud_url=True) as client:
        ip_allowlist = await client.read_account_ip_allowlist()
        ip_allowlist.entries = [
            entry
            for entry in ip_allowlist.entries
            if entry.ip_network != ip_address_or_range.parsed
        ]

        try:
            await client.update_account_ip_allowlist(ip_allowlist)
        except PrefectHTTPStatusError as exc:
            _handle_update_error(exc)

        updated_ip_allowlist = await client.read_account_ip_allowlist()
        _print_ip_allowlist_table(
            updated_ip_allowlist, enabled=ctx.meta["enforce_ip_allowlist"]
        )

require_access_to_ip_allowlisting(ctx)

Enforce access to IP allowlisting for all subcommands.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.callback()
def require_access_to_ip_allowlisting(ctx: typer.Context):
    """Enforce access to IP allowlisting for all subcommands."""
    asyncio.run(_require_access_to_ip_allowlisting(ctx))

toggle(ctx, ip_address_or_range) async

Toggle the enabled status of an individual IP entry in your account IP allowlist.

Source code in src/prefect/cli/cloud/ip_allowlist.py

@ip_allowlist_app.command()
async def toggle(ctx: typer.Context, ip_address_or_range: IP_ARGUMENT):
    """Toggle the enabled status of an individual IP entry in your account IP allowlist."""
    async with get_cloud_client(infer_cloud_url=True) as client:
        ip_allowlist = await client.read_account_ip_allowlist()

        found_matching_entry = False
        for entry in ip_allowlist.entries:
            if entry.ip_network == ip_address_or_range.parsed:
                entry.enabled = not entry.enabled
                found_matching_entry = True
                break

        if not found_matching_entry:
            exit_with_error(
                f"No entry found with IP address `{ip_address_or_range.raw}`."
            )

        try:
            await client.update_account_ip_allowlist(ip_allowlist)
        except PrefectHTTPStatusError as exc:
            _handle_update_error(exc)

        updated_ip_allowlist = await client.read_account_ip_allowlist()
        _print_ip_allowlist_table(
            updated_ip_allowlist, enabled=ctx.meta["enforce_ip_allowlist"]
        )