🔒️(summary) refactor configuration secrets to use Pydantic SecretStr

Replace plain string fields with Pydantic SecretStr class for all
sensitive configuration values in FastAPI settings to prevent accidental
exposure in logs, error messages, or debugging output, following
security best practices for credential handling.
This commit is contained in:
lebaudantoine
2025-12-11 15:56:27 +01:00
committed by aleb_the_flash
parent 43f3e4691b
commit 4256eb403d
3 changed files with 13 additions and 10 deletions

View File

@@ -93,7 +93,8 @@ class LLMService:
def __init__(self): def __init__(self):
"""Init the LLMService once.""" """Init the LLMService once."""
self._client = openai.OpenAI( self._client = openai.OpenAI(
base_url=settings.llm_base_url, api_key=settings.llm_api_key base_url=settings.llm_base_url,
api_key=settings.llm_api_key.get_secret_value(),
) )
def call( def call(
@@ -148,7 +149,9 @@ def format_actions(llm_output: dict) -> str:
def post_with_retries(url, data): def post_with_retries(url, data):
"""Send POST request with automatic retries.""" """Send POST request with automatic retries."""
session = create_retry_session() session = create_retry_session()
session.headers.update({"Authorization": f"Bearer {settings.webhook_api_token}"}) session.headers.update(
{"Authorization": f"Bearer {settings.webhook_api_token.get_secret_value()}"}
)
try: try:
response = session.post(url, json=data) response = session.post(url, json=data)
response.raise_for_status() response.raise_for_status()
@@ -195,7 +198,7 @@ def process_audio_transcribe_summarize_v2(
minio_client = Minio( minio_client = Minio(
settings.aws_s3_endpoint_url, settings.aws_s3_endpoint_url,
access_key=settings.aws_s3_access_key_id, access_key=settings.aws_s3_access_key_id,
secret_key=settings.aws_s3_secret_access_key, secret_key=settings.aws_s3_secret_access_key.get_secret_value(),
secure=settings.aws_s3_secure_access, secure=settings.aws_s3_secure_access,
) )
@@ -226,7 +229,7 @@ def process_audio_transcribe_summarize_v2(
logger.info("Initiating WhisperX client") logger.info("Initiating WhisperX client")
whisperx_client = openai.OpenAI( whisperx_client = openai.OpenAI(
api_key=settings.whisperx_api_key, api_key=settings.whisperx_api_key.get_secret_value(),
base_url=settings.whisperx_base_url, base_url=settings.whisperx_base_url,
max_retries=settings.whisperx_max_retries, max_retries=settings.whisperx_max_retries,
) )

View File

@@ -15,7 +15,7 @@ class Settings(BaseSettings):
app_name: str = "app" app_name: str = "app"
app_api_v1_str: str = "/api/v1" app_api_v1_str: str = "/api/v1"
app_api_token: str app_api_token: SecretStr
# Audio recordings # Audio recordings
recording_max_duration: Optional[int] = None recording_max_duration: Optional[int] = None
@@ -32,18 +32,18 @@ class Settings(BaseSettings):
aws_storage_bucket_name: str aws_storage_bucket_name: str
aws_s3_endpoint_url: str aws_s3_endpoint_url: str
aws_s3_access_key_id: str aws_s3_access_key_id: str
aws_s3_secret_access_key: str aws_s3_secret_access_key: SecretStr
aws_s3_secure_access: bool = True aws_s3_secure_access: bool = True
# AI-related settings # AI-related settings
whisperx_api_key: str whisperx_api_key: SecretStr
whisperx_base_url: str = "https://api.openai.com/v1" whisperx_base_url: str = "https://api.openai.com/v1"
whisperx_asr_model: str = "whisper-1" whisperx_asr_model: str = "whisper-1"
whisperx_max_retries: int = 0 whisperx_max_retries: int = 0
# ISO 639-1 language code (e.g., "en", "fr", "es") # ISO 639-1 language code (e.g., "en", "fr", "es")
whisperx_default_language: Optional[str] = None whisperx_default_language: Optional[str] = None
llm_base_url: str llm_base_url: str
llm_api_key: str llm_api_key: SecretStr
llm_model: str llm_model: str
# Transcription processing # Transcription processing
@@ -54,7 +54,7 @@ class Settings(BaseSettings):
webhook_max_retries: int = 2 webhook_max_retries: int = 2
webhook_status_forcelist: List[int] = [502, 503, 504] webhook_status_forcelist: List[int] = [502, 503, 504]
webhook_backoff_factor: float = 0.1 webhook_backoff_factor: float = 0.1
webhook_api_token: str webhook_api_token: SecretStr
webhook_url: str webhook_url: str
# Output related settings # Output related settings

View File

@@ -14,6 +14,6 @@ def verify_token(
): ):
"""Verify the bearer token from the Authorization header.""" """Verify the bearer token from the Authorization header."""
token = credentials.credentials token = credentials.credentials
if token != settings.app_api_token: if token != settings.app_api_token.get_secret_value():
raise HTTPException(status_code=401, detail="Invalid token") raise HTTPException(status_code=401, detail="Invalid token")
return token return token