I prefer using Cloudflare for its ease of use and super generous free tier offering. I like AWS Lightsail for similar reasons -- you can spin up, essentially, a minimal EC2 Linux instance for $5-$10 a month and it has a lot of configuration benefits out of the box. One of those being free DNS (no Route53 costs for zones running on Lightsail) and easy firewall configs with a simple GUI.

However, it's annoying to keep Cloudflare's IP ranges manually updated on the Lightsail instance firewall. So, I decided to take a lightweight automated approach to this.

Overview

We'll be using the AWS Eventbridge Scheduler to act as a cron that runs an AWS Lambda function once per day. This function will call the Lightsail API to get the firewall configs, get the Cloudflare IP ranges via a simple GET request of the text files they publish, and then make updates to the Lightsail firewall if there's any mismatch.

The Lambda

This Python only touches ports 80 and 443. Beyond what's described above, it'll also error if the Cloudflare IP range files are suspiciously small, assuming there's some error with the HTTP response received. Lastly, "LIGHTSAIL_INSTANCE_NAME" is a required environmental variable for the function that must match that of your Lightsail instance name. There are some other optional environmental variables noted in the comments of the funciton, notably DRY_RUN=1 if you just want the log output but not take action while testing... AI wrote most of this code -- prompt first coding is my new norm.

"""
Lightsail + Cloudflare Origin Lockdown (Ports 80/443)

Goal:
  Keep your Lightsail instance reachable on 80/443 ONLY from Cloudflare IP ranges,
  while preserving all other ports and rules.

Key safety properties:
  - Non-destructive: uses OpenInstancePublicPorts / CloseInstancePublicPorts (never PutInstancePublicPorts)
  - Guardrails: aborts if Cloudflare lists are empty or suspiciously small
  - Retries: exponential backoff + jitter for both HTTP fetch and AWS API calls
  - Override allowlist: hard-code extra CIDRs you want allowed in addition to Cloudflare

Cloudflare IP sources:
  - https://www.cloudflare.com/ips-v4
  - https://www.cloudflare.com/ips-v6

Environment variables:
  Required:
    LIGHTSAIL_INSTANCE_NAME   (e.g., "my-lightsail-instance")

  Optional:
    DRY_RUN=1                 (log changes but do not apply)
    LOG_LEVEL=DEBUG|INFO|...  (default INFO)

    MAX_RETRIES=6             (default 6)
    RETRY_BASE_SECONDS=0.75   (default 0.75)

    MIN_CF_IPV4_COUNT=5      (default 5)
    MIN_CF_IPV6_COUNT=3      (default 3)
    REQUIRE_BOTH_FAMILIES=1   (default 1)

IAM permissions (Lambda role):
  - lightsail:GetInstancePortStates
  - lightsail:OpenInstancePublicPorts
  - lightsail:CloseInstancePublicPorts
"""

from __future__ import annotations

import json
import logging
import os
import random
import time
from datetime import date, datetime
from typing import Any, Dict, List, Optional, Set, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import boto3
from botocore.exceptions import (
    ClientError,
    ConnectionClosedError,
    EndpointConnectionError,
)

# =============================================================================
# User overrides (hard-code)
# =============================================================================
# Add extra CIDRs you want allowed IN ADDITION to Cloudflare for ports 80/443.
# Examples:
#   - "203.0.113.10/32"   (IPv4 /32)
#   - "2001:db8::/128"    (IPv6 /128)
EXTRA_IPV4_CIDRS: Set[str] = {
    # "203.0.113.10/32",
}
EXTRA_IPV6_CIDRS: Set[str] = {
    # "2001:db8::/128",
}

# Only these ports are managed; nothing else is modified.
MANAGED_PORTS = (80, 443)
MANAGED_PROTOCOL = "tcp"

# Cloudflare public endpoints (plain text)
CF_IPV4_URL = "https://www.cloudflare.com/ips-v4"
CF_IPV6_URL = "https://www.cloudflare.com/ips-v6"

# =============================================================================
# Safety limits / runtime config
# =============================================================================
# Guardrail: if Cloudflare lists are unexpectedly small, abort and make no changes.
MIN_CF_IPV4_COUNT = int(os.environ.get("MIN_CF_IPV4_COUNT", "5"))
MIN_CF_IPV6_COUNT = int(os.environ.get("MIN_CF_IPV6_COUNT", "3"))
REQUIRE_BOTH_FAMILIES = os.environ.get("REQUIRE_BOTH_FAMILIES", "1").strip() == "1"

# Dry-run: calculate diffs + log, but do not open/close any ports.
DRY_RUN = os.environ.get("DRY_RUN", "0").strip() == "1"

# Retries
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "6"))
RETRY_BASE_SECONDS = float(os.environ.get("RETRY_BASE_SECONDS", "0.75"))

# Logging
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
logger = logging.getLogger()
logger.setLevel(LOG_LEVEL)

_CF_HEADERS = {
    "User-Agent": "lightsail-cf-origin-lockdown/1.0",
    "Accept": "text/plain,*/*",
}


# =============================================================================
# Logging helpers + JSON safety (Lambda response must be JSON-serializable)
# =============================================================================
def json_safe(value: Any) -> Any:
    """Convert common non-JSON types (like datetime) into JSON-safe equivalents."""
    if value is None:
        return None
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    if isinstance(value, dict):
        return {str(k): json_safe(v) for k, v in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [json_safe(v) for v in value]
    try:
        json.dumps(value)
        return value
    except TypeError:
        return str(value)


def log(msg: str, **fields: Any) -> None:
    if fields:
        logger.info("%s | %s", msg, json.dumps(json_safe(fields), separators=(",", ":"), sort_keys=True))
    else:
        logger.info("%s", msg)


def respond(ok: bool, **payload: Any) -> Dict[str, Any]:
    return json_safe({"ok": ok, **payload})


# =============================================================================
# Generic retry wrapper
# =============================================================================
def backoff_sleep_seconds(attempt: int, base: float) -> float:
    """Exponential backoff with jitter."""
    return base * (2 ** (attempt - 1)) + random.uniform(0, base)


def trim_boto_response(resp: Any) -> Any:
    """Reduce log noise; also avoids issues with datetime in ops objects."""
    if not isinstance(resp, dict):
        return {"resp": str(resp)}
    # Operations can contain datetimes; json_safe handles them.
    for key in ("operations", "operation", "warnings"):
        if key in resp:
            return json_safe({key: resp[key]})
    # Fallback: first few keys
    keys = list(resp.keys())[:6]
    return json_safe({k: resp[k] for k in keys})


def call_with_retries(fn, *, op_name: str, max_retries: int, base_sleep: float, **kwargs):
    last_err: Optional[Exception] = None

    for attempt in range(1, max_retries + 1):
        try:
            log(f"{op_name}: attempt", attempt=attempt, kwargs=kwargs)
            resp = fn(**kwargs)
            log(f"{op_name}: success", response=trim_boto_response(resp))
            return resp

        except (EndpointConnectionError, ConnectionClosedError) as e:
            last_err = e
            if attempt == max_retries:
                break
            sleep_s = backoff_sleep_seconds(attempt, base_sleep)
            log(f"{op_name}: network error, retrying", attempt=attempt, error=str(e), sleep_seconds=sleep_s)
            time.sleep(sleep_s)

        except ClientError as e:
            last_err = e
            code = e.response.get("Error", {}).get("Code", "Unknown")
            msg = e.response.get("Error", {}).get("Message", str(e))

            retryable = code in {
                "ThrottlingException",
                "TooManyRequestsException",
                "ServiceUnavailableException",
                "InternalFailure",
                "InternalError",
                "RequestLimitExceeded",
            }
            if retryable and attempt < max_retries:
                sleep_s = backoff_sleep_seconds(attempt, base_sleep)
                log(
                    f"{op_name}: retryable ClientError, retrying",
                    attempt=attempt,
                    code=code,
                    message=msg,
                    sleep_seconds=sleep_s,
                )
                time.sleep(sleep_s)
                continue

            log(f"{op_name}: ClientError (fatal)", code=code, message=msg, full_error=e.response)
            raise

        except Exception as e:
            last_err = e
            log(f"{op_name}: unexpected error (fatal)", error=str(e))
            raise

    raise RuntimeError(f"{op_name} failed after {max_retries} attempts: {last_err}")


# =============================================================================
# Cloudflare list fetch + validation
# =============================================================================
def http_get_text_with_retries(url: str, *, max_retries: int, base_sleep: float, timeout: int = 20) -> str:
    last_err: Optional[Exception] = None

    for attempt in range(1, max_retries + 1):
        try:
            req = Request(url, method="GET", headers=_CF_HEADERS)
            with urlopen(req, timeout=timeout) as resp:
                text = resp.read().decode("utf-8", errors="replace")

            if len(text.strip()) < 10:
                raise RuntimeError(f"Response too small/blank from {url}")
            return text

        except (HTTPError, URLError, TimeoutError, RuntimeError) as e:
            last_err = e
            if attempt == max_retries:
                break
            sleep_s = backoff_sleep_seconds(attempt, base_sleep)
            log("cloudflare_fetch: retrying", url=url, attempt=attempt, error=str(e), sleep_seconds=sleep_s)
            time.sleep(sleep_s)

    raise RuntimeError(f"Failed to fetch {url} after {max_retries} attempts: {last_err}")


def parse_cidrs(text: str) -> Set[str]:
    cidrs: Set[str] = set()
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        cidrs.add(line)
    return cidrs


def fetch_cloudflare_cidrs_txt(*, max_retries: int, base_sleep: float) -> Tuple[Set[str], Set[str]]:
    v4_text = http_get_text_with_retries(CF_IPV4_URL, max_retries=max_retries, base_sleep=base_sleep)
    v6_text = http_get_text_with_retries(CF_IPV6_URL, max_retries=max_retries, base_sleep=base_sleep)

    v4 = parse_cidrs(v4_text)
    v6 = parse_cidrs(v6_text)

    if REQUIRE_BOTH_FAMILIES and (not v4 or not v6):
        raise RuntimeError(f"Cloudflare lists invalid (require both). ipv4={len(v4)} ipv6={len(v6)}")
    if not REQUIRE_BOTH_FAMILIES and (not v4 and not v6):
        raise RuntimeError("Cloudflare lists invalid (both empty).")

    if len(v4) < MIN_CF_IPV4_COUNT:
        raise RuntimeError(f"Cloudflare ipv4 list too small ({len(v4)} < {MIN_CF_IPV4_COUNT})")
    if len(v6) < MIN_CF_IPV6_COUNT:
        raise RuntimeError(f"Cloudflare ipv6 list too small ({len(v6)} < {MIN_CF_IPV6_COUNT})")

    return v4, v6


# =============================================================================
# Lightsail state + sync (non-destructive to other ports)
# =============================================================================
def get_port_states(client, instance_name: str, *, max_retries: int, base_sleep: float) -> List[Dict[str, Any]]:
    resp = call_with_retries(
        client.get_instance_port_states,
        op_name="get_instance_port_states",
        max_retries=max_retries,
        base_sleep=base_sleep,
        instanceName=instance_name,
    )
    states = resp.get("portStates", []) or []
    normalized: List[Dict[str, Any]] = []

    for s in states:
        normalized.append(
            {
                "fromPort": int(s["fromPort"]),
                "toPort": int(s.get("toPort", s["fromPort"])),
                "protocol": str(s["protocol"]).lower(),
                "cidrs": set(s.get("cidrs") or []),
                "ipv6Cidrs": set(s.get("ipv6Cidrs") or []),
            }
        )

    return normalized


def find_exact_port_rule(states: List[Dict[str, Any]], port: int, protocol: str) -> Optional[Dict[str, Any]]:
    protocol = protocol.lower()
    for s in states:
        if s["protocol"] == protocol and s["fromPort"] == port and s["toPort"] == port:
            return s
    return None


def open_cidrs(
    client,
    instance_name: str,
    port: int,
    protocol: str,
    ipv4_to_add: List[str],
    ipv6_to_add: List[str],
    *,
    max_retries: int,
    base_sleep: float,
) -> None:
    if ipv4_to_add:
        call_with_retries(
            client.open_instance_public_ports,
            op_name="open_instance_public_ports(v4)",
            max_retries=max_retries,
            base_sleep=base_sleep,
            instanceName=instance_name,
            portInfo={"fromPort": port, "toPort": port, "protocol": protocol, "cidrs": ipv4_to_add},
        )
    if ipv6_to_add:
        call_with_retries(
            client.open_instance_public_ports,
            op_name="open_instance_public_ports(v6)",
            max_retries=max_retries,
            base_sleep=base_sleep,
            instanceName=instance_name,
            portInfo={"fromPort": port, "toPort": port, "protocol": protocol, "ipv6Cidrs": ipv6_to_add},
        )


def close_cidrs(
    client,
    instance_name: str,
    port: int,
    protocol: str,
    ipv4_to_remove: List[str],
    ipv6_to_remove: List[str],
    *,
    max_retries: int,
    base_sleep: float,
) -> None:
    if ipv4_to_remove:
        call_with_retries(
            client.close_instance_public_ports,
            op_name="close_instance_public_ports(v4)",
            max_retries=max_retries,
            base_sleep=base_sleep,
            instanceName=instance_name,
            portInfo={"fromPort": port, "toPort": port, "protocol": protocol, "cidrs": ipv4_to_remove},
        )
    if ipv6_to_remove:
        call_with_retries(
            client.close_instance_public_ports,
            op_name="close_instance_public_ports(v6)",
            max_retries=max_retries,
            base_sleep=base_sleep,
            instanceName=instance_name,
            portInfo={"fromPort": port, "toPort": port, "protocol": protocol, "ipv6Cidrs": ipv6_to_remove},
        )


# =============================================================================
# Lambda handler
# =============================================================================
def lambda_handler(event, context):
    start = time.time()

    instance_name = os.environ.get("LIGHTSAIL_INSTANCE_NAME", "").strip()
    if not instance_name:
        return respond(False, stage="config", error="Missing required env var: LIGHTSAIL_INSTANCE_NAME")

    region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")

    log(
        "sync starting",
        instance=instance_name,
        region=region,
        managed_ports=list(MANAGED_PORTS),
        protocol=MANAGED_PROTOCOL,
        dry_run=DRY_RUN,
        min_cf_ipv4=MIN_CF_IPV4_COUNT,
        min_cf_ipv6=MIN_CF_IPV6_COUNT,
        require_both=REQUIRE_BOTH_FAMILIES,
        max_retries=MAX_RETRIES,
    )

    client = boto3.client("lightsail", region_name=region)

    try:
        # 1) Fetch + validate Cloudflare CIDRs (guardrails prevent empty/partial data from being used)
        cf_v4, cf_v6 = fetch_cloudflare_cidrs_txt(max_retries=MAX_RETRIES, base_sleep=RETRY_BASE_SECONDS)

        desired_v4 = set(cf_v4) | set(EXTRA_IPV4_CIDRS)
        desired_v6 = set(cf_v6) | set(EXTRA_IPV6_CIDRS)

        log(
            "cloudflare fetched+validated",
            cf_ipv4_count=len(cf_v4),
            cf_ipv6_count=len(cf_v6),
            extra_ipv4_count=len(EXTRA_IPV4_CIDRS),
            extra_ipv6_count=len(EXTRA_IPV6_CIDRS),
            desired_ipv4_count=len(desired_v4),
            desired_ipv6_count=len(desired_v6),
        )

        # 2) Read current Lightsail port states
        states = get_port_states(client, instance_name, max_retries=MAX_RETRIES, base_sleep=RETRY_BASE_SECONDS)

        changes: List[Dict[str, Any]] = []

        for port in MANAGED_PORTS:
            rule = find_exact_port_rule(states, port, MANAGED_PROTOCOL)
            current_v4 = set(rule["cidrs"]) if rule else set()
            current_v6 = set(rule["ipv6Cidrs"]) if rule else set()

            v4_to_add = sorted(desired_v4 - current_v4)
            v6_to_add = sorted(desired_v6 - current_v6)
            v4_to_remove = sorted(current_v4 - desired_v4)
            v6_to_remove = sorted(current_v6 - desired_v6)

            summary = {
                "port": port,
                "current_ipv4": len(current_v4),
                "current_ipv6": len(current_v6),
                "add_ipv4": len(v4_to_add),
                "add_ipv6": len(v6_to_add),
                "remove_ipv4": len(v4_to_remove),
                "remove_ipv6": len(v6_to_remove),
            }
            changes.append(summary)
            log("port diff computed", **summary)

            if DRY_RUN:
                continue

            # Add first, then remove (reduces risk of temporary block during list changes)
            if v4_to_add or v6_to_add:
                log("applying adds", port=port, add_ipv4_sample=v4_to_add[:5], add_ipv6_sample=v6_to_add[:5])
                open_cidrs(
                    client,
                    instance_name,
                    port,
                    MANAGED_PROTOCOL,
                    v4_to_add,
                    v6_to_add,
                    max_retries=MAX_RETRIES,
                    base_sleep=RETRY_BASE_SECONDS,
                )

            if v4_to_remove or v6_to_remove:
                log(
                    "applying removals",
                    port=port,
                    remove_ipv4_sample=v4_to_remove[:5],
                    remove_ipv6_sample=v6_to_remove[:5],
                )
                close_cidrs(
                    client,
                    instance_name,
                    port,
                    MANAGED_PROTOCOL,
                    v4_to_remove,
                    v6_to_remove,
                    max_retries=MAX_RETRIES,
                    base_sleep=RETRY_BASE_SECONDS,
                )

        elapsed_ms = int((time.time() - start) * 1000)
        return respond(
            True,
            stage="complete",
            elapsed_ms=elapsed_ms,
            instance=instance_name,
            region=region,
            dry_run=DRY_RUN,
            changes_summary=changes,
        )

    except Exception as e:
        # If Cloudflare fetch/validation fails, no changes are applied because we exit before open/close.
        log("sync failed", error=str(e))
        return respond(
            False,
            stage="exception",
            error=str(e),
            elapsed_ms=int((time.time() - start) * 1000),
        )

Lambda IAM Role Policy

As the Lambda function will be taking action on your Lightsail instances, you'll need to grant the Lambda's IAM execution role the proper access. Here's the policy I'm using to manage all my instances. You can tighten this up to a specific region or specific instance, if desired.

  { "Version": "2012-10-17", "Statement": [ { "Sid": "LightsailFirewallSyncForOneInstance", "Effect": "Allow", "Action": [ "lightsail:GetInstancePortStates", "lightsail:OpenInstancePublicPorts", "lightsail:CloseInstancePublicPorts" ], "Resource": "arn:aws:lightsail:*::Instance/*" } ] }

EventBridge Scheduler

After texting everything to this point, the only remaining thing to do is schedule the Lambda function to execute. Again, I have mine set to once per day but...you do you. The do this, just go to EventBridge Schedules in the AWS Console and use the Console GUI to select your Lambda function and enter the rate you want it to execute.

Now you never have to worry about keeping your Lightsail instance firewall synced with the Cloudflare IP set and you no longer have an excuse to just leave it fully open. Simple as that!