qonto-assistant/src/qonto_assistant/config.py

97 lines
4.4 KiB
Python
Raw Normal View History

from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]
@dataclass(frozen=True, slots=True)
class Settings:
service_name: str
default_tenant_id: str
default_actor_lane: str
required_scope: str
enforce_scope: bool
policy_file: Path
qonto_base_url: str
qonto_fixture_dir: Path | None
qonto_auth_mode: str
qonto_organization_path: str
qonto_transactions_path: str
qonto_timeout_seconds: float
qonto_max_retries: int
qonto_secret_ttl_seconds: int
rate_limit_requests: int
rate_limit_window_seconds: int
max_concurrency: int
deny_escalation_enabled: bool
deny_escalation_threshold: int
deny_escalation_window_seconds: int
deny_escalation_lockout_seconds: int
QONTO-WP-0004-T03: verify key-cape IAM Profile tokens Replaces the interim shared-secret bearer token's role as the identity boundary with real key-cape JWKS-based verification, closing the gap docs/mcp-integration.md called out explicitly ("no OIDC issuer exists in this fleet yet") -- key-cape's /jwks is a standard RS256 endpoint and needed no key-cape-side work to consume. KeyCapeTokenVerifier fetches and caches signing keys over httpx (matching FlexAuthCheckClient's pattern elsewhere in this codebase), validates iss/aud/exp and the IAM Profile v0.3 required claims, and derives ActorClaims from the token (tenant, scopes, and a lane inferred from the roles claim). Wired into auth.actor_claims_from_headers, the single seam both REST and MCP already used -- a verified bearer token now takes precedence over self-asserted X-Actor-* headers, and can be made mandatory via QONTO_KEY_CAPE_REQUIRED once real tokens are issued to callers. Off by default (no QONTO_KEY_CAPE_JWKS_URL set) so existing deployments are unaffected until configured. The QONTO_ASSISTANT_MCP_TOKEN shared secret remains as a documented local-dev/legacy fallback, not the auth boundary going forward. Verified: 13 new tests (test_key_cape_auth.py) covering valid/expired/ wrong-audience/wrong-issuer/missing-claim/unknown-key/rotated-key tokens plus the auth.py precedence and required-vs-optional integration paths, using a real generated RSA keypair and JWKS served over httpx.MockTransport. Full suite -> 52 passed; REST and MCP smokes both still pass against fixtures; compileall clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-24 00:10:41 +02:00
key_cape_jwks_url: str | None
key_cape_issuer: str
key_cape_audience: str
key_cape_required: bool
key_cape_cache_seconds: float
key_cape_timeout_seconds: float
credential_source: str
openbao_path: str
openbao_command: str
openbao_timeout_seconds: float
mcp_auth_token: str | None
host: str
port: int
@classmethod
def from_env(cls) -> "Settings":
policy_file = Path(
os.getenv(
"QONTO_ASSISTANT_POLICY_FILE",
str(_repo_root() / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"),
)
)
return cls(
service_name=os.getenv("QONTO_ASSISTANT_SERVICE_NAME", "qonto-assistant"),
default_tenant_id=os.getenv("QONTO_ASSISTANT_DEFAULT_TENANT", "binky"),
default_actor_lane=os.getenv("QONTO_ASSISTANT_DEFAULT_LANE", "green"),
required_scope=os.getenv("QONTO_ASSISTANT_REQUIRED_SCOPE", "finance.qonto.read"),
enforce_scope=os.getenv("QONTO_ASSISTANT_ENFORCE_SCOPE", "false").lower() == "true",
policy_file=policy_file,
qonto_base_url=os.getenv("QONTO_BASE_URL", "https://thirdparty.qonto.com"),
qonto_fixture_dir=(
Path(os.environ["QONTO_FIXTURE_DIR"]).resolve()
if os.getenv("QONTO_FIXTURE_DIR")
else None
),
qonto_auth_mode=os.getenv("QONTO_AUTH_MODE", "legacy_api_key"),
qonto_organization_path=os.getenv("QONTO_ORGANIZATION_PATH", "/v2/organization"),
qonto_transactions_path=os.getenv("QONTO_TRANSACTIONS_PATH", "/v2/transactions"),
qonto_timeout_seconds=float(os.getenv("QONTO_TIMEOUT_SECONDS", "10")),
qonto_max_retries=int(os.getenv("QONTO_MAX_RETRIES", "1")),
qonto_secret_ttl_seconds=int(os.getenv("QONTO_SECRET_TTL_SECONDS", "300")),
rate_limit_requests=int(os.getenv("QONTO_RATE_LIMIT_REQUESTS", "20")),
rate_limit_window_seconds=int(os.getenv("QONTO_RATE_LIMIT_WINDOW_SECONDS", "60")),
max_concurrency=int(os.getenv("QONTO_MAX_CONCURRENCY", "4")),
deny_escalation_enabled=os.getenv("QONTO_DENY_ESCALATION_ENABLED", "true").lower() == "true",
deny_escalation_threshold=int(os.getenv("QONTO_DENY_ESCALATION_THRESHOLD", "3")),
deny_escalation_window_seconds=int(os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60")),
deny_escalation_lockout_seconds=int(os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300")),
QONTO-WP-0004-T03: verify key-cape IAM Profile tokens Replaces the interim shared-secret bearer token's role as the identity boundary with real key-cape JWKS-based verification, closing the gap docs/mcp-integration.md called out explicitly ("no OIDC issuer exists in this fleet yet") -- key-cape's /jwks is a standard RS256 endpoint and needed no key-cape-side work to consume. KeyCapeTokenVerifier fetches and caches signing keys over httpx (matching FlexAuthCheckClient's pattern elsewhere in this codebase), validates iss/aud/exp and the IAM Profile v0.3 required claims, and derives ActorClaims from the token (tenant, scopes, and a lane inferred from the roles claim). Wired into auth.actor_claims_from_headers, the single seam both REST and MCP already used -- a verified bearer token now takes precedence over self-asserted X-Actor-* headers, and can be made mandatory via QONTO_KEY_CAPE_REQUIRED once real tokens are issued to callers. Off by default (no QONTO_KEY_CAPE_JWKS_URL set) so existing deployments are unaffected until configured. The QONTO_ASSISTANT_MCP_TOKEN shared secret remains as a documented local-dev/legacy fallback, not the auth boundary going forward. Verified: 13 new tests (test_key_cape_auth.py) covering valid/expired/ wrong-audience/wrong-issuer/missing-claim/unknown-key/rotated-key tokens plus the auth.py precedence and required-vs-optional integration paths, using a real generated RSA keypair and JWKS served over httpx.MockTransport. Full suite -> 52 passed; REST and MCP smokes both still pass against fixtures; compileall clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-24 00:10:41 +02:00
key_cape_jwks_url=os.getenv("QONTO_KEY_CAPE_JWKS_URL") or None,
key_cape_issuer=os.getenv("QONTO_KEY_CAPE_ISSUER", "https://key-cape.netkingdom"),
key_cape_audience=os.getenv("QONTO_KEY_CAPE_AUDIENCE", "qonto-assistant"),
key_cape_required=os.getenv("QONTO_KEY_CAPE_REQUIRED", "false").lower() == "true",
key_cape_cache_seconds=float(os.getenv("QONTO_KEY_CAPE_CACHE_SECONDS", "300")),
key_cape_timeout_seconds=float(os.getenv("QONTO_KEY_CAPE_TIMEOUT_SECONDS", "5")),
credential_source=os.getenv("QONTO_CREDENTIAL_SOURCE", "env"),
openbao_path=os.getenv("QONTO_OPENBAO_PATH", "tenants/binky/qonto-api"),
openbao_command=os.getenv("QONTO_OPENBAO_COMMAND", "bao"),
openbao_timeout_seconds=float(os.getenv("QONTO_OPENBAO_TIMEOUT_SECONDS", "5")),
mcp_auth_token=os.getenv("QONTO_ASSISTANT_MCP_TOKEN") or None,
host=os.getenv("QONTO_ASSISTANT_HOST", "127.0.0.1"),
port=int(os.getenv("QONTO_ASSISTANT_PORT", "8080")),
)