import ipaddress

import logging

import os

import platform

import re

import socket

import subprocess

import tempfile

import time

import warnings

from collections import defaultdict

from collections.abc import Mapping

from typing import Any, TypeVar

from localstack import constants

from localstack.constants import (

DEFAULT_BUCKET_MARKER_LOCAL,

DEFAULT_DEVELOP_PORT,

DEFAULT_VOLUME_DIR,

ENV_INTERNAL_TEST_COLLECT_METRIC,

ENV_INTERNAL_TEST_RUN,

ENV_INTERNAL_TEST_STORE_METRICS_IN_LOCALSTACK,

FALSE_STRINGS,

LOCALHOST,

LOCALHOST_IP,

LOCALSTACK_ROOT_FOLDER,

LOG_LEVELS,

TRACE_LOG_LEVELS,

TRUE_STRINGS,

)

T = TypeVar("T", str, int)

# keep track of start time, for performance debugging

load_start_time = time.time()

class Directories:

"""

Holds different directories available to localstack. Some directories are shared between the host and the

localstack container, some live only on the host and others in the container.

Attributes:

static_libs: container only; binaries and libraries statically packaged with the image

var_libs: shared; binaries and libraries+data computed at runtime: lazy-loaded binaries, ssl cert, ...

cache: shared; ephemeral data that has to persist across localstack runs and reboots

tmp: container only; ephemeral data that has to persist across localstack runs but not reboots

mounted_tmp: shared; same as above, but shared for persistence across different containers, tests, ...

functions: shared; volume to communicate between host<->lambda containers

data: shared; holds localstack state, pods, ...

config: host only; pre-defined configuration values, cached credentials, machine id, ...

init: shared; user-defined provisioning scripts executed in the container when it starts

logs: shared; log files produced by localstack

"""

static_libs: str

var_libs: str

cache: str

tmp: str

mounted_tmp: str

functions: str

data: str

config: str

init: str

logs: str

def __init__(

self,

static_libs: str,

var_libs: str,

cache: str,

tmp: str,

mounted_tmp: str,

functions: str,

data: str,

config: str,

init: str,

logs: str,

) -> None:

super().__init__()

self.static_libs = static_libs

self.var_libs = var_libs

self.cache = cache

self.tmp = tmp

self.mounted_tmp = mounted_tmp

self.functions = functions

self.data = data

self.config = config

self.init = init

self.logs = logs

@staticmethod

def defaults() -> "Directories":

"""Returns Localstack directory paths based on the localstack filesystem hierarchy."""

return Directories(

static_libs="/usr/lib/localstack",

var_libs=f"{DEFAULT_VOLUME_DIR}/lib",

cache=f"{DEFAULT_VOLUME_DIR}/cache",

tmp=os.path.join(tempfile.gettempdir(), "localstack"),

mounted_tmp=f"{DEFAULT_VOLUME_DIR}/tmp",

functions=f"{DEFAULT_VOLUME_DIR}/tmp", # FIXME: remove - this was misconceived

data=f"{DEFAULT_VOLUME_DIR}/state",

logs=f"{DEFAULT_VOLUME_DIR}/logs",

config="/etc/localstack/conf.d", # for future use

init="/etc/localstack/init",

)

@staticmethod

def for_container() -> "Directories":

"""

Returns Localstack directory paths as they are defined within the container. Everything shared and writable

lives in /var/lib/localstack or {tempfile.gettempdir()}/localstack.

:returns: Directories object

"""

defaults = Directories.defaults()

return Directories(

static_libs=defaults.static_libs,

var_libs=defaults.var_libs,

cache=defaults.cache,

tmp=defaults.tmp,

mounted_tmp=defaults.mounted_tmp,

functions=defaults.functions,

data=defaults.data if PERSISTENCE else os.path.join(defaults.tmp, "state"),

config=defaults.config,

logs=defaults.logs,

init=defaults.init,

)

@staticmethod

def for_host() -> "Directories":

"""Return directories used for running localstack in host mode. Note that these are *not* the directories

that are mounted into the container when the user starts localstack."""

root = os.environ.get("FILESYSTEM_ROOT") or os.path.join(

LOCALSTACK_ROOT_FOLDER, ".filesystem"

)

root = os.path.abspath(root)

defaults = Directories.for_container()

tmp = os.path.join(root, defaults.tmp.lstrip("/"))

data = os.path.join(root, defaults.data.lstrip("/"))

return Directories(

static_libs=os.path.join(root, defaults.static_libs.lstrip("/")),

var_libs=os.path.join(root, defaults.var_libs.lstrip("/")),

cache=os.path.join(root, defaults.cache.lstrip("/")),

tmp=tmp,

mounted_tmp=os.path.join(root, defaults.mounted_tmp.lstrip("/")),

functions=os.path.join(root, defaults.functions.lstrip("/")),

data=data if PERSISTENCE else os.path.join(tmp, "state"),

config=os.path.join(root, defaults.config.lstrip("/")),

init=os.path.join(root, defaults.init.lstrip("/")),

logs=os.path.join(root, defaults.logs.lstrip("/")),

)

@staticmethod

def for_cli() -> "Directories":

"""Returns directories used for when running localstack CLI commands from the host system. Unlike

``for_container``, these needs to be cross-platform. Ideally, this should not be needed at all,

because the localstack runtime and CLI do not share any control paths. There are a handful of

situations where directories or files may be created lazily for CLI commands. Some paths are

intentionally set to None to provoke errors if these paths are used from the CLI - which they

shouldn't. This is a symptom of not having a clear separation between CLI/runtime code, which will

be a future project."""

import tempfile

from localstack.utils import files

tmp_dir = os.path.join(tempfile.gettempdir(), "localstack-cli")

cache_dir = (files.get_user_cache_dir()).absolute() / "localstack-cli"

return Directories(

static_libs=None,

var_libs=None,

cache=str(cache_dir), # used by analytics metadata

tmp=tmp_dir,

mounted_tmp=tmp_dir,

functions=None,

data=os.path.join(tmp_dir, "state"), # used by localstack-pro config TODO: remove

logs=os.path.join(tmp_dir, "logs"), # used for container logs

config=None, # in the context of the CLI, config.CONFIG_DIR should be used

init=None,

)

def mkdirs(self):

for folder in [

self.static_libs,

self.var_libs,

self.cache,

self.tmp,

self.mounted_tmp,

self.functions,

self.data,

self.config,

self.init,

self.logs,

]:

if folder and not os.path.exists(folder):

try:

os.makedirs(folder)

except Exception:

# this can happen due to a race condition when starting

# multiple processes in parallel. Should be safe to ignore

pass

def __str__(self):

return str(self.__dict__)

def eval_log_type(env_var_name: str) -> str | bool:

"""Get the log type from environment variable"""

ls_log = os.environ.get(env_var_name, "").lower().strip()

return ls_log if ls_log in LOG_LEVELS else False

def parse_boolean_env(env_var_name: str) -> bool | None:

"""Parse the value of the given env variable and return True/False, or None if it is not a boolean value."""

value = os.environ.get(env_var_name, "").lower().strip()

if value in TRUE_STRINGS:

return True

if value in FALSE_STRINGS:

return False

return None

def parse_comma_separated_list(env_var_name: str) -> list[str]:

"""Parse a comma separated list from the given environment variable."""

return os.environ.get(env_var_name, "").strip().split(",")

def is_env_true(env_var_name: str) -> bool:

"""Whether the given environment variable has a truthy value."""

return os.environ.get(env_var_name, "").lower().strip() in TRUE_STRINGS

def is_env_not_false(env_var_name: str) -> bool:

"""Whether the given environment variable is empty or has a truthy value."""

return os.environ.get(env_var_name, "").lower().strip() not in FALSE_STRINGS

def load_environment(profiles: str = None, env=os.environ) -> list[str]:

"""Loads the environment variables from ~/.localstack/{profile}.env, for each profile listed in the profiles.

:param env: environment to load profile to. Defaults to `os.environ`

:param profiles: a comma separated list of profiles to load (defaults to "default")

:returns str: the list of the actually loaded profiles (might be the fallback)

"""

if not profiles:

profiles = "default"

profiles = profiles.split(",")

environment = {}

import dotenv

for profile in profiles:

profile = profile.strip()

path = os.path.join(CONFIG_DIR, f"{profile}.env")

if not os.path.exists(path):

continue

environment.update(dotenv.dotenv_values(path))

for k, v in environment.items():

# we do not want to override the environment

if k not in env and v is not None:

env[k] = v

return profiles

def is_persistence_enabled() -> bool:

return PERSISTENCE and dirs.data

def is_linux() -> bool:

return platform.system() == "Linux"

def is_macos() -> bool:

return platform.system() == "Darwin"

def is_windows() -> bool:

return platform.system().lower() == "windows"

def is_wsl() -> bool:

return platform.system().lower() == "linux" and os.environ.get("WSL_DISTRO_NAME") is not None

def ping(host):

"""Returns True if the host responds to a ping request"""

is_in_windows = is_windows()

ping_opts = "-n 1 -w 2000" if is_in_windows else "-c 1 -W 2"

args = f"ping {ping_opts} {host}"

return (

subprocess.call(

args, shell=not is_in_windows, stdout=subprocess.PIPE, stderr=subprocess.PIPE

)

== 0

)

def in_docker():

"""

Returns True if running in a docker container, else False

Ref. https://docs.docker.com/config/containers/runmetrics/#control-groups

"""

if OVERRIDE_IN_DOCKER is not None:

return OVERRIDE_IN_DOCKER

# check some marker files that we create in our Dockerfiles

for path in [

"/usr/lib/localstack/.community-version",

"/usr/lib/localstack/.pro-version",

"/tmp/localstack/.marker",

]:

if os.path.isfile(path):

return True

# details: https://github.com/localstack/localstack/pull/4352

if os.path.exists("/.dockerenv"):

return True

if os.path.exists("/run/.containerenv"):

return True

if not os.path.exists("/proc/1/cgroup"):

return False

try:

if any(

[

os.path.exists("/sys/fs/cgroup/memory/docker/"),

any(

"docker-" in file_names

for file_names in os.listdir("/sys/fs/cgroup/memory/system.slice")

),

os.path.exists("/sys/fs/cgroup/docker/"),

any(

"docker-" in file_names

for file_names in os.listdir("/sys/fs/cgroup/system.slice/")

),

]

):

return False

except Exception:

pass

with open("/proc/1/cgroup") as ifh:

content = ifh.read()

if "docker" in content or "buildkit" in content:

return True

os_hostname = socket.gethostname()

if os_hostname and os_hostname in content:

return True

# containerd does not set any specific file or config, but it does use

# io.containerd.snapshotter.v1.overlayfs as the overlay filesystem for `/`.

try:

with open("/proc/mounts") as infile:

for line in infile:

line = line.strip()

if not line:

continue

# skip comments

if line[0] == "#":

continue

# format (man 5 fstab)

# <spec> <mount point> <type> <options> <rest>...

parts = line.split()

if len(parts) < 4:

# badly formatted line

continue

mount_point = parts[1]

options = parts[3]

# only consider the root filesystem

if mount_point != "/":

continue

if "io.containerd" in options:

return True

except FileNotFoundError:

pass

return False

# whether the `in_docker` check should always return True or False

OVERRIDE_IN_DOCKER = parse_boolean_env("OVERRIDE_IN_DOCKER")

is_in_docker = in_docker()

is_in_linux = is_linux()

is_in_macos = is_macos()

is_in_windows = is_windows()

is_in_wsl = is_wsl()

default_ip = "0.0.0.0" if is_in_docker else "127.0.0.1"

# CLI specific: the configuration profile to load

CONFIG_PROFILE = os.environ.get("CONFIG_PROFILE", "").strip()

# CLI specific: host configuration directory

CONFIG_DIR = os.environ.get("CONFIG_DIR", os.path.expanduser("~/.localstack"))

# keep this on top to populate the environment

try:

# CLI specific: the actually loaded configuration profile

LOADED_PROFILES = load_environment(CONFIG_PROFILE)

except ImportError:

# dotenv may not be available in lambdas or other environments where config is loaded

LOADED_PROFILES = None

# loaded components name - default: all components are loaded and the first one is chosen

RUNTIME_COMPONENTS = os.environ.get("RUNTIME_COMPONENTS", "").strip()

# directory for persisting data (TODO: deprecated, simply use PERSISTENCE=1)

DATA_DIR = os.environ.get("DATA_DIR", "").strip()

# whether localstack should persist service state across localstack runs

PERSISTENCE = is_env_true("PERSISTENCE")

# the strategy for loading snapshots from disk when `PERSISTENCE=1` is used (on_startup, on_request, manual)

SNAPSHOT_LOAD_STRATEGY = os.environ.get("SNAPSHOT_LOAD_STRATEGY", "").upper()

# the strategy saving snapshots to disk when `PERSISTENCE=1` is used (on_shutdown, on_request, scheduled, manual)

SNAPSHOT_SAVE_STRATEGY = os.environ.get("SNAPSHOT_SAVE_STRATEGY", "").upper()

# the flush interval (in seconds) for persistence when the snapshot save strategy is set to "scheduled"

SNAPSHOT_FLUSH_INTERVAL = int(os.environ.get("SNAPSHOT_FLUSH_INTERVAL") or 15)

# whether to clear config.dirs.tmp on startup and shutdown

CLEAR_TMP_FOLDER = is_env_not_false("CLEAR_TMP_FOLDER")

# folder for temporary files and data

TMP_FOLDER = os.path.join(tempfile.gettempdir(), "localstack")

# this is exclusively for the CLI to configure the container mount into /var/lib/localstack

VOLUME_DIR = os.environ.get("LOCALSTACK_VOLUME_DIR", "").strip() or TMP_FOLDER

# fix for Mac OS, to be able to mount /var/folders in Docker

if TMP_FOLDER.startswith("/var/folders/") and os.path.exists(f"/private{TMP_FOLDER}"):

TMP_FOLDER = f"/private{TMP_FOLDER}"

# whether to enable verbose debug logging ("LOG" is used when using the CLI with LOCALSTACK_LOG instead of LS_LOG)

LS_LOG = eval_log_type("LS_LOG") or eval_log_type("LOG")

DEBUG = is_env_true("DEBUG") or LS_LOG in TRACE_LOG_LEVELS

# PUBLIC PREVIEW: 0 (default), 1 (preview)

# When enabled it triggers specialised workflows for the debugging.

LAMBDA_DEBUG_MODE = is_env_true("LAMBDA_DEBUG_MODE")

# path to the lambda debug mode configuration file.

LAMBDA_DEBUG_MODE_CONFIG_PATH = os.environ.get("LAMBDA_DEBUG_MODE_CONFIG_PATH")

# EXPERIMENTAL: allow setting custom log levels for individual loggers

LOG_LEVEL_OVERRIDES = os.environ.get("LOG_LEVEL_OVERRIDES", "")

# whether to enable debugpy

DEVELOP = is_env_true("DEVELOP")

# PORT FOR DEBUGGER

DEVELOP_PORT = int(os.environ.get("DEVELOP_PORT", "").strip() or DEFAULT_DEVELOP_PORT)

# whether to make debugpy wait for a debbuger client

WAIT_FOR_DEBUGGER = is_env_true("WAIT_FOR_DEBUGGER")

# whether to assume http or https for `get_protocol`

USE_SSL = is_env_true("USE_SSL")

# Whether to report internal failures as 500 or 501 errors.

FAIL_FAST = is_env_true("FAIL_FAST")

# whether to run in TF compatibility mode for TF integration tests

# (e.g., returning verbatim ports for ELB resources, rather than edge port 4566, etc.)

TF_COMPAT_MODE = is_env_true("TF_COMPAT_MODE")

# default encoding used to convert strings to byte arrays (mainly for Python 3 compatibility)

DEFAULT_ENCODING = "utf-8"

# path to local Docker UNIX domain socket

DOCKER_SOCK = os.environ.get("DOCKER_SOCK", "").strip() or "/var/run/docker.sock"

# additional flags to pass to "docker run" when starting the stack in Docker

DOCKER_FLAGS = os.environ.get("DOCKER_FLAGS", "").strip()

# command used to run Docker containers (e.g., set to "sudo docker" to run as sudo)

DOCKER_CMD = os.environ.get("DOCKER_CMD", "").strip() or "docker"

# use the command line docker client instead of the new sdk version, might get removed in the future

LEGACY_DOCKER_CLIENT = is_env_true("LEGACY_DOCKER_CLIENT")

# Docker image to use when starting up containers for port checks

PORTS_CHECK_DOCKER_IMAGE = os.environ.get("PORTS_CHECK_DOCKER_IMAGE", "").strip()

# global prefix to prepend to Docker image names (e.g., for using a custom registry mirror)

DOCKER_GLOBAL_IMAGE_PREFIX = os.environ.get("DOCKER_GLOBAL_IMAGE_PREFIX", "").strip()

def is_trace_logging_enabled():

if LS_LOG:

log_level = str(LS_LOG).upper()

return log_level.lower() in TRACE_LOG_LEVELS

return False

# set log levels immediately, but will be overwritten later by setup_logging

if DEBUG:

logging.getLogger("").setLevel(logging.DEBUG)

logging.getLogger("localstack").setLevel(logging.DEBUG)

LOG = logging.getLogger(__name__)

if is_trace_logging_enabled():

load_end_time = time.time()

LOG.debug(

"Initializing the configuration took %s ms", int((load_end_time - load_start_time) * 1000)

)

def is_ipv6_address(host: str) -> bool:

"""

Returns True if the given host is an IPv6 address.

"""

if not host:

return False

try:

ipaddress.IPv6Address(host)

return True

except ipaddress.AddressValueError:

return False

class HostAndPort:

"""

Definition of an address for a server to listen to.

Includes a `parse` method to convert from `str`, allowing for default fallbacks, as well as

some helper methods to help tests - particularly testing for equality and a hash function

so that `HostAndPort` instances can be used as keys to dictionaries.

"""

host: str

port: int

def __init__(self, host: str, port: int):

self.host = host

self.port = port

@classmethod

def parse(

cls,

input: str,

default_host: str,

default_port: int,

) -> "HostAndPort":

"""

Parse a `HostAndPort` from strings like:

- 0.0.0.0:4566 -> host=0.0.0.0, port=4566

- 0.0.0.0 -> host=0.0.0.0, port=`default_port`

- :4566 -> host=`default_host`, port=4566

- [::]:4566 -> host=[::], port=4566

- [::1] -> host=[::1], port=`default_port`

"""

host, port = default_host, default_port

# recognize IPv6 addresses (+ port)

if input.startswith("["):

ipv6_pattern = re.compile(r"^\[(?P<host>[^]]+)\](:(?P<port>\d+))?$")

match = ipv6_pattern.match(input)

if match:

host = match.group("host")

if not is_ipv6_address(host):

raise ValueError(

f"input looks like an IPv6 address (is enclosed in square brackets), but is not valid: {host}"

)

port_s = match.group("port")

if port_s:

port = cls._validate_port(port_s)

else:

raise ValueError(

f'input looks like an IPv6 address, but is invalid. Should be formatted "[ip]:port": {input}'

)

# recognize IPv4 address + port

elif ":" in input:

hostname, port_s = input.split(":", 1)

if hostname.strip():

host = hostname.strip()

port = cls._validate_port(port_s)

else:

if input.strip():

host = input.strip()

# validation

if port < 0 or port >= 2**16:

raise ValueError("port out of range")

return cls(host=host, port=port)

@classmethod

def _validate_port(cls, port_s: str) -> int:

try:

port = int(port_s)

except ValueError as e:

raise ValueError(f"specified port {port_s} not a number") from e

return port

def _get_unprivileged_port_range_start(self) -> int:

try:

with open("/proc/sys/net/ipv4/ip_unprivileged_port_start") as unprivileged_port_start:

port = unprivileged_port_start.read()

return int(port.strip())

except Exception:

return 1024

def is_unprivileged(self) -> bool:

return self.port >= self._get_unprivileged_port_range_start()

def host_and_port(self) -> str:

formatted_host = f"[{self.host}]" if is_ipv6_address(self.host) else self.host

return f"{formatted_host}:{self.port}" if self.port is not None else formatted_host

def __hash__(self) -> int:

return hash((self.host, self.port))

# easier tests

def __eq__(self, other: "str | HostAndPort") -> bool:

if isinstance(other, self.__class__):

return self.host == other.host and self.port == other.port

elif isinstance(other, str):

return str(self) == other

else:

raise TypeError(f"cannot compare {self.__class__} to {other.__class__}")

def __str__(self) -> str:

return self.host_and_port()

def __repr__(self) -> str:

return f"HostAndPort(host={self.host}, port={self.port})"

class UniqueHostAndPortList(list[HostAndPort]):

"""

Container type that ensures that ports added to the list are unique based

on these rules:

- :: "trumps" any other binding on the same port, including both IPv6 and IPv4

addresses. All other bindings for this port are removed, since :: already

covers all interfaces. For example, adding 127.0.0.1:4566, [::1]:4566,

and [::]:4566 would result in only [::]:4566 being preserved.

- 0.0.0.0 "trumps" any other binding on IPv4 addresses only. IPv6 addresses

are not removed.

- Identical identical hosts and ports are de-duped

"""

def __init__(self, iterable: list[HostAndPort] | None = None):

super().__init__(iterable or [])

self._ensure_unique()

def _ensure_unique(self):

"""

Ensure that all bindings on the same port are de-duped.

"""

if len(self) <= 1:

return

unique: list[HostAndPort] = []

# Build a dictionary of hosts by port

hosts_by_port: dict[int, list[str]] = defaultdict(list)

for item in self:

hosts_by_port[item.port].append(item.host)

# For any given port, dedupe the hosts

for port, hosts in hosts_by_port.items():

deduped_hosts = set(hosts)

# IPv6 all interfaces: this is the most general binding.

# Any others should be removed.

if "::" in deduped_hosts:

unique.append(HostAndPort(host="::", port=port))

continue

# IPv4 all interfaces: this is the next most general binding.

# Any others should be removed.

if "0.0.0.0" in deduped_hosts:

unique.append(HostAndPort(host="0.0.0.0", port=port))

continue

# All other bindings just need to be unique

unique.extend([HostAndPort(host=host, port=port) for host in deduped_hosts])

self.clear()

self.extend(unique)

def append(self, value: HostAndPort):

super().append(value)

self._ensure_unique()

def populate_edge_configuration(

environment: Mapping[str, str],

) -> tuple[HostAndPort, UniqueHostAndPortList]:

"""Populate the LocalStack edge configuration from environment variables."""

localstack_host_raw = environment.get("LOCALSTACK_HOST")

gateway_listen_raw = environment.get("GATEWAY_LISTEN")

# parse gateway listen from multiple components

if gateway_listen_raw is not None:

gateway_listen = []

for address in gateway_listen_raw.split(","):

gateway_listen.append(

HostAndPort.parse(

address.strip(),

default_host=default_ip,

default_port=constants.DEFAULT_PORT_EDGE,

)

)

else:

# use default if gateway listen is not defined

gateway_listen = [HostAndPort(host=default_ip, port=constants.DEFAULT_PORT_EDGE)]

# the actual value of the LOCALSTACK_HOST port now depends on what gateway listen actually listens to.

if localstack_host_raw is None:

localstack_host = HostAndPort(

host=constants.LOCALHOST_HOSTNAME, port=gateway_listen[0].port

)

else:

localstack_host = HostAndPort.parse(

localstack_host_raw,

default_host=constants.LOCALHOST_HOSTNAME,

default_port=gateway_listen[0].port,

)

assert gateway_listen is not None

assert localstack_host is not None

return (

localstack_host,

UniqueHostAndPortList(gateway_listen),

)

# How to access LocalStack

(

# -- Cosmetic

LOCALSTACK_HOST,

# -- Edge configuration

# Main configuration of the listen address of the hypercorn proxy. Of the form

# <ip_address>:<port>(,<ip_address>:port>)*

GATEWAY_LISTEN,

) = populate_edge_configuration(os.environ)

GATEWAY_WORKER_COUNT = int(os.environ.get("GATEWAY_WORKER_COUNT") or 1000)

# the gateway server that should be used (supported: hypercorn, twisted dev: werkzeug)

GATEWAY_SERVER = os.environ.get("GATEWAY_SERVER", "").strip() or "twisted"

# IP of the docker bridge used to enable access between containers

DOCKER_BRIDGE_IP = os.environ.get("DOCKER_BRIDGE_IP", "").strip()

# Default timeout for Docker API calls sent by the Docker SDK client, in seconds.

DOCKER_SDK_DEFAULT_TIMEOUT_SECONDS = int(os.environ.get("DOCKER_SDK_DEFAULT_TIMEOUT_SECONDS") or 60)

# Default number of retries to connect to the Docker API by the Docker SDK client.

DOCKER_SDK_DEFAULT_RETRIES = int(os.environ.get("DOCKER_SDK_DEFAULT_RETRIES") or 0)

# whether to enable API-based updates of configuration variables at runtime

ENABLE_CONFIG_UPDATES = is_env_true("ENABLE_CONFIG_UPDATES")

# CORS settings

DISABLE_CORS_HEADERS = is_env_true("DISABLE_CORS_HEADERS")

DISABLE_CORS_CHECKS = is_env_true("DISABLE_CORS_CHECKS")

DISABLE_CUSTOM_CORS_S3 = is_env_true("DISABLE_CUSTOM_CORS_S3")

DISABLE_CUSTOM_CORS_APIGATEWAY = is_env_true("DISABLE_CUSTOM_CORS_APIGATEWAY")

EXTRA_CORS_ALLOWED_HEADERS = os.environ.get("EXTRA_CORS_ALLOWED_HEADERS", "").strip()

EXTRA_CORS_EXPOSE_HEADERS = os.environ.get("EXTRA_CORS_EXPOSE_HEADERS", "").strip()

EXTRA_CORS_ALLOWED_ORIGINS = os.environ.get("EXTRA_CORS_ALLOWED_ORIGINS", "").strip()

DISABLE_PREFLIGHT_PROCESSING = is_env_true("DISABLE_PREFLIGHT_PROCESSING")

# whether to disable publishing events to the API

DISABLE_EVENTS = is_env_true("DISABLE_EVENTS")

DEBUG_ANALYTICS = is_env_true("DEBUG_ANALYTICS")

# whether to log fine-grained debugging information for the handler chain

DEBUG_HANDLER_CHAIN = is_env_true("DEBUG_HANDLER_CHAIN")

# whether to eagerly start services

EAGER_SERVICE_LOADING = is_env_true("EAGER_SERVICE_LOADING")

# whether to selectively load services in SERVICES

STRICT_SERVICE_LOADING = is_env_not_false("STRICT_SERVICE_LOADING")

# Whether to skip downloading additional infrastructure components (e.g., custom Elasticsearch versions)

SKIP_INFRA_DOWNLOADS = os.environ.get("SKIP_INFRA_DOWNLOADS", "").strip()

# Whether to skip downloading our signed SSL cert.

SKIP_SSL_CERT_DOWNLOAD = is_env_true("SKIP_SSL_CERT_DOWNLOAD")

# Absolute path to a custom certificate (pem file)

CUSTOM_SSL_CERT_PATH = os.environ.get("CUSTOM_SSL_CERT_PATH", "").strip()

# Whether delete the cached signed SSL certificate at startup

REMOVE_SSL_CERT = is_env_true("REMOVE_SSL_CERT")

# Allow non-standard AWS regions

ALLOW_NONSTANDARD_REGIONS = is_env_true("ALLOW_NONSTANDARD_REGIONS")

if ALLOW_NONSTANDARD_REGIONS:

os.environ["MOTO_ALLOW_NONEXISTENT_REGION"] = "true"

# name of the main Docker container

MAIN_CONTAINER_NAME = os.environ.get("MAIN_CONTAINER_NAME", "").strip() or "localstack-main"

# the latest commit id of the repository when the docker image was created

LOCALSTACK_BUILD_GIT_HASH = os.environ.get("LOCALSTACK_BUILD_GIT_HASH", "").strip() or None

# the date on which the docker image was created

LOCALSTACK_BUILD_DATE = os.environ.get("LOCALSTACK_BUILD_DATE", "").strip() or None

# Equivalent to HTTP_PROXY, but only applicable for external connections

OUTBOUND_HTTP_PROXY = os.environ.get("OUTBOUND_HTTP_PROXY", "")

# Equivalent to HTTPS_PROXY, but only applicable for external connections

OUTBOUND_HTTPS_PROXY = os.environ.get("OUTBOUND_HTTPS_PROXY", "")

# Feature flag to enable validation of internal endpoint responses in the handler chain. For test use only.

OPENAPI_VALIDATE_RESPONSE = is_env_true("OPENAPI_VALIDATE_RESPONSE")

# Flag to enable the validation of the requests made to the LocalStack internal endpoints. Active by default.

OPENAPI_VALIDATE_REQUEST = is_env_true("OPENAPI_VALIDATE_REQUEST")

# environment variable to determine whether to include stack traces in http responses

INCLUDE_STACK_TRACES_IN_HTTP_RESPONSE = is_env_true("INCLUDE_STACK_TRACES_IN_HTTP_RESPONSE")

# whether to skip waiting for the infrastructure to shut down, or exit immediately

FORCE_SHUTDOWN = is_env_not_false("FORCE_SHUTDOWN")

# set variables no_proxy, i.e., run internal service calls directly

no_proxy = ",".join([constants.LOCALHOST_HOSTNAME, LOCALHOST, LOCALHOST_IP, "[::1]"])

if os.environ.get("no_proxy"):

os.environ["no_proxy"] += "," + no_proxy

elif os.environ.get("NO_PROXY"):

os.environ["NO_PROXY"] += "," + no_proxy

else:

os.environ["no_proxy"] = no_proxy

# additional CLI commands, can be set by plugins

CLI_COMMANDS = {}

# determine IP of Docker bridge

if not DOCKER_BRIDGE_IP:

DOCKER_BRIDGE_IP = "172.17.0.1"

if is_in_docker:

candidates = (DOCKER_BRIDGE_IP, "172.18.0.1")

for ip in candidates:

# TODO: remove from here - should not perform I/O operations in top-level config.py

if ping(ip):

DOCKER_BRIDGE_IP = ip

break

# AWS account used to store internal resources such as Lambda archives or internal SQS queues.

# It should not be modified by the user, or visible to him, except as through a presigned url with the

# get-function call.

INTERNAL_RESOURCE_ACCOUNT = os.environ.get("INTERNAL_RESOURCE_ACCOUNT") or "949334387222"

# -----

# SERVICE-SPECIFIC CONFIGS BELOW

# -----

# port ranges for external service instances (f.e. elasticsearch clusters, opensearch clusters,...)

EXTERNAL_SERVICE_PORTS_START = int(

os.environ.get("EXTERNAL_SERVICE_PORTS_START")

or os.environ.get("SERVICE_INSTANCES_PORTS_START")

or 4510

)

EXTERNAL_SERVICE_PORTS_END = int(

os.environ.get("EXTERNAL_SERVICE_PORTS_END")

or os.environ.get("SERVICE_INSTANCES_PORTS_END")

or (EXTERNAL_SERVICE_PORTS_START + 50)

)

# The default container runtime to use

CONTAINER_RUNTIME = os.environ.get("CONTAINER_RUNTIME", "").strip() or "docker"

# PUBLIC v1: -Xmx512M (example) Currently not supported in new provider but possible via custom entrypoint.

# Allow passing custom JVM options to Java Lambdas executed in Docker.

LAMBDA_JAVA_OPTS = os.environ.get("LAMBDA_JAVA_OPTS", "").strip()

# limit in which to kinesis-mock will start throwing exceptions

KINESIS_SHARD_LIMIT = os.environ.get("KINESIS_SHARD_LIMIT", "").strip() or "100"

KINESIS_PERSISTENCE = is_env_not_false("KINESIS_PERSISTENCE")

# limit in which to kinesis-mock will start throwing exceptions

KINESIS_ON_DEMAND_STREAM_COUNT_LIMIT = (

os.environ.get("KINESIS_ON_DEMAND_STREAM_COUNT_LIMIT", "").strip() or "10"

)

# delay in kinesis-mock response when making changes to streams

KINESIS_LATENCY = os.environ.get("KINESIS_LATENCY", "").strip() or "500"

# Delay between data persistence (in seconds)

KINESIS_MOCK_PERSIST_INTERVAL = os.environ.get("KINESIS_MOCK_PERSIST_INTERVAL", "").strip() or "5s"

# Kinesis mock log level override when inconsistent with LS_LOG (e.g., when LS_LOG=debug)

KINESIS_MOCK_LOG_LEVEL = os.environ.get("KINESIS_MOCK_LOG_LEVEL", "").strip()

# randomly inject faults to Kinesis

KINESIS_ERROR_PROBABILITY = float(os.environ.get("KINESIS_ERROR_PROBABILITY", "").strip() or 0.0)

# SEMI-PUBLIC: "node" (default); not actively communicated

# Select whether to use the node or scala build when running Kinesis Mock

KINESIS_MOCK_PROVIDER_ENGINE = os.environ.get("KINESIS_MOCK_PROVIDER_ENGINE", "").strip() or "node"

# set the maximum Java heap size corresponding to the '-Xmx<size>' flag

KINESIS_MOCK_MAXIMUM_HEAP_SIZE = (

os.environ.get("KINESIS_MOCK_MAXIMUM_HEAP_SIZE", "").strip() or "512m"

)

# set the initial Java heap size corresponding to the '-Xms<size>' flag

KINESIS_MOCK_INITIAL_HEAP_SIZE = (

os.environ.get("KINESIS_MOCK_INITIAL_HEAP_SIZE", "").strip() or "256m"

)

# randomly inject faults to DynamoDB

DYNAMODB_ERROR_PROBABILITY = float(os.environ.get("DYNAMODB_ERROR_PROBABILITY", "").strip() or 0.0)

DYNAMODB_READ_ERROR_PROBABILITY = float(

os.environ.get("DYNAMODB_READ_ERROR_PROBABILITY", "").strip() or 0.0

)

DYNAMODB_WRITE_ERROR_PROBABILITY = float(

os.environ.get("DYNAMODB_WRITE_ERROR_PROBABILITY", "").strip() or 0.0

)

# JAVA EE heap size for dynamodb

DYNAMODB_HEAP_SIZE = os.environ.get("DYNAMODB_HEAP_SIZE", "").strip() or "256m"

# single DB instance across multiple credentials are regions

DYNAMODB_SHARE_DB = int(os.environ.get("DYNAMODB_SHARE_DB") or 0)

# the port on which to expose dynamodblocal

DYNAMODB_LOCAL_PORT = int(os.environ.get("DYNAMODB_LOCAL_PORT") or 0)

# Enables the automatic removal of stale KV pais based on TTL

DYNAMODB_REMOVE_EXPIRED_ITEMS = is_env_true("DYNAMODB_REMOVE_EXPIRED_ITEMS")

# Used to toggle PurgeInProgress exceptions when calling purge within 60 seconds

SQS_DELAY_PURGE_RETRY = is_env_true("SQS_DELAY_PURGE_RETRY")

# Used to toggle QueueDeletedRecently errors when re-creating a queue within 60 seconds of deleting it

SQS_DELAY_RECENTLY_DELETED = is_env_true("SQS_DELAY_RECENTLY_DELETED")

# Used to toggle MessageRetentionPeriod functionality in SQS queues

SQS_ENABLE_MESSAGE_RETENTION_PERIOD = is_env_true("SQS_ENABLE_MESSAGE_RETENTION_PERIOD")

# Strategy used when creating SQS queue urls. can be "off", "standard" (default), "domain", "path", or "dynamic"

SQS_ENDPOINT_STRATEGY = os.environ.get("SQS_ENDPOINT_STRATEGY", "") or "standard"

# Disable the check for MaxNumberOfMessage in SQS ReceiveMessage

SQS_DISABLE_MAX_NUMBER_OF_MESSAGE_LIMIT = is_env_true("SQS_DISABLE_MAX_NUMBER_OF_MESSAGE_LIMIT")

# Disable cloudwatch metrics for SQS

SQS_DISABLE_CLOUDWATCH_METRICS = is_env_true("SQS_DISABLE_CLOUDWATCH_METRICS")

# Interval for reporting "approximate" metrics to cloudwatch, default is 60 seconds

SQS_CLOUDWATCH_METRICS_REPORT_INTERVAL = int(

os.environ.get("SQS_CLOUDWATCH_METRICS_REPORT_INTERVAL") or 60

)

# PUBLIC: Endpoint host under which LocalStack APIs are accessible from Lambda Docker containers.

HOSTNAME_FROM_LAMBDA = os.environ.get("HOSTNAME_FROM_LAMBDA", "").strip()

# PUBLIC: hot-reload (default v2), __local__ (default v1)

# Magic S3 bucket name for Hot Reloading. The S3Key points to the source code on the local file system.

BUCKET_MARKER_LOCAL = (

os.environ.get("BUCKET_MARKER_LOCAL", "").strip() or DEFAULT_BUCKET_MARKER_LOCAL

)

# PUBLIC: Opt-out to inject the environment variable AWS_ENDPOINT_URL for automatic configuration of AWS SDKs:

# https://docs.aws.amazon.com/sdkref/latest/guide/feature-ss-endpoints.html

LAMBDA_DISABLE_AWS_ENDPOINT_URL = is_env_true("LAMBDA_DISABLE_AWS_ENDPOINT_URL")

# PUBLIC: bridge (Docker default)

# Docker network driver for the Lambda and ECS containers. https://docs.docker.com/network/

LAMBDA_DOCKER_NETWORK = os.environ.get("LAMBDA_DOCKER_NETWORK", "").strip()

# PUBLIC v1: LocalStack DNS (default)

# Custom DNS server for the container running your lambda function.

LAMBDA_DOCKER_DNS = os.environ.get("LAMBDA_DOCKER_DNS", "").strip()

# PUBLIC: -e KEY=VALUE -v host:container

# Additional flags passed to Docker run|create commands.

LAMBDA_DOCKER_FLAGS = os.environ.get("LAMBDA_DOCKER_FLAGS", "").strip()

# PUBLIC: 0 (default)

# Enable this flag to run cross-platform compatible lambda functions natively (i.e., Docker selects architecture) and

# ignore the AWS architectures (i.e., x86_64, arm64) configured for the lambda function.

LAMBDA_IGNORE_ARCHITECTURE = is_env_true("LAMBDA_IGNORE_ARCHITECTURE")