"""Example: Parameter Logging with Sensitive Data Sanitization
Demonstrates how to use the @log_call decorator to log function parameters
while automatically redacting sensitive values like API keys, tokens, and secrets.
This is useful for debugging and auditing tool calls while maintaining security.
"""
import asyncio
from src.utils.logging import log_call, setup_correlation_logging
from src.utils.context import set_correlation_id
import logging
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(levelname)s: %(message)s'
)
# Enable correlation logging
setup_correlation_logging()
# Example 1: Tool with sensitive parameters
@log_call(
action_name="configure_webhook",
level_name="tool",
log_params=True,
sensitive_params=["webhook_secret", "api_key"]
)
async def configure_webhook(
url: str,
webhook_secret: str,
api_key: str | None = None,
timeout: int = 30
) -> dict:
"""Configure webhook endpoint with secret authentication.
Args:
url: Webhook endpoint URL
webhook_secret: Secret for webhook signature validation
api_key: Optional API key for additional authentication
timeout: Request timeout in seconds
Returns:
Configuration status
"""
# Simulate webhook configuration
await asyncio.sleep(0.1)
return {
"status": "configured",
"url": url,
"timeout": timeout
}
# Example 2: Model upload with credentials
@log_call(
action_name="upload_model",
level_name="tool",
log_params=True,
sensitive_params=["upload_token", "s3_secret_key", "password"]
)
async def upload_model(
model_path: str,
destination: str,
upload_token: str,
s3_secret_key: str | None = None,
password: str | None = None
) -> dict:
"""Upload model to remote storage with credentials.
Args:
model_path: Local path to model file
destination: Remote destination URL
upload_token: Authentication token for upload
s3_secret_key: Optional S3 secret key
password: Optional password for encrypted uploads
Returns:
Upload status
"""
# Simulate upload
await asyncio.sleep(0.2)
return {
"status": "uploaded",
"destination": destination,
"size_mb": 1234
}
# Example 3: External API call with bearer token
@log_call(
action_name="call_external_api",
level_name="client",
log_params=True,
sensitive_params=["bearer_token", "client_secret"]
)
async def call_external_api(
endpoint: str,
bearer_token: str,
client_secret: str | None = None,
method: str = "GET"
) -> dict:
"""Call external API with authentication.
Args:
endpoint: API endpoint URL
bearer_token: Bearer token for authorization
client_secret: Optional client secret for OAuth
method: HTTP method
Returns:
API response
"""
# Simulate API call
await asyncio.sleep(0.15)
return {
"status": "success",
"data": {"result": "some data"}
}
# Example 4: No sensitive parameters (all logged)
@log_call(
action_name="list_workflows",
level_name="tool",
log_params=True
)
async def list_workflows(
filter_by: str | None = None,
limit: int = 100
) -> dict:
"""List workflows - no sensitive parameters.
Args:
filter_by: Optional filter criteria
limit: Maximum number of results
Returns:
Workflow list
"""
await asyncio.sleep(0.05)
return {
"workflows": ["workflow1", "workflow2"],
"count": 2
}
# Example 5: Environment-based parameter logging
import os
DEBUG_LOGGING = os.getenv("DEBUG_LOGGING", "false").lower() == "true"
@log_call(
action_name="run_workflow",
level_name="tool",
log_params=DEBUG_LOGGING, # Only log in debug mode
sensitive_params=["auth_token"]
)
async def run_workflow(
workflow_id: str,
auth_token: str | None = None,
overrides: dict | None = None
) -> dict:
"""Run workflow with optional auth token.
Args:
workflow_id: Workflow identifier
auth_token: Optional authentication token
overrides: Parameter overrides
Returns:
Execution result
"""
await asyncio.sleep(0.3)
return {
"status": "completed",
"workflow_id": workflow_id
}
async def main():
"""Demonstrate parameter logging with sanitization."""
# Set correlation ID for this execution
set_correlation_id("demo-123")
print("=" * 70)
print("Parameter Logging with Sanitization Examples")
print("=" * 70)
# Example 1: Webhook configuration (secrets redacted)
print("\n1. Configure webhook (secrets will be [REDACTED]):")
result1 = await configure_webhook(
url="https://example.com/webhook",
webhook_secret="super-secret-key-12345", # nosec B106
api_key="sk-1234567890abcdef",
timeout=60
)
print(f" Result: {result1}\n")
# Example 2: Model upload (credentials redacted)
print("2. Upload model (credentials will be [REDACTED]):")
result2 = await upload_model(
model_path="/models/my_model.safetensors",
destination="s3://bucket/models/",
upload_token="tok_9876543210fedcba", # nosec B106
s3_secret_key="aws-secret-key-xyz",
password="encryption-password-123"
)
print(f" Result: {result2}\n")
# Example 3: External API (tokens redacted)
print("3. Call external API (tokens will be [REDACTED]):")
result3 = await call_external_api(
endpoint="https://api.example.com/data",
bearer_token="Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...", # nosec B106
client_secret="oauth-client-secret-abc",
method="POST"
)
print(f" Result: {result3}\n")
# Example 4: No sensitive data (all logged)
print("4. List workflows (all parameters logged):")
result4 = await list_workflows(
filter_by="image",
limit=50
)
print(f" Result: {result4}\n")
# Example 5: Environment-based logging
print(f"5. Run workflow (DEBUG_LOGGING={DEBUG_LOGGING}):")
result5 = await run_workflow(
workflow_id="generate_image",
auth_token="secret-token-abc123", # nosec B106
overrides={"steps": 30}
)
print(f" Result: {result5}\n")
print("=" * 70)
print("Notice how sensitive parameters show '[REDACTED]' in logs")
print("but the actual functions received the real values.")
print("=" * 70)
if __name__ == "__main__":
# Run the demo
asyncio.run(main())
print("\n\nTo enable debug logging for Example 5, run:")
print(" DEBUG_LOGGING=true python 04_parameter_sanitization.py")
