"""Environment-owned server configuration with fail-closed validation.""" from __future__ import annotations import base64 from dataclasses import dataclass, field from email.headerregistry import Address from ipaddress import ip_network import json import os from pathlib import Path from types import MappingProxyType from typing import Mapping, Optional from urllib.parse import urlparse from licensing_shared.constants import validate_identifier from .constants import ( FINGERPRINT_PEPPER_MINIMUM_BYTES, NOTIFICATION_PEPPER_MINIMUM_BYTES, RATE_LIMIT_MAXIMUM_TRUSTED_PROXY_NETWORKS, SERIAL_PEPPER_MINIMUM_BYTES, ) MAXIMUM_SECRET_FILE_BYTES = 64 * 1024 DEFAULT_SMTP_PORT = 465 SMTP_TLS_MODES = frozenset(("implicit", "starttls")) def _environment_text(environment: Mapping[str, str], key: str) -> Optional[str]: direct = environment.get(key) file_value = environment.get(f"{key}_FILE") if direct and file_value: raise ValueError(f"configure only one of {key} and {key}_FILE") if file_value: if not isinstance(file_value, str) or not file_value.strip(): raise ValueError(f"{key}_FILE must identify a secret file") path = Path(file_value.strip()) if not path.is_file() or path.is_symlink(): raise ValueError(f"{key}_FILE must identify a regular non-symlink file") if path.stat().st_size > MAXIMUM_SECRET_FILE_BYTES: raise ValueError(f"{key}_FILE is too large") try: return path.read_text(encoding="utf-8").strip() except (OSError, UnicodeError) as exc: raise ValueError(f"{key}_FILE could not be read as UTF-8") from exc if direct is None: return None if not isinstance(direct, str): raise TypeError(f"{key} must be a string") return direct.strip() def _required_text(environment: Mapping[str, str], key: str) -> str: value = _environment_text(environment, key) if not value: raise ValueError(f"required environment variable is missing: {key}") return value def _secret_bytes(environment: Mapping[str, str], key: str, minimum_bytes: int) -> bytes: encoded = _required_text(environment, key) try: value = base64.b64decode(encoded, validate=True) except (TypeError, ValueError) as exc: raise ValueError(f"{key} must contain canonical Base64") from exc if len(value) < minimum_bytes: raise ValueError(f"{key} must decode to at least {minimum_bytes} bytes") return value def _email_address(value: object, field_name: str) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty email address") normalized = value.strip() if len(normalized) > 256 or "\r" in normalized or "\n" in normalized: raise ValueError(f"{field_name} is invalid") try: address = Address(addr_spec=normalized) except (TypeError, ValueError) as exc: raise ValueError(f"{field_name} is invalid") from exc if not address.username or not address.domain or address.addr_spec != normalized: raise ValueError(f"{field_name} must be a canonical address without a display name") return address.addr_spec @dataclass(frozen=True) class ServerSettings: database_url: str environment_name: str signing_key_id: str signing_private_key_path: Path serial_pepper: bytes fingerprint_pepper: bytes oidc_issuer: str oidc_audience: str oidc_jwks_url: str allow_development_signer: bool = False stripe_webhook_secret: Optional[str] = None stripe_price_sku_map: Mapping[str, str] = field(default_factory=dict) oidc_authorization_endpoint: Optional[str] = None oidc_token_endpoint: Optional[str] = None oidc_native_client_id: Optional[str] = None oidc_native_scopes: tuple[str, ...] = ( "openid", "profile", "email", "offline_access", ) allow_production_file_signer: bool = False stripe_commerce_enabled: bool = False stripe_secret_key: Optional[str] = None stripe_checkout_success_url: Optional[str] = None stripe_checkout_cancel_url: Optional[str] = None stripe_billing_portal_return_url: Optional[str] = None email_notifications_enabled: bool = False smtp_host: Optional[str] = None smtp_port: int = DEFAULT_SMTP_PORT smtp_tls_mode: str = "implicit" smtp_username: Optional[str] = None smtp_password: Optional[str] = None email_from_address: Optional[str] = None account_portal_url: Optional[str] = None trusted_proxy_ips: tuple[str, ...] = () def __post_init__(self) -> None: for field_name in ( "database_url", "environment_name", "oidc_issuer", "oidc_audience", "oidc_jwks_url", ): value = getattr(self, field_name) if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") object.__setattr__( self, "signing_key_id", validate_identifier(self.signing_key_id, "signing_key_id"), ) if not isinstance(self.signing_private_key_path, Path): raise TypeError("signing_private_key_path must be a Path") if len(self.serial_pepper) < SERIAL_PEPPER_MINIMUM_BYTES: raise ValueError("serial_pepper is too short") if len(self.fingerprint_pepper) < FINGERPRINT_PEPPER_MINIMUM_BYTES: raise ValueError("fingerprint_pepper is too short") if not isinstance(self.allow_development_signer, bool): raise TypeError("allow_development_signer must be a Boolean") if not isinstance(self.allow_production_file_signer, bool): raise TypeError("allow_production_file_signer must be a Boolean") if not isinstance(self.stripe_commerce_enabled, bool): raise TypeError("stripe_commerce_enabled must be a Boolean") if self.stripe_webhook_secret is not None: if ( not isinstance(self.stripe_webhook_secret, str) or not self.stripe_webhook_secret.strip() ): raise ValueError("stripe_webhook_secret must be a non-empty string or None") object.__setattr__( self, "stripe_webhook_secret", self.stripe_webhook_secret.strip(), ) if not isinstance(self.stripe_price_sku_map, Mapping): raise TypeError("stripe_price_sku_map must be a mapping") normalized_prices: dict[str, str] = {} for price_id, sku_id in self.stripe_price_sku_map.items(): if not isinstance(price_id, str) or not price_id.strip(): raise ValueError("Stripe price IDs must be non-empty strings") normalized_prices[price_id.strip()] = validate_identifier( sku_id, "Stripe price SKU ID", ) if len(set(normalized_prices.values())) != len(normalized_prices): raise ValueError("each licensing SKU must map to exactly one Stripe price") object.__setattr__( self, "stripe_price_sku_map", MappingProxyType(normalized_prices), ) if self.stripe_secret_key is not None: if not isinstance(self.stripe_secret_key, str) or not self.stripe_secret_key.strip(): raise ValueError("stripe_secret_key must be a non-empty string or None") object.__setattr__(self, "stripe_secret_key", self.stripe_secret_key.strip()) for field_name in ( "stripe_checkout_success_url", "stripe_checkout_cancel_url", "stripe_billing_portal_return_url", ): value = getattr(self, field_name) if value is None: continue if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string or None") normalized = value.strip() parsed = urlparse(normalized) if ( parsed.scheme != "https" or not parsed.netloc or parsed.username is not None or parsed.password is not None or parsed.fragment ): raise ValueError( f"{field_name} must be an HTTPS URL without credentials or a fragment" ) object.__setattr__(self, field_name, normalized) if self.stripe_commerce_enabled: missing_commerce_fields = tuple( field_name for field_name in ( "stripe_webhook_secret", "stripe_secret_key", "stripe_checkout_success_url", "stripe_checkout_cancel_url", "stripe_billing_portal_return_url", ) if getattr(self, field_name) is None ) if missing_commerce_fields or not normalized_prices: raise ValueError( "enabled Stripe commerce requires webhook/API secrets, Checkout and " "portal URLs, and at least one price-to-SKU mapping" ) if not isinstance(self.email_notifications_enabled, bool): raise TypeError("email_notifications_enabled must be a Boolean") if self.smtp_host is not None: if not isinstance(self.smtp_host, str) or not self.smtp_host.strip(): raise ValueError("smtp_host must be a non-empty string or None") normalized_host = self.smtp_host.strip() if ( len(normalized_host) > 253 or any(character.isspace() for character in normalized_host) or any(character in normalized_host for character in ("/", "@", "?", "#")) ): raise ValueError("smtp_host is invalid") object.__setattr__(self, "smtp_host", normalized_host) if isinstance(self.smtp_port, bool) or not isinstance(self.smtp_port, int): raise TypeError("smtp_port must be an integer") if self.smtp_port < 1 or self.smtp_port > 65535: raise ValueError("smtp_port must be between one and 65535") if not isinstance(self.smtp_tls_mode, str): raise TypeError("smtp_tls_mode must be a string") normalized_tls_mode = self.smtp_tls_mode.strip().lower() if normalized_tls_mode not in SMTP_TLS_MODES: raise ValueError("smtp_tls_mode must be implicit or starttls") object.__setattr__(self, "smtp_tls_mode", normalized_tls_mode) for field_name in ("smtp_username", "smtp_password"): value = getattr(self, field_name) if value is None: continue if not isinstance(value, str) or not value.strip() or len(value) > 4096: raise ValueError(f"{field_name} must be a bounded non-empty string or None") object.__setattr__(self, field_name, value.strip()) if (self.smtp_username is None) != (self.smtp_password is None): raise ValueError("smtp_username and smtp_password must be configured together") if self.email_from_address is not None: object.__setattr__( self, "email_from_address", _email_address(self.email_from_address, "email_from_address"), ) if self.account_portal_url is not None: if not isinstance(self.account_portal_url, str): raise TypeError("account_portal_url must be a string or None") normalized_account_url = self.account_portal_url.strip() parsed_account_url = urlparse(normalized_account_url) if ( parsed_account_url.scheme != "https" or not parsed_account_url.netloc or parsed_account_url.username is not None or parsed_account_url.password is not None or parsed_account_url.fragment ): raise ValueError( "account_portal_url must be an HTTPS URL without credentials or fragment" ) object.__setattr__(self, "account_portal_url", normalized_account_url) if not isinstance(self.trusted_proxy_ips, tuple): raise TypeError("trusted_proxy_ips must be a tuple") if len(self.trusted_proxy_ips) > RATE_LIMIT_MAXIMUM_TRUSTED_PROXY_NETWORKS: raise ValueError("trusted_proxy_ips contains too many networks") normalized_proxy_ips: list[str] = [] for value in self.trusted_proxy_ips: if not isinstance(value, str) or not value.strip(): raise ValueError("trusted_proxy_ips values must be non-empty strings") try: normalized_proxy_ips.append(str(ip_network(value.strip(), strict=False))) except ValueError as exc: raise ValueError("trusted_proxy_ips contains an invalid IP or network") from exc if len(set(normalized_proxy_ips)) != len(normalized_proxy_ips): raise ValueError("trusted_proxy_ips must not contain duplicates") object.__setattr__(self, "trusted_proxy_ips", tuple(normalized_proxy_ips)) if self.email_notifications_enabled and any( value is None for value in ( self.smtp_host, self.smtp_username, self.smtp_password, self.email_from_address, self.account_portal_url, ) ): raise ValueError( "enabled email notifications require SMTP host, username/password, " "from address, and account portal URL" ) native_values = ( self.oidc_authorization_endpoint, self.oidc_token_endpoint, self.oidc_native_client_id, ) if any(value is not None for value in native_values) and not all( isinstance(value, str) and value.strip() for value in native_values ): raise ValueError( "OIDC native authorization endpoint, token endpoint, and client ID " "must be configured together" ) for field_name in ("oidc_authorization_endpoint", "oidc_token_endpoint"): value = getattr(self, field_name) if value is None: continue normalized = value.strip() parsed = urlparse(normalized) if ( parsed.scheme != "https" or not parsed.netloc or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment ): raise ValueError(f"{field_name} must be an HTTPS endpoint without credentials") object.__setattr__(self, field_name, normalized) if self.oidc_native_client_id is not None: object.__setattr__( self, "oidc_native_client_id", self.oidc_native_client_id.strip(), ) if not isinstance(self.oidc_native_scopes, tuple): raise TypeError("oidc_native_scopes must be a tuple") if any(not isinstance(scope, str) for scope in self.oidc_native_scopes): raise TypeError("oidc_native_scopes values must be strings") scopes = tuple(dict.fromkeys(scope.strip() for scope in self.oidc_native_scopes)) if not scopes or any(not scope or len(scope) > 128 for scope in scopes): raise ValueError("oidc_native_scopes must contain bounded non-empty values") if "openid" not in scopes: raise ValueError("oidc_native_scopes must include openid") object.__setattr__(self, "oidc_native_scopes", scopes) if self.environment_name.lower() == "production": if self.allow_development_signer: raise ValueError("the development signer cannot be enabled in production") if not urlparse(self.database_url).scheme.startswith("postgresql"): raise ValueError("production licensing requires PostgreSQL") for field_name in ("oidc_issuer", "oidc_jwks_url"): parsed = urlparse(getattr(self, field_name)) if ( parsed.scheme != "https" or not parsed.netloc or parsed.username is not None or parsed.password is not None or parsed.fragment ): raise ValueError( f"production {field_name} must be an HTTPS URL without credentials" ) if ( self.allow_production_file_signer and not self.signing_private_key_path.is_absolute() ): raise ValueError("production signing key path must be absolute") @classmethod def from_environment( cls, environment: Optional[Mapping[str, str]] = None, ) -> "ServerSettings": source = os.environ if environment is None else environment if not isinstance(source, Mapping): raise TypeError("environment must be a mapping or None") allow_development_signer = source.get("LICENSING_ALLOW_DEVELOPMENT_SIGNER", "") allow_production_file_signer = source.get( "LICENSING_ALLOW_PRODUCTION_FILE_SIGNER", "", ) raw_price_map = str(source.get("LICENSING_STRIPE_PRICE_SKU_MAP_JSON") or "{}").strip() try: price_map = json.loads(raw_price_map) except json.JSONDecodeError as exc: raise ValueError("LICENSING_STRIPE_PRICE_SKU_MAP_JSON must be valid JSON") from exc if not isinstance(price_map, dict): raise ValueError("LICENSING_STRIPE_PRICE_SKU_MAP_JSON must be a JSON object") webhook_secret = _environment_text(source, "LICENSING_STRIPE_WEBHOOK_SECRET") or "" stripe_secret_key = _environment_text(source, "LICENSING_STRIPE_SECRET_KEY") or "" stripe_commerce_enabled = str( source.get("LICENSING_STRIPE_COMMERCE_ENABLED") or "" ).strip().lower() in ("1", "true", "yes") email_notifications_enabled = str( source.get("LICENSING_EMAIL_NOTIFICATIONS_ENABLED") or "" ).strip().lower() in ("1", "true", "yes") smtp_username = _environment_text(source, "LICENSING_SMTP_USERNAME") or "" smtp_password = _environment_text(source, "LICENSING_SMTP_PASSWORD") or "" raw_smtp_port = str( source.get("LICENSING_SMTP_PORT") or DEFAULT_SMTP_PORT ).strip() try: smtp_port = int(raw_smtp_port) except ValueError as exc: raise ValueError("LICENSING_SMTP_PORT must be an integer") from exc authorization_endpoint = str( source.get("LICENSING_OIDC_AUTHORIZATION_ENDPOINT") or "" ).strip() token_endpoint = str(source.get("LICENSING_OIDC_TOKEN_ENDPOINT") or "").strip() native_client_id = str(source.get("LICENSING_OIDC_NATIVE_CLIENT_ID") or "").strip() native_scopes = tuple( scope for scope in str( source.get("LICENSING_OIDC_NATIVE_SCOPES") or "openid profile email offline_access" ).split() if scope ) trusted_proxy_ips = tuple( value.strip() for value in str(source.get("LICENSING_TRUSTED_PROXY_IPS") or "").split(",") if value.strip() ) return cls( database_url=_required_text(source, "LICENSING_DATABASE_URL"), environment_name=_required_text(source, "LICENSING_ENVIRONMENT"), signing_key_id=_required_text(source, "LICENSING_SIGNING_KEY_ID"), signing_private_key_path=Path( _required_text(source, "LICENSING_SIGNING_PRIVATE_KEY_PATH") ), serial_pepper=_secret_bytes( source, "LICENSING_SERIAL_PEPPER_BASE64", SERIAL_PEPPER_MINIMUM_BYTES, ), fingerprint_pepper=_secret_bytes( source, "LICENSING_FINGERPRINT_PEPPER_BASE64", FINGERPRINT_PEPPER_MINIMUM_BYTES, ), oidc_issuer=_required_text(source, "LICENSING_OIDC_ISSUER"), oidc_audience=_required_text(source, "LICENSING_OIDC_AUDIENCE"), oidc_jwks_url=_required_text(source, "LICENSING_OIDC_JWKS_URL"), allow_development_signer=allow_development_signer.strip().lower() in ("1", "true", "yes"), stripe_webhook_secret=webhook_secret or None, stripe_price_sku_map=price_map, oidc_authorization_endpoint=authorization_endpoint or None, oidc_token_endpoint=token_endpoint or None, oidc_native_client_id=native_client_id or None, oidc_native_scopes=native_scopes, allow_production_file_signer=allow_production_file_signer.strip().lower() in ("1", "true", "yes"), stripe_commerce_enabled=stripe_commerce_enabled, stripe_secret_key=stripe_secret_key or None, stripe_checkout_success_url=( str(source.get("LICENSING_STRIPE_CHECKOUT_SUCCESS_URL") or "").strip() or None ), stripe_checkout_cancel_url=( str(source.get("LICENSING_STRIPE_CHECKOUT_CANCEL_URL") or "").strip() or None ), stripe_billing_portal_return_url=( str(source.get("LICENSING_STRIPE_BILLING_PORTAL_RETURN_URL") or "").strip() or None ), email_notifications_enabled=email_notifications_enabled, smtp_host=str(source.get("LICENSING_SMTP_HOST") or "").strip() or None, smtp_port=smtp_port, smtp_tls_mode=( str(source.get("LICENSING_SMTP_TLS_MODE") or "implicit").strip() ), smtp_username=smtp_username or None, smtp_password=smtp_password or None, email_from_address=( str(source.get("LICENSING_EMAIL_FROM_ADDRESS") or "").strip() or None ), account_portal_url=( str(source.get("LICENSING_ACCOUNT_PORTAL_URL") or "").strip() or None ), trusted_proxy_ips=trusted_proxy_ips, ) @dataclass(frozen=True) class DatabaseWorkerSettings: """Database-only configuration for maintenance paths with no other secrets.""" database_url: str environment_name: str def __post_init__(self) -> None: for field_name in ("database_url", "environment_name"): value = getattr(self, field_name) if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") object.__setattr__(self, field_name, value.strip()) if self.environment_name.lower() == "production" and not urlparse( self.database_url ).scheme.startswith("postgresql"): raise ValueError("production licensing requires PostgreSQL") @classmethod def from_environment( cls, environment: Optional[Mapping[str, str]] = None, ) -> "DatabaseWorkerSettings": source = os.environ if environment is None else environment if not isinstance(source, Mapping): raise TypeError("environment must be a mapping or None") return cls( database_url=_required_text(source, "LICENSING_DATABASE_URL"), environment_name=_required_text(source, "LICENSING_ENVIRONMENT"), ) @dataclass(frozen=True) class SubscriptionWorkerSettings(DatabaseWorkerSettings): """Provider-projection configuration without signer, identity, or peppers.""" stripe_secret_key: Optional[str] = None stripe_price_sku_map: Mapping[str, str] = field(default_factory=dict) def __post_init__(self) -> None: super().__post_init__() if self.stripe_secret_key is not None: if ( not isinstance(self.stripe_secret_key, str) or not self.stripe_secret_key.strip() or len(self.stripe_secret_key) > 4096 ): raise ValueError("stripe_secret_key must be a bounded string or None") object.__setattr__( self, "stripe_secret_key", self.stripe_secret_key.strip(), ) if not isinstance(self.stripe_price_sku_map, Mapping): raise TypeError("stripe_price_sku_map must be a mapping") normalized_prices: dict[str, str] = {} for price_id, sku_id in self.stripe_price_sku_map.items(): if not isinstance(price_id, str) or not price_id.strip(): raise ValueError("Stripe price IDs must be non-empty strings") normalized_prices[price_id.strip()] = validate_identifier( sku_id, "Stripe price SKU ID", ) if len(set(normalized_prices.values())) != len(normalized_prices): raise ValueError("each licensing SKU must map to exactly one Stripe price") object.__setattr__( self, "stripe_price_sku_map", MappingProxyType(normalized_prices), ) @classmethod def from_environment( cls, environment: Optional[Mapping[str, str]] = None, ) -> "SubscriptionWorkerSettings": source = os.environ if environment is None else environment if not isinstance(source, Mapping): raise TypeError("environment must be a mapping or None") raw_price_map = str( source.get("LICENSING_STRIPE_PRICE_SKU_MAP_JSON") or "{}" ).strip() try: price_map = json.loads(raw_price_map) except json.JSONDecodeError as exc: raise ValueError( "LICENSING_STRIPE_PRICE_SKU_MAP_JSON must be valid JSON" ) from exc if not isinstance(price_map, dict): raise ValueError("LICENSING_STRIPE_PRICE_SKU_MAP_JSON must be a JSON object") return cls( database_url=_required_text(source, "LICENSING_DATABASE_URL"), environment_name=_required_text(source, "LICENSING_ENVIRONMENT"), stripe_secret_key=( _environment_text(source, "LICENSING_STRIPE_SECRET_KEY") or None ), stripe_price_sku_map=price_map, ) @dataclass(frozen=True) class NotificationWorkerSettings(DatabaseWorkerSettings): """Outbound-email configuration without signer, identity, or billing secrets.""" email_notifications_enabled: bool = False smtp_host: Optional[str] = None smtp_port: int = DEFAULT_SMTP_PORT smtp_tls_mode: str = "implicit" smtp_username: Optional[str] = None smtp_password: Optional[str] = None email_from_address: Optional[str] = None account_portal_url: Optional[str] = None notification_pepper: Optional[bytes] = None def __post_init__(self) -> None: super().__post_init__() if not isinstance(self.email_notifications_enabled, bool): raise TypeError("email_notifications_enabled must be a Boolean") if self.smtp_host is not None: if not isinstance(self.smtp_host, str) or not self.smtp_host.strip(): raise ValueError("smtp_host must be a non-empty string or None") normalized_host = self.smtp_host.strip() if ( len(normalized_host) > 253 or any(character.isspace() for character in normalized_host) or any(character in normalized_host for character in ("/", "@", "?", "#")) ): raise ValueError("smtp_host is invalid") object.__setattr__(self, "smtp_host", normalized_host) if isinstance(self.smtp_port, bool) or not isinstance(self.smtp_port, int): raise TypeError("smtp_port must be an integer") if self.smtp_port < 1 or self.smtp_port > 65535: raise ValueError("smtp_port must be between one and 65535") if not isinstance(self.smtp_tls_mode, str): raise TypeError("smtp_tls_mode must be a string") normalized_tls_mode = self.smtp_tls_mode.strip().lower() if normalized_tls_mode not in SMTP_TLS_MODES: raise ValueError("smtp_tls_mode must be implicit or starttls") object.__setattr__(self, "smtp_tls_mode", normalized_tls_mode) for field_name in ("smtp_username", "smtp_password"): value = getattr(self, field_name) if value is None: continue if not isinstance(value, str) or not value.strip() or len(value) > 4096: raise ValueError(f"{field_name} must be a bounded non-empty string or None") object.__setattr__(self, field_name, value.strip()) if (self.smtp_username is None) != (self.smtp_password is None): raise ValueError("smtp_username and smtp_password must be configured together") if self.email_from_address is not None: object.__setattr__( self, "email_from_address", _email_address(self.email_from_address, "email_from_address"), ) if self.account_portal_url is not None: if not isinstance(self.account_portal_url, str): raise TypeError("account_portal_url must be a string or None") normalized_account_url = self.account_portal_url.strip() parsed_account_url = urlparse(normalized_account_url) if ( parsed_account_url.scheme != "https" or not parsed_account_url.netloc or parsed_account_url.username is not None or parsed_account_url.password is not None or parsed_account_url.fragment ): raise ValueError( "account_portal_url must be an HTTPS URL without credentials or fragment" ) object.__setattr__(self, "account_portal_url", normalized_account_url) if self.notification_pepper is not None: if not isinstance(self.notification_pepper, bytes): raise TypeError("notification_pepper must be bytes or None") if len(self.notification_pepper) < NOTIFICATION_PEPPER_MINIMUM_BYTES: raise ValueError("notification_pepper is too short") if self.email_notifications_enabled and any( value is None for value in ( self.smtp_host, self.smtp_username, self.smtp_password, self.email_from_address, self.account_portal_url, self.notification_pepper, ) ): raise ValueError( "enabled email notifications require SMTP host, username/password, " "from address, account portal URL, and notification pepper" ) @classmethod def from_environment( cls, environment: Optional[Mapping[str, str]] = None, ) -> "NotificationWorkerSettings": source = os.environ if environment is None else environment if not isinstance(source, Mapping): raise TypeError("environment must be a mapping or None") raw_smtp_port = str( source.get("LICENSING_SMTP_PORT") or DEFAULT_SMTP_PORT ).strip() try: smtp_port = int(raw_smtp_port) except ValueError as exc: raise ValueError("LICENSING_SMTP_PORT must be an integer") from exc email_notifications_enabled = str( source.get("LICENSING_EMAIL_NOTIFICATIONS_ENABLED") or "" ).strip().lower() in ("1", "true", "yes") notification_pepper_configured = bool( source.get("LICENSING_NOTIFICATION_PEPPER_BASE64") or source.get("LICENSING_NOTIFICATION_PEPPER_BASE64_FILE") ) return cls( database_url=_required_text(source, "LICENSING_DATABASE_URL"), environment_name=_required_text(source, "LICENSING_ENVIRONMENT"), email_notifications_enabled=email_notifications_enabled, smtp_host=str(source.get("LICENSING_SMTP_HOST") or "").strip() or None, smtp_port=smtp_port, smtp_tls_mode=str( source.get("LICENSING_SMTP_TLS_MODE") or "implicit" ).strip(), smtp_username=( _environment_text(source, "LICENSING_SMTP_USERNAME") or None ), smtp_password=( _environment_text(source, "LICENSING_SMTP_PASSWORD") or None ), email_from_address=( str(source.get("LICENSING_EMAIL_FROM_ADDRESS") or "").strip() or None ), account_portal_url=( str(source.get("LICENSING_ACCOUNT_PORTAL_URL") or "").strip() or None ), notification_pepper=( _secret_bytes( source, "LICENSING_NOTIFICATION_PEPPER_BASE64", NOTIFICATION_PEPPER_MINIMUM_BYTES, ) if notification_pepper_configured else None ), ) __all__ = [ "DatabaseWorkerSettings", "NotificationWorkerSettings", "ServerSettings", "SubscriptionWorkerSettings", ]