from ipaddress import ip_address
from logging import getLogger
from os import getenv
from socket import AF_INET, AF_INET6, SOCK_STREAM, gaierror, getaddrinfo
from typing import Literal
from pytz import timezone
__all__ = [
"ENVIRONMENT", "HEALTH_CHECK_TIMEOUT", "HEALTH_CHECK_FAILURE_LIMIT", "HEALTH_CHECK_LOG",
"CORS_ORIGINS", "CORS_HEADERS", "CORS_CREDENTIALS", "CORS_MAX_AGE",
"API_RESPONSE_TIMEOUT", "API_SENDER_THREADS", "API_RESPONSE_WRAPPED",
"API_RESPONSE_TIMING", "API_CACHE_CONTROL", "API_CACHE_HEADERS",
"API_ENABLE_DOCUMENTATION", "API_ENABLE_TOPOLOGY", "API_EXPOSE_ERROR_DETAILS",
"SWITCH_HOST", "SWITCH_PORT", "SWITCH_STATUS_ENABLED",
"SWITCH_STATUS_HOST", "SWITCH_STATUS_PORT", "SWITCH_MAX_MESSAGE",
"is_loopback_only",
"UPLOAD_FILES_PREFIX", "UPLOAD_FILES_PATH",
"CACHE_TIMEOUT", "CACHE_INFO_MESSAGE",
"ROUTER_OUTBOUND_QUEUE_SOFT_LIMIT", "ROUTER_OUTBOUND_QUEUE_HARD_LIMIT",
"ROUTER_OUTBOUND_JOIN_TIMEOUT", "ROUTER_OUTBOUND_STALL_SECONDS",
"HOST", "PORT", "SSL_KEY", "SSL_CERTIFICATE",
"GRPC_HOST", "GRPC_PORT",
"TEMPLATE_PATH", "ASSETS_PATH",
"MAX_BODY_SIZE", "MAX_MESSAGE_SIZE", "MAX_FILE_SIZE", "CHUNK_SIZE",
"ASGI_OPTIONS",
]
def getenv_bool(name: str, default: bool) -> bool:
"""Read an environment variable as a boolean, fallback to default if unset."""
val = getenv(name)
if val is None:
return default
return val.lower() in ('1', 'true', 'yes', 'on')
def _resolve_environment() -> Literal["development", "production"]:
"""Resolve ``ENVIRONMENT``, failing closed: only an explicit ``development`` is development.
Debug endpoints and error details default to enabled in development, so an unset, empty or
unrecognised value such as ``prod`` must not silently select it. It is treated as production.
"""
raw = getenv("ENVIRONMENT", "").strip()
logger = getLogger("edri.config")
if not raw:
if getenv("EDRI_ENVIRONMENT"):
logger.warning(
"EDRI_ENVIRONMENT is set but not read; the variable is named ENVIRONMENT. "
"ENVIRONMENT is unset, so running in production mode."
)
return "production"
value = raw.lower()
if value in ("development", "production"):
return value # type: ignore[return-value]
logger.warning(
"Unrecognised ENVIRONMENT value %r; expected 'development' or 'production'. "
"Treating it as production.", raw
)
return "production"
ENVIRONMENT: Literal["development", "production"] = _resolve_environment()
HEALTH_CHECK_TIMEOUT = int(getenv("EDRI_HEALTH_CHECK_TIMEOUT", 10))
HEALTH_CHECK_FAILURE_LIMIT = int(getenv("EDRI_HEALTH_CHECK_FAILURE_LIMIT", 3))
HEALTH_CHECK_LOG = getenv_bool("EDRI_HEALTH_CHECK_LOG", True)
CORS_ORIGINS = getenv("EDRI_CORS_ORIGINS")
CORS_HEADERS = getenv("EDRI_CORS_HEADERS")
CORS_CREDENTIALS = getenv_bool("EDRI_CORS_CREDENTIALS", False)
CORS_MAX_AGE = getenv("EDRI_CORS_MAX_AGE", None)
API_RESPONSE_TIMEOUT = int(getenv("EDRI_API_RESPONSE_TIMEOUT", 60))
API_SENDER_THREADS = int(getenv("EDRI_API_SENDER_THREADS", 32))
API_RESPONSE_WRAPPED = getenv_bool("EDRI_API_RESPONSE_WRAPPED", True)
API_RESPONSE_TIMING = getenv_bool("EDRI_API_RESPONSE_TIMING", ENVIRONMENT == 'development')
API_CACHE_CONTROL = getenv("EDRI_API_CACHE_CONTROL", "max-age=0, must-revalidate")
API_CACHE_HEADERS = getenv("EDRI_API_CACHE_HEADERS", CORS_HEADERS)
API_ENABLE_DOCUMENTATION = getenv_bool("EDRI_API_ENABLE_DOCUMENTATION", ENVIRONMENT == "development")
API_ENABLE_TOPOLOGY = getenv_bool("EDRI_API_ENABLE_TOPOLOGY", ENVIRONMENT == "development")
API_EXPOSE_ERROR_DETAILS = getenv_bool("EDRI_API_EXPOSE_ERROR_DETAILS", ENVIRONMENT == "development")
SWITCH_HOST = getenv("EDRI_SWITCH_HOST", "localhost")
SWITCH_PORT = int(getenv("EDRI_SWITCH_PORT", 8899))
SWITCH_STATUS_ENABLED = getenv_bool("EDRI_SWITCH_STATUS_ENABLED", True)
SWITCH_STATUS_HOST = getenv("EDRI_SWITCH_STATUS_HOST", "localhost")
SWITCH_STATUS_PORT = int(getenv("EDRI_SWITCH_STATUS_PORT", 8898))
# EDRI_SWITCH_SECRET and EDRI_SWITCH_SECRET_FILE are read in edri.protocol.keys.KeySet
# to avoid freezing their values at import time and allow tests to patch the environment.
# 64 MiB. The 4-byte length field can express 4 GiB, but a receiver that accepts
# what the field can express is not bounded in any useful sense. Bulk payloads
# travel through SharedMemoryPipe, not the Switch.
SWITCH_MAX_MESSAGE = int(getenv("EDRI_SWITCH_MAX_MESSAGE", 67108864))
[docs]
def is_loopback_only(host: str) -> bool:
"""True only when every address ``host`` resolves to is a loopback address.
Fails closed: a name that cannot be resolved, or one with a single routable
address among loopback ones, is not loopback. ``0.0.0.0`` and ``::`` mean
"every interface" and are never loopback.
"""
if host in ("0.0.0.0", "::"):
return False
try:
candidates = getaddrinfo(host, None, type=SOCK_STREAM)
except gaierror:
return False
if not candidates:
return False
for family, _, _, _, sockaddr in candidates:
if family not in (AF_INET, AF_INET6):
return False
try:
if not ip_address(sockaddr[0]).is_loopback:
return False
except ValueError:
return False
return True
UPLOAD_FILES_PREFIX = getenv("EDRI_UPLOAD_FILES_PREFIX", "edri_")
UPLOAD_FILES_PATH = getenv("EDRI_UPLOAD_FILES_PATH", "/tmp/upload")
CACHE_TIMEOUT = int(getenv("EDRI_CACHE_TIMEOUT", 30))
CACHE_INFO_MESSAGE = int(getenv("EDRI_CACHE_INFO_MESSAGE", 60))
# Per-connection outbound queue used by the Router to fan events out to
# subscribers without blocking its main loop on a slow consumer.
# Backlog at which a consumer is flagged WARNING (events keep queueing).
ROUTER_OUTBOUND_QUEUE_SOFT_LIMIT = int(getenv("EDRI_ROUTER_OUTBOUND_QUEUE_SOFT_LIMIT", 800))
# Backlog at which a consumer is considered stuck: disconnect + restart.
ROUTER_OUTBOUND_QUEUE_HARD_LIMIT = int(getenv("EDRI_ROUTER_OUTBOUND_QUEUE_HARD_LIMIT", 5000))
# Seconds to wait for a worker thread to drain and stop on unregister/shutdown.
ROUTER_OUTBOUND_JOIN_TIMEOUT = float(getenv("EDRI_ROUTER_OUTBOUND_JOIN_TIMEOUT", 5.0))
# Seconds a single outbound send may run before /edri/topology reports the
# consumer as stalled (a healthy send returns in milliseconds).
ROUTER_OUTBOUND_STALL_SECONDS = float(getenv("EDRI_ROUTER_OUTBOUND_STALL_SECONDS", 5.0))
HOST = getenv("EDRI_HOST", "localhost")
PORT = int(getenv("EDRI_PORT", 8080))
SSL_KEY = getenv("EDRI_SSL_KEY")
SSL_CERTIFICATE = getenv("EDRI_SSL_CERTIFICATE")
GRPC_HOST = getenv("EDRI_GRPC_HOST", "localhost")
GRPC_PORT = int(getenv("EDRI_GRPC_PORT", 50051))
TEMPLATE_PATH = getenv("EDRI_TEMPLATE_PATH", "templates")
ASSETS_PATH = getenv("EDRI_ASSETS_PATH", "assets")
MAX_BODY_SIZE = int(getenv("EDRI_MAX_BODY_SIZE", 4096 * 1024))
MAX_MESSAGE_SIZE = int(getenv("EDRI_MAX_MESSAGE_SIZE", MAX_BODY_SIZE))
MAX_FILE_SIZE = int(getenv("EDRI_MAX_FILE_SIZE", 100 * 1024 * 1024))
CHUNK_SIZE = int(getenv("EDRI_CHUNK_SIZE", 256 * 1024))
ASGI_OPTIONS = getenv("EDRI_ASGI_OPTIONS")