aegis/config.py
Rama main · réplica de lectura del repositorio original.
"""
aegis.config — Centralized configuration via environment variables.
"""
# Copyright (c) 2026 Juan Luna. All rights reserved.
# SPDX-License-Identifier: LicenseRef-Aegis-Commercial-1.0
# Licensed under the Aegis Commercial License 1.0; see LICENSE and NOTICE.
from __future__ import annotations
import hashlib
import hmac
import json
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal
from urllib.parse import urlparse
from pydantic import AnyHttpUrl, Field, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class AegisSettings(BaseSettings):
model_config = SettingsConfigDict(
env_prefix="AEGIS_",
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# ── Provider ─────────────────────────────────────────────────────────
provider: str = Field(
default="openai",
description=(
"LLM provider adapter to use. "
"Valid values: openai, anthropic, gemini, openrouter. "
"Controls request/response translation and auth header format. "
"Set to 'openai' for any OpenAI-compatible endpoint (vLLM, Ollama, LM Studio)."
),
)
provider_model: str = Field(
default="",
description=(
"Override the model name sent to the upstream provider. "
"When set, replaces the 'model' field in every request. "
"Useful for routing all traffic through a single backend model "
"regardless of what the client requests."
),
)
# ── OpenRouter extras ─────────────────────────────────────────────
openrouter_site_url: str = Field(
default="",
description="HTTP-Referer header for OpenRouter analytics (optional).",
)
openrouter_site_name: str = Field(
default="",
description="X-Title header for OpenRouter analytics (optional).",
)
# ── Anthropic extras ──────────────────────────────────────────────
anthropic_api_version: str = Field(
default="2023-06-01",
description="anthropic-version header value. Change only when targeting a beta API version.",
)
# ── Backend ──────────────────────────────────────────────────────────
backend_url: AnyHttpUrl = Field(
# Built as AnyHttpUrl rather than str so the declared default matches the
# declared type. pydantic-settings validates defaults, so the runtime
# value was already AnyHttpUrl; only the annotation disagreed.
default=AnyHttpUrl("http://localhost:11434"),
description=(
"URL of the upstream LLM backend (OpenAI-compatible). "
"Ignored for providers with a fixed base URL "
"(anthropic: api.anthropic.com, gemini: generativelanguage.googleapis.com, "
"openrouter: openrouter.ai/api/v1)."
),
)
backend_api_key: str = Field(
default="",
description="API key forwarded to the upstream backend. If empty, will be fetched from Vault.",
)
backend_timeout_seconds: float = Field(
default=120.0,
ge=1.0,
le=600.0,
description="Total timeout for upstream requests (seconds).",
)
backend_connect_timeout_seconds: float = Field(
default=10.0,
ge=0.5,
le=30.0,
)
# ── Vault Integration ────────────────────────────────────────────────
vault_url: str = Field(
default="",
description="URL of the HashiCorp Vault server.",
)
vault_role_id: str = Field(
default="",
description="AppRole RoleID for Vault authentication.",
)
vault_secret_id: str = Field(
default="",
description="AppRole SecretID for Vault authentication.",
)
vault_backend_secret_path: str = Field(
default="secret/data/aegis/backend",
description="Vault path for the LLM backend API key.",
)
# ── Auth ─────────────────────────────────────────────────────────────
api_keys: str = Field(
default="",
description=(
"Comma-separated list of allowed API keys for proxy access. "
"Leave empty to disable auth (NOT recommended for production)."
),
)
audit_api_keys: str = Field(
default="",
description="Comma-separated API keys for the /audit/* read endpoints.",
)
auth_disabled: bool = Field(
default=False,
description="Set True only for local development. Never in production.",
)
api_key_scopes: str = Field(
default="",
description=(
"Semicolon-separated HIPAA minimum-necessary scope restrictions per API key "
"(a scoping control, not a compliance determination). "
"Format: 'key1:scope1,scope2;key2:scope3'. "
"Valid scopes: proxy:completions, audit:read, audit:export, audit:analytics. "
"With legacy unmapped-principal compatibility enabled, keys not listed here receive "
"all scopes; otherwise unmapped keys receive no scopes. "
"Example: 'read-only-key:audit:read;export-key:audit:read,audit:export'. "
"Set via AEGIS_API_KEY_SCOPES environment variable."
),
)
signing_key: str = Field(
default="",
description=(
"Dedicated HMAC-SHA256 signing key for the Merkle audit chain. "
"MUST be separate from AEGIS_API_KEYS. "
"Generate with: python -c 'import secrets; print(secrets.token_hex(32))' "
"If empty and auth is not disabled, a warning is emitted at startup."
),
)
auth_mode: Literal["api_key", "oidc", "mtls", "api_key_mtls", "oidc_mtls"] = Field(
default="api_key",
description="Single configured authentication mode; composite modes require both factors.",
)
auth_identity_hmac_key: str = Field(
default="",
description="Secret used only to derive opaque stable credential identifiers.",
)
api_key_principals_json: str = Field(
default="",
description=(
"JSON object keyed by lowercase HMAC-SHA256 API-key digest using "
"auth_identity_hmac_key. Each value contains "
"tenant_id, roles, and scopes. Strict mode requires an entry for every configured key."
),
)
allow_legacy_unmapped_api_key_principals: bool = Field(
default=False,
description=(
"Compatibility-only opt-in that grants legacy API_KEY_SCOPES permissions to valid "
"API keys without an explicit principal mapping in development mode. Unmapped keys "
"otherwise authenticate with no roles or scopes, and strict mode always rejects them."
),
)
development_tenant_id: str = Field(default="development")
oidc_issuer: str = Field(default="")
oidc_audience: str = Field(default="")
oidc_jwks_url: str = Field(default="")
oidc_algorithms: str = Field(default="RS256")
oidc_tenant_claim: str = Field(default="tenant_id")
oidc_roles_claim: str = Field(default="roles")
oidc_leeway_seconds: int = Field(default=0, ge=0, le=300)
oidc_http_timeout_seconds: float = Field(default=5.0, ge=0.1, le=30.0)
mtls_trusted_proxy_cidrs: str = Field(default="")
mtls_allowed_sha256_fingerprints: str = Field(default="")
mtls_san_allowlist: str = Field(default="")
mtls_tenant_san_prefix: str = Field(default="tenant:")
mtls_default_roles: str = Field(default="proxy_user")
# ── LDAP / Active Directory ───────────────────────────────────────────────
ldap_url: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"LDAP server URL for multi-factor identity assertion. "
"Use 'ldaps://' for direct TLS (recommended) or 'ldap://' with "
"AEGIS_LDAP_USE_START_TLS=true for StartTLS. "
"Example: ldaps://dc.corp.example.com:636. "
"Leave empty to disable LDAP authentication."
),
)
ldap_base_dn: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"LDAP base Distinguished Name for user and group searches. "
"Example: DC=corp,DC=example,DC=com"
),
)
ldap_bind_dn: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Service-account DN for the initial directory search bind (least-privilege read). "
"Example: CN=svc-aegis,OU=ServiceAccounts,DC=corp,DC=example,DC=com. "
"Provide via AEGIS_LDAP_BIND_DN environment variable or Vault."
),
)
ldap_bind_password: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Password for the LDAP service account. "
"Provide via AEGIS_LDAP_BIND_PASSWORD or Vault; never hard-code."
),
)
ldap_user_search_filter: str = Field(
default="(|(sAMAccountName={username})(userPrincipalName={username}))",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"LDAP search filter template for user lookup. {username} is substituted "
"with the RFC 4515-escaped login name. "
"AD default covers sAMAccountName and UPN formats. "
"POSIX LDAP alternative: (uid={username})"
),
)
ldap_user_search_base: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"DN subtree for user searches. Defaults to ldap_base_dn when empty. "
"Example: OU=Users,DC=corp,DC=example,DC=com"
),
)
ldap_required_groups: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Comma-separated CN or DN values of LDAP groups the authenticated user "
"must belong to (at least one). Empty string disables group check. "
"Example: AegisUsers,AegisAdmins"
),
)
ldap_ad_mode: bool = Field(
default=True,
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Enable Active Directory extensions: nested group OID "
"(1.2.840.113556.1.4.1941), memberOf enumeration, sAMAccountName lookup. "
"Set False for plain RFC 4519 LDAP directories."
),
)
ldap_use_start_tls: bool = Field(
default=False,
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Upgrade plain ldap:// to TLS via StartTLS before bind."
),
)
ldap_ca_certs_file: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Path to PEM CA bundle for LDAP TLS peer verification. "
"Empty string uses the system default trust store."
),
)
ldap_timeout_seconds: float = Field(
default=10.0,
ge=1.0,
le=60.0,
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"Socket-level timeout (seconds) for LDAP connect and operations."
),
)
def get_ldap_required_groups(self) -> frozenset[str]:
if not self.ldap_required_groups:
return frozenset()
return frozenset(g.strip() for g in self.ldap_required_groups.split(",") if g.strip())
# ── mTLS / SSL Hardening ──────────────────────────────────────────────
ssl_certfile: Path | None = Field(
default=None,
description="Path to the server SSL certificate file.",
)
ssl_keyfile: Path | None = Field(
default=None,
description="Path to the server SSL private key file.",
)
ssl_ca_certs: Path | None = Field(
default=None,
description="Path to the CA bundle used to verify client certificates (mTLS).",
)
mtls_required: bool = Field(
default=False,
description="If True, the server will require and verify a client certificate.",
)
cac_piv_required: bool = Field(
default=False,
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-064).]** "
"When True, mTLS client certificates must carry a recognized DoD CAC or GSA PIV "
"certificate policy OID (DoDI 8520.02 / NIST SP 800-73-4 / GSA FPKI) and the "
"Client Authentication EKU. EDIPI (CAC) or UUID (PIV-I) is extracted and logged. "
"Requires ssl_ca_certs to be configured for proper chain validation."
),
)
# ── Storage ───────────────────────────────────────────────────────────
wal_path: Path = Field(
default=Path("./aegis.wal.jsonl"),
description="Path to the Write-Ahead Log for the Merkle audit chain.",
)
max_memory_nodes: int = Field(
default=100_000,
ge=1_000,
description="In-memory Merkle chain cap (deque sliding window).",
)
max_wal_bytes: int = Field(
default=0,
ge=0,
description=(
"Rotate the active WAL into an archived, immutable segment once it "
"reaches this many bytes. 0 disables rotation (single unbounded WAL). "
"Archived segments (named '<wal_path>.NNNNNN', mode 0o600) are "
"retained for forensic completeness and replayed on startup; the "
"audit chain is never truncated."
),
)
max_forensic_bytes: int = Field(
default=65_536,
ge=0,
le=65_536,
description=(
"Cap, in bytes, on each of the request and response previews an audit "
"leaf embeds (hex-encoded, so up to 4x this many leaf bytes). Digests "
"and sizes are always recorded; 0 keeps no preview. Lowering it is "
"the precondition for zero-knowledge inclusion proofs (DOC-08 §6.3) "
"and costs the preview evidence those bytes carry. Safe to lower on "
"an existing chain: verification never rebuilds a leaf. The default "
"is also the ceiling."
),
)
s3_archive_enabled: bool = Field(default=False)
s3_archive_bucket: str = Field(default="")
s3_archive_prefix: str = Field(default="aegis/wal")
s3_archive_region: str = Field(default="")
s3_archive_retention_days: int = Field(default=365, ge=1, le=36_500)
s3_archive_lock_mode: Literal["GOVERNANCE", "COMPLIANCE"] = Field(default="COMPLIANCE")
s3_archive_spool_dir: Path = Field(default=Path("./aegis-archive-spool"))
s3_archive_journal_path: Path = Field(default=Path("./aegis-archive.sqlite3"))
s3_archive_max_spool_bytes: int = Field(default=1_073_741_824, ge=1)
tsa_url: str = Field(default="")
tsa_ca_file: Path | None = Field(default=None)
tsa_evidence_dir: Path = Field(default=Path("./aegis-rfc3161"))
# ── Runtime enforcement and bounded resources ─────────────────────────
security_enforcement_mode: Literal["strict", "development"] = Field(
default="strict",
description=(
"strict is the only production mode: required auth, durable signed evidence, "
"distributed rate limiting, and privileged kernel controls. development is for "
"isolated tests/local work and must never be used for a production claim."
),
)
require_durable_evidence: bool = Field(
default=True,
description="Reject governed requests when signed durable evidence cannot be committed.",
)
require_lsm: bool = Field(
default=True,
description="Require active AppArmor/SELinux confinement in strict runtime mode.",
)
require_seccomp: bool = Field(
default=True,
description="Require an active seccomp filter in strict runtime mode.",
)
max_request_body_bytes: int = Field(
default=1_048_576,
ge=1_024,
le=16_777_216,
description="Streaming HTTP body limit enforced before JSON parsing.",
)
max_stream_response_bytes: int = Field(
default=16_777_216,
ge=1_024,
le=268_435_456,
description=(
"Maximum cumulative canonical SSE response bytes before upstream is aborted. "
"This is a total-output policy limit, not the retained-memory strategy."
),
)
stream_queue_max_items: int = Field(
default=64,
ge=1,
le=1_024,
description="Maximum number of canonical SSE events retained per active stream.",
)
stream_queue_max_bytes: int = Field(
default=1_048_576,
ge=1_024,
le=16_777_216,
description="Maximum canonical SSE bytes retained in one stream queue.",
)
max_stream_event_bytes: int = Field(
default=65_536,
ge=256,
le=1_048_576,
description="Maximum accepted upstream or canonical SSE event size.",
)
stream_deidentifier_window_chars: int = Field(
default=128,
ge=64,
le=4_096,
description="Finite logical-text holdback for cross-event PHI/PCI interception.",
)
mmr_hash_scheme: Literal["auto", "v1-asciihex", "v2-binary-domain-separated"] = Field(
default="auto",
description=(
"MMR hash construction for the audit ledger. 'auto' (the default) starts a NEW "
"chain on 'v2-binary-domain-separated', which applies RFC 6962 domain tags and so "
"prevents the leaf/interior-node type confusion v1 admits, and reopens an EXISTING "
"chain under whichever scheme its WAL recorded. Pinning a scheme explicitly is "
"fail-closed: a WAL written under the other one is refused with fault state "
"'mmr_scheme_mismatch' rather than replayed to a different root. There is no "
"in-place upgrade — the scheme decides every root a chain has recorded — so moving "
"an existing v1 chain to v2 means starting a new chain."
),
)
enable_cryptographic_shredding: bool = Field(
default=False,
description=(
"Seal each committed leaf under a per-subject AES-256-GCM key so the record can "
"later be rendered unreadable by destroying that key, leaving every root, peak and "
"issued inclusion proof untouched. Off by default: it changes what the MMR commits "
"to, so it cannot be turned on for a chain that already has records. It also makes "
"the WAL carry ciphertext rather than only digests, so WAL growth tracks payload "
"size. Destroying a key makes the sealed leaf unrecoverable to a holder of the "
"ciphertext; it is not media sanitisation and does not remove the request and "
"response digests the node still carries."
),
)
shredder_vault_path: str = Field(
default="",
description=(
"Where per-subject shredding keys live. Empty places the vault beside the WAL "
"('<wal_path>.shredder.db'), which keeps the two under one custody boundary. The "
"vault holds the keys that make the ledger readable: back it up with the WAL or "
"lose every plaintext, and sanitise it on erasure or keep them past it."
),
)
pqc_identity_path: str = Field(
default="",
description=(
"Filesystem path to the ledger's persistent ML-DSA-65 signing identity. Empty (the "
"default) disables the post-quantum signing tier entirely, so signing falls through "
"to HMAC-SHA256. When set, the identity is created on first use and reused for every "
"signature, which is what makes a signature attributable. The file holds the raw "
"private key and is written 0600; give it the custody any signing secret needs."
),
)
trusted_signing_public_keys: str = Field(
default="",
description=(
"Comma-separated hex ML-DSA public keys whose signatures the ledger accepts in "
"addition to its own identity's (AEGIS_PQC_IDENTITY_PATH): earlier identities after "
"a rotation, and the identity of every other replica that writes the same chain in "
"high availability. While the ledger holds its own identity or this list is "
"non-empty, a pqc-ml-dsa record signed under any other key is invalid and fails "
"verify_integrity, because a key read from the record itself is whatever whoever "
"last wrote the WAL chose; with neither, such records read unverified. Take each "
"key from the identity at provisioning, never from the chain."
),
)
terminal_outbox_enabled: bool = Field(
default=False,
description=(
"Spool every terminal commit a stream hands off at teardown to a durable, "
"content-free outbox beside the WAL, and replay what is still pending at the next "
"start, before traffic is served (REG-D32). Off, a crash or SIGKILL between "
"teardown and the handed-off commit loses that terminal node, and a full handoff "
"queue drops it. A recovered node is signed as stream-terminal-evidence-recovered "
"with evidence_status recovered-terminal, carries no request or response preview, "
"and is timestamped at replay. Requires AEGIS_SIGNING_KEY: every spool line is "
"authenticated with a key derived from it, and a line that fails is quarantined, "
"never replayed. If enabled without a signing key, or if the outbox cannot be "
"opened, startup fails."
),
)
terminal_outbox_path: Path | None = Field(
default=None,
description=(
"Outbox spool file. Unset, it is <wal_path>.terminal-outbox.jsonl, beside the WAL "
"and so covered by the same volume and single-writer lock."
),
)
terminal_outbox_max_bytes: int = Field(
default=16 * 1024 * 1024,
ge=64 * 1024,
description=(
"Size at which the outbox stops accepting new records (each is about 0.5 KiB); "
"past it a handoff is in-memory only and counted in "
"aegis_terminal_outbox_errors_total. Compaction reclaims done records."
),
)
max_concurrent_streams: int = Field(
default=256,
ge=0,
description=(
"Admission ceiling for concurrent governed SSE streams in this process. "
"Per-stream memory is bounded (R_max), but until this existed the number of "
"streams was not, so aggregate retained bytes scaled with however many "
"connections an adversary opened. Over the ceiling, admission is refused with "
"HTTP 429 before the upstream call rather than after buffers are allocated. "
"0 disables the ceiling and restores the previous unbounded behaviour. "
"This counts streams in ONE process: it is not a cluster-wide limit, and "
"N replicas admit up to N x this value."
),
)
streaming_engine: Literal["grammar_frontier", "deidentifier"] = Field(
default="grammar_frontier",
description=(
"Streaming redactor for governed SSE. 'grammar_frontier' runs the Safe Harbor "
"de-identifier and then the grammar-frontier rules, which add instruction-override "
"and system-prompt-disclosure matching the de-identifier does not carry. "
"'deidentifier' runs the Safe Harbor set alone, as releases before 4.3.0 did. "
"The composite holds back the de-identifier window plus the 28-character frontier, "
"so the per-stream retained-byte ceiling rises by 4x28 bytes."
),
)
gossip_enabled: bool = Field(
default=False,
description=(
"Run the cross-replica reconciliation daemon. Off by default: it opens a "
"second listener, requires a cluster CA and per-replica certificates, and "
"reconciles an accumulator that is separate from the audit ledger — so "
"enabling it changes nothing about what the WAL records or what a receipt "
"proves. It gives replicas one shared CRDT root; it is not consensus, it "
"does not order the ledger across replicas, and it is not Byzantine "
"fault tolerant."
),
)
gossip_replica_id: int = Field(
default=0,
ge=0,
le=4_294_967_295,
description=(
"This replica's identifier in the gossip mesh. Must be unique per replica "
"and stable across restarts: it is the key in every vector clock, so two "
"replicas sharing an id are indistinguishable to the causality tracking "
"and a changed id abandons the entries recorded under the old one. Under "
"the Helm chart, derive it from the StatefulSet ordinal."
),
)
gossip_self_name: str = Field(
default="",
description=(
"This replica's own name in the mesh, excluded from the peer list. The "
"peer list is rendered from the replica count and cannot know which "
"ordinal it is being rendered for, so a replica removes itself here. "
"Under the Helm chart this is the pod name."
),
)
gossip_peers: str = Field(
default="",
description=(
"Comma-separated 'name=https://host:port' peers, excluding this replica. "
"Static rather than discovered: there is no membership protocol, so "
"adding or removing a replica is an operator action. The name must match "
"the peer certificate."
),
)
gossip_interval_seconds: float = Field(
default=5.0,
gt=0.0,
le=3_600.0,
description=(
"Seconds between anti-entropy rounds. A converged cluster exchanges only "
"a root digest per peer per round, so this is cheap to keep short; the "
"cost of a longer interval is how stale a replica may be, not bandwidth."
),
)
gossip_bind_host: str = Field(
default="0.0.0.0", # noqa: S104
description=(
"Interface the inbound gossip listener binds. Defaults to all interfaces "
"because the mesh is reached by pod DNS, and the boundary that protects it "
"is the mandatory client certificate rather than the bind address; narrow "
"it further with a NetworkPolicy or by setting a specific address."
),
)
gossip_bind_port: int = Field(
default=9443,
ge=1,
le=65_535,
description=(
"Port for the inbound gossip listener. Separate from the gateway's port on "
"purpose: customer traffic and replica traffic never share a listener, so "
"one misconfiguration cannot expose both."
),
)
# ── Multi-replica operation (aegis.core.ha) ────────────────────────────
ha_mode: str = Field(
default="single",
description=(
"'single' (one writer, no coordination), 'active_passive' (replicas compete "
"for one chain's writer lease; the holder serves, the others stand by), or "
"'active_active' (every replica writes its own chain under its own lease and "
"every committed node is appended to one shared, hash-linked global sequence)."
),
)
ha_chain_id: str = Field(
default="",
description=(
"Chain this replica writes. active_passive: the one chain all replicas "
"share (required). active_active: this replica's own chain; defaults to the "
"host name, which a StatefulSet makes unique and stable."
),
)
ha_redis_url: str = Field(
default="",
description="Redis for the writer lease. Empty = redis_url.",
)
ha_lease_ttl_seconds: float = Field(
default=15.0,
ge=3.0,
le=300.0,
description=(
"Writer lease lifetime. The holder renews every third of it and stops "
"admitting requests once less than a fifth remains unrenewed."
),
)
ha_standby_poll_seconds: float = Field(
default=2.0,
ge=0.2,
le=60.0,
description="How often a standby replica retries the writer lease.",
)
ha_sequencer_url: str = Field(
default="",
description=(
"Global sequence store: postgresql://… (production) or sqlite:///absolute/path "
"(single host, non-strict only). Required for active_active; optional for "
"active_passive, where it adds storage-level fencing by lease epoch."
),
)
ha_max_sequencing_lag_seconds: float = Field(
default=30.0,
ge=1.0,
le=3600.0,
description=(
"Admission is refused (503) while the oldest durable, not-yet-sequenced "
"node is older than this: the bound on how far the global sequence may "
"trail the local chains."
),
)
gossip_client_certificate: str = Field(
default="",
description="PEM certificate this replica presents to peers, in both directions.",
)
gossip_client_private_key: str = Field(
default="",
description="PEM private key for the gossip certificate. Give it signing-key custody.",
)
gossip_certificate_authority: str = Field(
default="",
description=(
"PEM CA bundle that issues replica certificates. Required whenever gossip "
"is enabled and there is no flag to skip verification: a peer supplies "
"leaves that enter this replica's accumulator, so an unauthenticated mesh "
"is an open input rather than a cluster."
),
)
max_stream_duration_seconds: float = Field(
default=60.0,
ge=0.1,
le=900.0,
description="Total wall-clock deadline for consuming one upstream SSE response.",
)
analysis_queue_size: int = Field(
default=2_048,
ge=1,
le=100_000,
description="Bounded asynchronous response-analysis queue capacity.",
)
analysis_worker_count: int = Field(
default=2,
ge=1,
le=64,
description="Number of bounded response-analysis workers.",
)
analysis_sample_rate: float = Field(
default=1.0,
ge=0.0,
le=1.0,
description="Fraction of successful responses submitted for asynchronous analysis.",
)
analysis_timeout_seconds: float = Field(
default=5.0,
ge=0.1,
le=60.0,
description="Maximum wall time for one asynchronous response analysis job.",
)
analysis_shutdown_timeout_seconds: float = Field(
default=2.0,
ge=0.1,
le=30.0,
description="Maximum shutdown wait for cancelled response-analysis workers.",
)
# ── Telemetry ─────────────────────────────────────────────────────────
force_logprobs: bool = Field(
default=False,
description=(
"Inject logprobs=True and top_logprobs=N into every chat request "
"for entropy analysis. Disabled by default because it inflates response "
"size 5-10x and is unsupported by Anthropic/Gemini providers. "
"Enable explicitly when using an OpenAI-compatible backend that supports it."
),
)
top_logprobs: int = Field(
default=20,
ge=1,
le=20,
description="Number of top logprobs to request (max 20, OpenAI limit).",
)
entropy_alert_threshold_bits: float = Field(
default=1.0,
ge=0.0,
description="Shannon entropy drop (bits) below baseline that triggers ALERT.",
)
kl_alert_threshold: float = Field(
default=2.0,
ge=0.0,
description="KL divergence above which an ALERT is emitted.",
)
js_alert_threshold: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description="JS divergence above which an ALERT is emitted.",
)
# ── Rate Limiting ─────────────────────────────────────────────────────
rate_limit_requests_per_minute: int = Field(
default=60,
ge=1,
le=10_000,
description="Default request limit per minute per tenant/session.",
)
rate_limit_threshold: int = Field(
default=60,
ge=1,
le=10_000,
description="Rate limit threshold (requests per minute).",
)
rate_limit_window: int = Field(
default=60,
ge=1,
le=3600,
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-066).]** "
"Rate limit window in seconds. "
"The active limiter uses its own fixed window (rate_limit_requests_per_minute, "
"rate_limit_burst, rate_limit_token_capacity); this value is read by no code path."
),
)
rate_limit_burst: int = Field(
default=10,
ge=1,
le=100,
description="Maximum burst capacity for the token bucket.",
)
redis_url: str = Field(
default="redis://localhost:6379",
description="Redis connection URL for distributed rate limiting.",
)
rate_limit_backend: str = Field(
default="memory",
description="Rate limiter backend: 'memory' (default, no Redis) or 'redis'.",
)
require_distributed_limiter: bool = Field(
default=False,
description=(
"Refuse startup, in any enforcement mode, unless rate_limit_backend is "
"'redis'. Strict mode already requires Redis; this makes the example "
"deployment variable an enforced control rather than an inert one (REG-D60)."
),
)
rate_limit_token_capacity: int = Field(default=100_000, ge=1)
rate_limit_tokens_per_minute: int = Field(default=100_000, ge=1)
rate_limit_default_output_tokens: int = Field(default=4_096, ge=1)
rate_limit_max_output_tokens: int = Field(default=16_384, ge=1)
rate_limit_max_buckets: int = Field(default=10_000, ge=2, le=1_000_000)
request_entropy_guard: bool = Field(
default=False,
description="Block requests failing Shannon-entropy heuristics (enable in hardened mode).",
)
waf_strict_mode: bool = Field(
default=True,
description="Reject payloads that match known prompt-injection patterns.",
)
wal_min_free_bytes: int = Field(
default=0,
ge=0,
description=(
"Refuse governed traffic at ingress when the WAL volume has less than this "
"many bytes free. 0 (the default) disables the check. "
"This is an EARLY-REFUSAL optimisation, not a missing safety net: a commit "
"onto a full volume already fails closed -- the write or the fsync raises, "
"`wal_persist_failed` latches, and governed endpoints answer 503. What the "
"preflight adds is refusing BEFORE the upstream provider is called and billed, "
"and an unambiguous operator signal. "
"It defaults off because a false refusal on a nearly-full but working volume "
"is a total outage, which is a worse failure than the one it prevents; an "
"operator who knows their headroom sets it deliberately."
),
)
rag_injection_scanning: bool = Field(
default=True,
description=(
"Scan retrieved content — tool results, function results, and RAG context "
"blocks — for embedded prompt injection before forwarding. This covers the "
"indirect case the WAF structurally cannot: the user turn is clean and the "
"payload arrives inside data the application fetched on the model's behalf. "
"Set false where retrieved content legitimately contains role-like headers "
"(system logs, chat transcripts, structured records), which the scanner's "
"own documentation identifies as its false-positive mode. Disabling removes "
"a detection; it does not weaken any evidence or durability guarantee."
),
)
rag_injection_block_threshold: float = Field(
default=0.5,
gt=0.0,
le=1.0,
description=(
"Minimum aggregate risk score at which retrieved content is refused. "
"Raise toward 1.0 to admit more; 1.0 refuses only content that trips a "
"signal weighted at full confidence. Validated here so a bad value fails "
"at startup rather than on the first request that would have been scanned."
),
)
waf_session_window: int = Field(
default=10,
ge=2,
le=100,
description=(
"Sliding-window size (number of recent turns) examined by the multi-turn "
"behavioral WAF. Older turns are evicted automatically."
),
)
waf_session_cumulative_threshold: float = Field(
default=2.0,
ge=0.1,
description=(
"Sum of per-turn WAF scores within the session window that triggers "
"a session-level block. Score per turn is in [0, 1]; default 2.0 "
"requires at least two full-weight soft hits in the window."
),
)
waf_session_crescendo_turns: int = Field(
default=3,
ge=2,
le=20,
description=(
"Number of consecutive turns each producing a non-zero WAF score "
"before a crescendo (gradual constraint erosion) block is triggered."
),
)
# ── Server ────────────────────────────────────────────────────────────
debug_mode: bool = Field(
default=False,
description=(
"Enable /docs and /redoc OpenAPI endpoints. "
"NEVER enable in production — exposes full API schema. "
"Automatically forces auth_disabled=False check at startup."
),
)
host: str = Field(default="0.0.0.0")
port: int = Field(default=8080, ge=1, le=65535)
workers: int = Field(default=1, ge=1, le=64)
log_level: str = Field(default="INFO")
cors_origins: str = Field(
default="",
description="Comma-separated CORS allowed origins. Empty = no CORS headers.",
)
dmz_allowed_source_ips: str = Field(
default="",
description=(
"DMZ-mode: comma-separated list of IPv4/IPv6 addresses or CIDR networks "
"that are allowed to reach the proxy. When non-empty, every request whose "
"client IP is not in the allowlist is rejected with 403 Forbidden before "
"any authentication is attempted. Supports exact addresses (e.g. 10.0.0.1) "
"and prefix notation (e.g. 10.0.0.0/24, ::1/128). Empty string (default) "
"disables DMZ mode — all source IPs are permitted. "
"Configure via AEGIS_DMZ_ALLOWED_SOURCE_IPS."
),
)
dmz_trust_proxy_headers: bool = Field(
default=False,
description=(
"When True, DMZ-mode reads the real client IP from X-Forwarded-For or "
"X-Real-IP headers (trusted reverse-proxy scenario). "
"Never set this when the proxy is internet-facing — it allows IP spoofing. "
"Only enable behind a trusted load balancer or nginx ingress."
),
)
# ── Air-gap egress enforcement ────────────────────────────────────────
airgap_mode: bool = Field(
default=False,
description=(
"When True, the proxy blocks all outbound HTTP connections except to hosts "
"listed in AEGIS_AIRGAP_ALLOWED_HOSTS plus the configured upstream backend. "
"Provides application-layer OT network zone enforcement "
"(IEC 62443 Zone and Conduit model). "
"Pair with kernel-level controls (nftables, network namespaces) for defense-in-depth. "
"Configure via AEGIS_AIRGAP_MODE."
),
)
airgap_allowed_hosts: str = Field(
default="",
description=(
"Comma-separated list of allowed outbound hostnames or host:port pairs "
"when AEGIS_AIRGAP_MODE=true. The upstream backend host is always implicitly "
"allowed. Example: 'internal-llm.corp.example.com,10.0.0.5:8080'. "
"An empty value (with airgap_mode=true) only permits the upstream backend. "
"Configure via AEGIS_AIRGAP_ALLOWED_HOSTS."
),
)
# ── HSM / PKCS#11 signing ─────────────────────────────────────────────
pkcs11_library_path: str = Field(
default="",
description=(
"Filesystem path to the PKCS#11 shared library. Empty string disables HSM signing. "
"Examples: /usr/lib/softhsm/libsofthsm2.so (SoftHSM2), "
"/opt/cloudhsm/lib/libcloudhsm_pkcs11.so (AWS CloudHSM). "
"Requires python-pkcs11 to be installed (pip install python-pkcs11). "
"When configured, audit nodes are signed by the HSM-resident key instead of the "
"in-memory AEGIS_SIGNING_KEY; the private key NEVER enters application memory."
),
)
pkcs11_slot_id: int = Field(
default=0,
ge=0,
description="PKCS#11 slot index (0-based). Ignored when AEGIS_PKCS11_TOKEN_LABEL is set.",
)
pkcs11_pin: str = Field(
default="",
description=(
"PKCS#11 User PIN for C_Login. Provide via environment variable "
"AEGIS_PKCS11_PIN or Vault; never hard-code in config files."
),
)
pkcs11_key_label: str = Field(
default="aegis-signing-key",
description="CKA_LABEL of the private signing key stored in the PKCS#11 token.",
)
pkcs11_token_label: str = Field(
default="",
description=(
"When non-empty, resolve the slot by token label rather than pkcs11_slot_id. "
"Useful when slot numbering is dynamic (e.g. AWS CloudHSM)."
),
)
# ── Reliability: circuit breaker ──────────────────────────────────────
circuit_breaker_failure_threshold: int = Field(
default=5,
ge=1,
le=100,
description=(
"Consecutive upstream failures before the circuit opens. "
"While OPEN all requests return 503 immediately (fail-fast). "
"After circuit_breaker_recovery_timeout seconds one probe is allowed."
),
)
circuit_breaker_recovery_timeout: float = Field(
default=30.0,
ge=1.0,
le=600.0,
description="Seconds the circuit stays OPEN before allowing a recovery probe.",
)
circuit_breaker_success_threshold: int = Field(
default=2,
ge=1,
le=20,
description="Consecutive probe successes in HALF_OPEN required to re-close the circuit.",
)
# ── Privacy / PII redaction ────────────────────────────────────────────
phi_deidentify: bool = Field(
default=False,
description=(
"When True, apply best-effort regex PHI identifier redaction "
"to request message content before forwarding to the upstream LLM, and to response "
"content before returning to the client. Scrubs 18 HIPAA identifier categories via "
"regex (names, DOB, SSN, MRN, phone, email, IP, URL, etc.). Does not require an "
"NLP model. This is not a HIPAA Safe Harbor or Expert Determination conclusion."
),
)
pci_scrub: bool = Field(
default=False,
description=(
"When True, apply real-time PCI-DSS v4.0 cardholder-data detection and masking "
"to request message content before forwarding to the upstream LLM, and to response "
"content before returning to the client. Detects PANs (Luhn + IIN gate), "
"CVV/CVC security codes (in context), and Track 1/2 magnetic-stripe data. "
"PANs are masked to last-4 following the PCI-DSS §3.4 masking requirement — a technical control, not a PCI certification; CVV and track data are fully redacted. "
"Enable for PCI-DSS-scoped deployments (payments, e-commerce, financial services)."
),
)
pii_redact_tenant_id: bool = Field(
default=False,
description=(
"When True, replace tenant_id (session/user identifier) in the WAL with "
"a one-way SHA-256 prefix before committing to the audit chain. "
"Supports data-minimization review where session IDs are "
"considered personal data. Does not affect in-flight analysis — only "
"the durable WAL record. Cannot be reversed without the original ID."
),
)
# ── Alerting ──────────────────────────────────────────────────────────
webhook_url: str = Field(
default="",
description=(
"**[Not wired in 5.0.3 \u2014 no code path reads this field; setting it enforces nothing (UC-066).]** "
"HTTP(S) URL to POST alert payloads to (Slack, Teams, custom SIEM). "
"The alert sender reads siem_url / siem_* instead, so setting only this "
"variable delivers nothing; see AUD-35 in ROADMAP.md."
),
)
webhook_timeout_seconds: float = Field(default=5.0, ge=0.5, le=30.0)
siem_url: str = Field(default="")
siem_bearer_token: str = Field(default="")
siem_format: Literal["cef", "rfc5424", "splunk", "datadog"] = Field(default="cef")
siem_spool_path: Path = Field(default=Path("./aegis-siem.sqlite3"))
siem_spool_max_rows: int = Field(default=100_000, ge=1)
siem_spool_max_bytes: int = Field(default=268_435_456, ge=1)
siem_max_payload_bytes: int = Field(default=16_384, ge=256, le=1_048_576)
@field_validator("provider")
@classmethod
def _validate_provider(cls, v: str) -> str:
allowed = {"openai", "anthropic", "gemini", "openrouter"}
v_lower = v.strip().lower()
if v_lower not in allowed:
raise ValueError(f"AEGIS_PROVIDER must be one of {sorted(allowed)}, got {v!r}")
return v_lower
@field_validator("security_enforcement_mode")
@classmethod
def _validate_security_enforcement_mode(cls, v: str) -> str:
value = v.strip().lower()
if value not in {"strict", "development"}:
raise ValueError("security_enforcement_mode must be 'strict' or 'development'")
return value
@field_validator("trusted_signing_public_keys")
@classmethod
def _validate_trusted_signing_public_keys(cls, v: str) -> str:
keys = [key.strip() for key in v.split(",") if key.strip()]
for key in keys:
if len(key) % 2 or any(c not in "0123456789abcdefABCDEF" for c in key):
raise ValueError(
"AEGIS_TRUSTED_SIGNING_PUBLIC_KEYS must be comma-separated hex public keys"
)
return ",".join(key.lower() for key in keys)
@field_validator("rate_limit_backend")
@classmethod
def _validate_rate_limit_backend(cls, v: str) -> str:
allowed = {"memory", "redis"}
v_lower = v.lower()
if v_lower not in allowed:
raise ValueError(f"rate_limit_backend must be one of {allowed}")
return v_lower
@field_validator("log_level")
@classmethod
def _validate_log_level(cls, v: str) -> str:
allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
v_upper = v.upper()
if v_upper not in allowed:
raise ValueError(f"log_level must be one of {allowed}")
return v_upper
@model_validator(mode="after")
def _enforce_auth_posture(self) -> AegisSettings:
"""Validate backend URL and refuse authentication disablement in production.
The ``debug_mode`` field documents that auth cannot be silently disabled
in production. This makes that promise enforceable: ``auth_disabled`` is
only honoured when ``debug_mode`` is also set. Without this check a stray
``AEGIS_AUTH_DISABLED=true`` in a production environment would open both
the proxy and the ``/v1/audit/*`` endpoints with no credential check —
a silent, config-only privilege escalation that no code path guards.
"""
parsed_backend = urlparse(str(self.backend_url))
if parsed_backend.scheme not in {"http", "https"} or not parsed_backend.hostname:
raise ValueError("backend_url must be an absolute http/https URL with a hostname")
if parsed_backend.username is not None or parsed_backend.password is not None:
raise ValueError("backend_url must not contain URL userinfo")
if self.auth_disabled and not self.debug_mode:
raise ValueError(
"auth_disabled=True requires debug_mode=True. Refusing to start "
"with authentication disabled outside debug mode. Set "
"AEGIS_DEBUG_MODE=true for local development, or remove "
"AEGIS_AUTH_DISABLED and configure AEGIS_API_KEYS for production."
)
if self.auth_disabled and "host" not in self.model_fields_set and self.host == "0.0.0.0":
# REG-D75: an unauthenticated gateway holding the operator's backend
# key used to listen on every interface by default. With auth
# disabled and no explicit host, it now binds loopback only; an
# explicit AEGIS_HOST (the evaluation compose file sets 0.0.0.0 inside
# its container and publishes on host loopback) is respected.
self.host = "127.0.0.1"
if self.max_stream_event_bytes > self.stream_queue_max_bytes:
raise ValueError("max_stream_event_bytes must not exceed stream_queue_max_bytes")
if self.rate_limit_default_output_tokens > self.rate_limit_max_output_tokens:
raise ValueError(
"rate_limit_default_output_tokens must not exceed rate_limit_max_output_tokens"
)
if self.auth_mode in {"oidc", "oidc_mtls"}:
for name, value in (
("oidc_issuer", self.oidc_issuer),
("oidc_audience", self.oidc_audience),
("oidc_jwks_url", self.oidc_jwks_url),
):
if not value:
raise ValueError(f"auth_mode={self.auth_mode!r} requires {name}")
if self.auth_mode in {"mtls", "api_key_mtls", "oidc_mtls"}:
if not self.get_mtls_fingerprints() or not self.get_mtls_san_allowlist():
raise ValueError("mTLS auth mode requires fingerprint and SAN allowlists")
direct_tls = self.mtls_required and self.ssl_ca_certs is not None
if not self.mtls_trusted_proxy_cidrs and not direct_tls:
raise ValueError(
"mTLS auth mode requires verified direct TLS or trusted proxy CIDRs"
)
if self.s3_archive_enabled and not self.s3_archive_bucket:
raise ValueError("s3_archive_enabled=True requires s3_archive_bucket")
if self.s3_archive_enabled and self.max_wal_bytes == 0:
raise ValueError("s3 archival requires max_wal_bytes > 0 to finalize segments")
if self.tsa_url and self.tsa_ca_file is None:
raise ValueError("tsa_url requires tsa_ca_file for explicit trust-policy verification")
if self.siem_url:
parsed_siem = urlparse(self.siem_url)
if parsed_siem.scheme != "https" or not parsed_siem.hostname:
raise ValueError("siem_url must be an absolute HTTPS URL")
return self
def validate_runtime_invariants(self) -> None:
"""Raise before binding sockets when strict runtime invariants are absent."""
if self.require_distributed_limiter and self.rate_limit_backend != "redis":
raise ValueError("require_distributed_limiter=True requires rate_limit_backend='redis'")
self.validate_ha()
if self.security_enforcement_mode != "strict":
return
if self.debug_mode:
raise ValueError("strict runtime cannot enable debug_mode")
if self.auth_disabled:
raise ValueError("strict runtime cannot disable authentication")
if not self.require_durable_evidence:
raise ValueError("strict runtime requires require_durable_evidence=True")
if self.rate_limit_backend != "redis":
raise ValueError("strict runtime requires rate_limit_backend='redis'")
if self.auth_mode in {"api_key", "api_key_mtls"} and not self.get_api_keys():
raise ValueError("strict runtime requires at least one AEGIS_API_KEYS value")
if not self.signing_key and not self.pkcs11_library_path:
raise ValueError("strict runtime requires signing_key or pkcs11_library_path")
if self.mtls_required and not self.ssl_ca_certs:
raise ValueError("mtls_required=True requires ssl_ca_certs")
if len(self.auth_identity_hmac_key.encode("utf-8")) < 32:
raise ValueError("strict runtime requires auth_identity_hmac_key of at least 32 bytes")
if self.auth_mode in {"api_key", "api_key_mtls"}:
configured = self.get_api_key_principals()
missing = sorted(
self.api_key_principal_digest(key)
for key in self.get_api_keys()
if self.api_key_principal_digest(key) not in configured
)
if missing:
raise ValueError(
"strict API-key mode requires an explicit principal mapping per key"
)
def validate_ha(self) -> None:
if self.ha_mode not in {"single", "active_passive", "active_active"}:
raise ValueError("ha_mode must be 'single', 'active_passive' or 'active_active'")
if self.ha_mode == "single":
return
if self.workers != 1:
raise ValueError("ha_mode requires workers=1: each replica is one writer process")
if not (self.ha_redis_url or self.redis_url):
raise ValueError("ha_mode requires ha_redis_url or redis_url for the writer lease")
if self.ha_mode == "active_passive" and not self.ha_chain_id:
raise ValueError("ha_mode='active_passive' requires ha_chain_id (the shared chain)")
if self.ha_mode == "active_active" and not self.ha_sequencer_url:
raise ValueError("ha_mode='active_active' requires ha_sequencer_url")
url = self.ha_sequencer_url
if url and not url.startswith(("postgresql://", "postgres://", "sqlite:///")):
raise ValueError("ha_sequencer_url must be postgresql://… or sqlite:///absolute/path")
if url.startswith("sqlite") and self.security_enforcement_mode == "strict":
# SQLite's file locking is not safe on the network filesystems replicas
# would share, and its I/O path is outside the strict seccomp allowlist.
raise ValueError("strict runtime requires a postgresql:// ha_sequencer_url")
def get_api_keys(self) -> frozenset[str]:
if not self.api_keys:
return frozenset()
return frozenset(k.strip() for k in self.api_keys.split(",") if k.strip())
def get_audit_api_keys(self) -> frozenset[str]:
if not self.audit_api_keys:
return self.get_api_keys()
return frozenset(k.strip() for k in self.audit_api_keys.split(",") if k.strip())
def api_key_principal_digest(self, key: str) -> str:
"""Return the domain-separated keyed lookup digest for an API key."""
return hmac.new(
self.auth_identity_hmac_key.encode("utf-8"),
b"aegis-api-key-principal-v1\x00" + key.encode("utf-8"),
hashlib.sha256,
).hexdigest()
def get_api_key_principals(self) -> dict[str, dict[str, object]]:
if not self.api_key_principals_json:
return {}
try:
value = json.loads(self.api_key_principals_json)
except json.JSONDecodeError as exc:
raise ValueError("api_key_principals_json must be valid JSON") from exc
if not isinstance(value, dict):
raise ValueError("api_key_principals_json must be a JSON object")
result: dict[str, dict[str, object]] = {}
for digest, record in value.items():
if (
not isinstance(digest, str)
or len(digest) != 64
or any(character not in "0123456789abcdef" for character in digest)
or not isinstance(record, dict)
):
raise ValueError(
"API-key principal mappings require HMAC-SHA256 keys and object values"
)
result[digest] = dict(record)
return result
def get_oidc_algorithms(self) -> tuple[str, ...]:
values = tuple(value.strip() for value in self.oidc_algorithms.split(",") if value.strip())
if not values:
raise ValueError("oidc_algorithms must contain at least one algorithm")
return values
def get_mtls_fingerprints(self) -> frozenset[str]:
return frozenset(
value.strip()
for value in self.mtls_allowed_sha256_fingerprints.split(",")
if value.strip()
)
def get_mtls_san_allowlist(self) -> frozenset[str]:
return frozenset(
value.strip() for value in self.mtls_san_allowlist.split(",") if value.strip()
)
def get_mtls_proxy_cidrs(self) -> tuple[str, ...]:
return tuple(
value.strip() for value in self.mtls_trusted_proxy_cidrs.split(",") if value.strip()
)
def get_cors_origins(self) -> list[str]:
if not self.cors_origins:
return []
return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
def get_dmz_networks(self) -> list[Any]:
"""Parse AEGIS_DMZ_ALLOWED_SOURCE_IPS into ipaddress network objects."""
import ipaddress
if not self.dmz_allowed_source_ips:
return []
networks = []
for entry in self.dmz_allowed_source_ips.split(","):
entry = entry.strip()
if not entry:
continue
try:
# Accept bare IPs (treat as /32 or /128) or CIDR notation
networks.append(ipaddress.ip_network(entry, strict=False))
except ValueError as exc:
raise ValueError(
f"AEGIS_DMZ_ALLOWED_SOURCE_IPS: invalid address or network {entry!r}: {exc}"
) from exc
return networks
def get_egress_guard(self) -> Any:
"""Build an EgressGuard from airgap_mode + airgap_allowed_hosts."""
from aegis.proxy.egress_guard import build_egress_guard
return build_egress_guard(
airgap_mode=self.airgap_mode,
allowed_hosts_csv=self.airgap_allowed_hosts,
upstream_url=self.backend_url_str,
)
@property
def backend_url_str(self) -> str:
return str(self.backend_url).rstrip("/")
@lru_cache(maxsize=1)
def get_settings() -> AegisSettings:
return AegisSettings()