tegwick
b9349782f4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ec5-7e2b-7743-ac08-719e1b0f42e2
89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from collections.abc import Callable
|
|
|
|
from qonto_assistant.config import Settings
|
|
from qonto_assistant.contracts import QontoCredentials
|
|
from qonto_assistant.errors import CredentialError
|
|
|
|
|
|
class EnvironmentCredentialProvider:
|
|
def get_credentials(self) -> QontoCredentials:
|
|
api_key = os.getenv("API_KEY") or os.getenv("QONTO_API_KEY")
|
|
api_user = os.getenv("API_USER") or os.getenv("QONTO_ORGANIZATION_ID")
|
|
if not api_key or not api_user:
|
|
raise CredentialError(
|
|
"Missing Qonto credentials in environment (API_KEY/API_USER or QONTO_* vars)"
|
|
)
|
|
return QontoCredentials(api_user=api_user, api_key=api_key)
|
|
|
|
def invalidate(self) -> None:
|
|
return None
|
|
|
|
|
|
class OpenBaoCliCredentialProvider:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
command: str,
|
|
path: str,
|
|
timeout_seconds: float,
|
|
ttl_seconds: int,
|
|
runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
|
|
) -> None:
|
|
self.command = command
|
|
self.path = path
|
|
self.timeout_seconds = timeout_seconds
|
|
self.ttl_seconds = ttl_seconds
|
|
self.runner = runner
|
|
self._cached: tuple[float, QontoCredentials] | None = None
|
|
|
|
def get_credentials(self) -> QontoCredentials:
|
|
now = time.monotonic()
|
|
if self._cached is not None and now - self._cached[0] < self.ttl_seconds:
|
|
return self._cached[1]
|
|
|
|
api_key = self._read_field("API_KEY")
|
|
api_user = self._read_field("API_USER")
|
|
credentials = QontoCredentials(api_user=api_user, api_key=api_key)
|
|
self._cached = (now, credentials)
|
|
return credentials
|
|
|
|
def invalidate(self) -> None:
|
|
self._cached = None
|
|
|
|
def _read_field(self, field_name: str) -> str:
|
|
result = self.runner(
|
|
[self.command, "kv", "get", f"-field={field_name}", self.path],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=self.timeout_seconds,
|
|
)
|
|
if result.returncode != 0:
|
|
raise CredentialError(
|
|
f"OpenBao CLI failed while reading {field_name} from {self.path}: "
|
|
f"{result.stderr.strip() or result.stdout.strip() or 'unknown error'}"
|
|
)
|
|
value = result.stdout.strip()
|
|
if not value:
|
|
raise CredentialError(f"OpenBao returned an empty value for {field_name}")
|
|
return value
|
|
|
|
|
|
def build_credential_provider(
|
|
settings: Settings,
|
|
) -> EnvironmentCredentialProvider | OpenBaoCliCredentialProvider:
|
|
if settings.credential_source == "bao-cli":
|
|
return OpenBaoCliCredentialProvider(
|
|
command=settings.openbao_command,
|
|
path=settings.openbao_path,
|
|
timeout_seconds=settings.openbao_timeout_seconds,
|
|
ttl_seconds=settings.qonto_secret_ttl_seconds,
|
|
)
|
|
if settings.credential_source == "env":
|
|
return EnvironmentCredentialProvider()
|
|
raise CredentialError(f"Unsupported credential source: {settings.credential_source}")
|