feat(security): add secrets masking and validation (H-21/SEC5)

Implement centralized redaction utility that masks API keys, tokens,
and credentials across CLI output, structlog logs, and error messages.

- Add shared/redaction.py with pattern-based secret detection (sk-*,
  sk-ant-*, tok_*, Bearer tokens), sensitive key name detection,
  database URL masking, custom pattern registration, and thread-safe
  global show_secrets flag
- Add config/logging.py with structlog configuration integrating the
  secrets_masking_processor into the processor chain
- Add --show-secrets global CLI option to reveal secrets when needed
- Redact error details in main.py, project.py, and auto_debug.py
  error handlers before printing
- Wrap format_output() in formatting.py with automatic dict redaction
- Add show_secrets field and safe __repr__ to Settings model
- Add 43-scenario Behave feature (features/security_secrets.feature)
- Add 10 Robot Framework smoke tests (robot/security_secrets.robot)
- Add ASV benchmarks (benchmarks/security_secrets_bench.py)
- Add reference docs (docs/reference/secrets_handling.md)
This commit is contained in:
khyari hamza
2026-02-17 16:01:40 +00:00
parent 359e3c4af2
commit d815339c52
13 changed files with 1256 additions and 12 deletions
+108
View File
@@ -0,0 +1,108 @@
"""ASV benchmarks for SEC5 secrets redaction throughput.
Measures the performance of the core redaction functions to ensure
masking overhead stays within acceptable bounds.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.shared.redaction import ( # noqa: E402
is_sensitive_key,
mask_database_url,
redact_dict,
redact_value,
secrets_masking_processor,
)
# ── Sample data ───────────────────────────────────────────────────
_SECRET_STRING = "My API key is sk-proj-abc123def456ghi789jkl012mno345"
_CLEAN_STRING = "This is a perfectly normal log message with no secrets"
_MIXED_DICT: dict[str, object] = {
"api_key": "sk-proj-abc123def456ghi789",
"username": "alice",
"password": "hunter2",
"config": {
"token": "tok_01HXYZ1234567890ABCDEF",
"endpoint": "https://api.example.com",
},
"message": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWI",
}
_POSTGRES_URL = "postgresql://admin:s3cret@db.prod.internal:5432/mydb"
_SQLITE_URL = "sqlite:///data/cleveragents.db"
# ── Benchmarks ────────────────────────────────────────────────────
class RedactValue:
"""Benchmark ``redact_value`` for secret and clean strings."""
def time_redact_secret_string(self) -> None:
redact_value(_SECRET_STRING)
def time_redact_clean_string(self) -> None:
redact_value(_CLEAN_STRING)
def time_redact_empty_string(self) -> None:
redact_value("")
class RedactDict:
"""Benchmark ``redact_dict`` for nested dicts."""
def time_redact_mixed_dict(self) -> None:
redact_dict(_MIXED_DICT) # type: ignore[arg-type]
def time_redact_empty_dict(self) -> None:
redact_dict({})
class IsSensitiveKey:
"""Benchmark ``is_sensitive_key`` lookups."""
def time_sensitive_key(self) -> None:
is_sensitive_key("api_key")
def time_non_sensitive_key(self) -> None:
is_sensitive_key("username")
def time_false_positive_key(self) -> None:
is_sensitive_key("token_count")
class MaskDatabaseUrl:
"""Benchmark ``mask_database_url`` for various URL schemes."""
def time_mask_postgres_url(self) -> None:
mask_database_url(_POSTGRES_URL)
def time_mask_sqlite_url(self) -> None:
mask_database_url(_SQLITE_URL)
class StructlogProcessor:
"""Benchmark the structlog masking processor."""
def time_processor_with_secrets(self) -> None:
event = dict(_MIXED_DICT)
event["event"] = "Request failed with key sk-ant-api03-xyz123abc456"
secrets_masking_processor(None, "info", event)
def time_processor_clean_event(self) -> None:
event = {"event": "User logged in", "user": "alice", "status": "ok"}
secrets_masking_processor(None, "info", event)
+112
View File
@@ -0,0 +1,112 @@
# Secrets Handling (SEC5)
CleverAgents automatically masks API keys, tokens, and credentials in
CLI output, structured logs, and error messages. This document covers
the design and usage of the redaction subsystem.
## Overview
The `cleveragents.shared.redaction` module provides centralised,
pattern-based secret detection and masking. It is integrated into:
- **CLI output** via `format_output()` in `cleveragents.cli.formatting`
- **Error handlers** in `main.py`, `project.py`, and `auto_debug.py`
- **structlog processors** via `secrets_masking_processor`
- **Settings repr** via `Settings.__repr__`
## Detected patterns
| Pattern family | Regex summary | Example |
|---|---|---|
| OpenAI keys | `sk-(?:proj-)?[A-Za-z0-9_-]{10,}` | `sk-proj-abc123…` |
| Anthropic keys | `sk-ant-[A-Za-z0-9_-]{10,}` | `sk-ant-api03-xyz…` |
| Token IDs | `tok_[A-Za-z0-9]{10,}` | `tok_01HXYZ…` |
| Bearer tokens | `Bearer\s+[A-Za-z0-9._~+/=-]{20,}` | `Bearer eyJhb…` |
| Generic keys | `(?:key\|KEY)-[A-Za-z0-9]{20,}` | `KEY-abcdef…` |
Sensitive **key names** (dictionary keys, config fields) are also
detected by substring matching:
`api_key`, `apikey`, `password`, `passwd`, `secret`, `token`,
`credential`, `private_key`, `access_key`, `auth`
False-positive key names like `token_count`, `max_tokens`, etc. are
excluded.
## Replacement token
All detected secrets are replaced with `***REDACTED***`.
## Global `--show-secrets` flag
Pass `--show-secrets` to any CLI command to reveal secrets in that
invocation:
```bash
agents --show-secrets info
agents --show-secrets diagnostics
```
The flag sets a global thread-safe boolean that the redaction layer
checks. When `True`, all masking is bypassed.
Programmatic access:
```python
from cleveragents.shared.redaction import get_show_secrets, set_show_secrets
set_show_secrets(True) # reveal secrets
set_show_secrets(False) # re-enable masking (default)
```
## Custom patterns
Register additional secret patterns at runtime:
```python
from cleveragents.shared.redaction import register_pattern
register_pattern(r"myorg_[a-z0-9]{32}")
```
Patterns are compiled once and appended to the global list.
## Database URL masking
`mask_database_url()` replaces embedded passwords in connection
strings while preserving the host and database name:
```python
from cleveragents.shared.redaction import mask_database_url
mask_database_url("postgresql://user:pass@host/db")
# → "postgresql://user:***@host/db"
mask_database_url("sqlite:///data.db")
# → "sqlite:///data.db" (unchanged)
```
## structlog integration
The `secrets_masking_processor` is inserted into the structlog
processor chain by `configure_structlog()`:
```python
from cleveragents.config.logging import configure_structlog
configure_structlog(env="production", log_level="INFO")
```
Every log event is scanned for secret patterns before rendering.
## Settings safety
`Settings.__repr__()` masks any field whose name matches a sensitive
key pattern, so accidental `print(settings)` or debug output never
leaks credentials.
## Testing
- **Behave**: `features/security_secrets.feature` (43 scenarios)
- **Robot Framework**: `robot/security_secrets.robot` (10 smoke tests)
- **ASV benchmarks**: `benchmarks/security_secrets_bench.py`
+188
View File
@@ -0,0 +1,188 @@
Feature: SEC5 - Secrets masking and validation
As a security-conscious developer
I want secrets automatically masked in logs and CLI output
So that API keys and tokens are never leaked
# --- Core redaction utility ---
Scenario: Redact OpenAI API key pattern
Given a string containing "sk-proj-abc123def456ghi789"
When I redact the string
Then the redacted value should be "***REDACTED***"
Scenario: Redact Anthropic API key pattern
Given a string containing "sk-ant-api03-xyzzy1234567890abcdef"
When I redact the string
Then the redacted value should be "***REDACTED***"
Scenario: Redact token pattern
Given a string containing "tok_01HXYZ1234567890ABCDEF"
When I redact the string
Then the redacted value should be "***REDACTED***"
Scenario: Redact generic bearer token
Given a string containing "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJz"
When I redact the string
Then the redacted value should contain "***REDACTED***"
Scenario: Non-secret strings are not redacted
Given a string containing "hello world this is normal text"
When I redact the string
Then the redacted value should be "hello world this is normal text"
Scenario: Mixed text with embedded secret is partially redacted
Given a string containing "My key is sk-proj-abc123def456ghi789 and name is Bob"
When I redact the string
Then the redacted value should contain "***REDACTED***"
And the redacted value should contain "My key is"
And the redacted value should contain "and name is Bob"
Scenario: Empty string is returned unchanged
Given a string containing ""
When I redact the string
Then the redacted value should be empty
# --- Dict redaction ---
Scenario: Dict with sensitive key names are masked
Given a dict with key "api_key" and value "sk-proj-abc123"
When I redact the dict
Then the key "api_key" should have value "***REDACTED***"
Scenario: Dict with password key is masked
Given a dict with key "password" and value "mysecretpass"
When I redact the dict
Then the key "password" should have value "***REDACTED***"
Scenario: Dict with secret key is masked
Given a dict with key "secret" and value "topsecretvalue"
When I redact the dict
Then the key "secret" should have value "***REDACTED***"
Scenario: Dict with token key is masked
Given a dict with key "token" and value "sometoken123"
When I redact the dict
Then the key "token" should have value "***REDACTED***"
Scenario: Dict with credential key is masked
Given a dict with key "credential" and value "cred-value"
When I redact the dict
Then the key "credential" should have value "***REDACTED***"
Scenario: Dict with non-sensitive key is not masked
Given a dict with key "username" and value "bob"
When I redact the dict
Then the key "username" should have value "bob"
Scenario: Nested dict values are redacted recursively
Given a nested dict with inner key "api_key" and value "sk-proj-nested"
When I redact the dict
Then the nested key "api_key" should have value "***REDACTED***"
Scenario: Dict with pattern-matched value is masked regardless of key name
Given a dict with key "config_value" and value "sk-ant-api03-somekey123"
When I redact the dict
Then the key "config_value" should have value "***REDACTED***"
Scenario: Dict redaction respects show_secrets flag
Given a dict with key "api_key" and value "sk-proj-abc123"
When I redact the dict with show_secrets enabled
Then the key "api_key" should have value "sk-proj-abc123"
# --- Sensitive key detection ---
Scenario Outline: Sensitive key names are detected
When I check if "<key>" is a sensitive key
Then the sensitivity check should be true
Examples:
| key |
| api_key |
| API_KEY |
| password |
| secret |
| token |
| credential |
| openai_api_key |
| ANTHROPIC_API_KEY |
| access_token |
| auth_token |
| private_key |
Scenario Outline: Non-sensitive key names are not flagged
When I check if "<key>" is a sensitive key
Then the sensitivity check should be false
Examples:
| key |
| username |
| email |
| name |
| description |
| project_id |
| token_count |
# --- Database URL masking ---
Scenario: SQLite URL is not masked
Given a database URL "sqlite:///path/to/db.sqlite"
When I mask the database URL
Then the masked URL should be "sqlite:///path/to/db.sqlite"
Scenario: PostgreSQL URL with credentials is masked
Given a database URL "postgresql://admin:s3cret@localhost:5432/mydb"
When I mask the database URL
Then the masked URL should contain "***"
And the masked URL should contain "localhost"
And the masked URL should not contain "s3cret"
# --- Global show_secrets flag ---
Scenario: Global show_secrets defaults to false
When I check the global show_secrets flag
Then the flag should be false
Scenario: Global show_secrets can be toggled
When I set the global show_secrets flag to true
Then the flag should be true
When I set the global show_secrets flag to false
Then the flag should be false
# --- structlog processor ---
Scenario: Structlog masking processor redacts event dict values
Given a structlog event dict with key "api_key" and value "sk-proj-secret123"
When the masking processor is applied
Then the event dict key "api_key" should be "***REDACTED***"
Scenario: Structlog masking processor redacts event message
Given a structlog event dict with event "Failed with key sk-ant-api03-xyz"
When the masking processor is applied
Then the event message should contain "***REDACTED***"
And the event message should not contain "sk-ant-api03-xyz"
Scenario: Structlog masking processor passes non-secret events
Given a structlog event dict with event "User logged in successfully"
When the masking processor is applied
Then the event message should be "User logged in successfully"
# --- CLI error detail redaction ---
Scenario: Error details with secret values are redacted
Given error details with key "api_key" and value "sk-proj-leaked"
When I redact the error details for CLI display
Then the redacted details key "api_key" should be "***REDACTED***"
Scenario: Error details with non-secret values are preserved
Given error details with key "plan_id" and value "01HXR123"
When I redact the error details for CLI display
Then the redacted details key "plan_id" should be "01HXR123"
# --- Secret pattern registration ---
Scenario: Custom secret pattern can be registered
Given the default secret patterns
When I register a custom pattern "custom_[a-z0-9]{16}"
And I redact the string "my custom_abcdef1234567890 key"
Then the redacted value should contain "***REDACTED***"
And the redacted value should not contain "custom_abcdef1234567890"
+320
View File
@@ -0,0 +1,320 @@
"""Step definitions for SEC5 - Secrets masking and validation.
Tests the centralized redaction utility, structlog masking processor,
database URL masking, the global show_secrets flag, CLI error detail
redaction, and custom pattern registration.
"""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from cleveragents.shared.redaction import (
get_show_secrets,
is_sensitive_key,
mask_database_url,
redact_dict,
redact_value,
register_pattern,
secrets_masking_processor,
set_show_secrets,
)
# ── Helpers ───────────────────────────────────────────────────────
def _reset_show_secrets() -> None:
"""Reset the global show_secrets flag to its default (False)."""
set_show_secrets(False)
# ── Core redaction: Given / When / Then ───────────────────────────
@given('a string containing "{value}"')
def step_given_string(context: Context, value: str) -> None:
context.input_string = value
@given('a string containing ""')
def step_given_empty_string(context: Context) -> None:
context.input_string = ""
@when("I redact the string")
def step_redact_string(context: Context) -> None:
context.redacted_result = redact_value(context.input_string)
@then('the redacted value should be "{expected}"')
def step_redacted_value_should_be(context: Context, expected: str) -> None:
assert context.redacted_result == expected, (
f"Expected '{expected}', got '{context.redacted_result}'"
)
@then("the redacted value should be empty")
def step_redacted_value_should_be_empty(context: Context) -> None:
assert context.redacted_result == "", (
f"Expected empty string, got '{context.redacted_result}'"
)
@then('the redacted value should contain "{substring}"')
def step_redacted_value_should_contain(context: Context, substring: str) -> None:
assert substring in context.redacted_result, (
f"Expected '{substring}' in '{context.redacted_result}'"
)
@then('the redacted value should not contain "{substring}"')
def step_redacted_value_should_not_contain(
context: Context,
substring: str,
) -> None:
assert substring not in context.redacted_result, (
f"Did not expect '{substring}' in '{context.redacted_result}'"
)
# ── Dict redaction ────────────────────────────────────────────────
@given('a dict with key "{key}" and value "{value}"')
def step_given_dict(context: Context, key: str, value: str) -> None:
context.input_dict = {key: value}
@given('a nested dict with inner key "{key}" and value "{value}"')
def step_given_nested_dict(context: Context, key: str, value: str) -> None:
context.input_dict = {"outer": {key: value}}
@when("I redact the dict")
def step_redact_dict(context: Context) -> None:
context.redacted_dict = redact_dict(context.input_dict)
@when("I redact the dict with show_secrets enabled")
def step_redact_dict_show_secrets(context: Context) -> None:
context.redacted_dict = redact_dict(context.input_dict, show_secrets=True)
@then('the key "{key}" should have value "{expected}"')
def step_key_should_have_value(context: Context, key: str, expected: str) -> None:
actual = context.redacted_dict[key]
assert actual == expected, f"Key '{key}': expected '{expected}', got '{actual}'"
@then('the nested key "{key}" should have value "{expected}"')
def step_nested_key_should_have_value(
context: Context,
key: str,
expected: str,
) -> None:
actual = context.redacted_dict["outer"][key]
assert actual == expected, (
f"Nested key '{key}': expected '{expected}', got '{actual}'"
)
# ── Sensitive key detection ───────────────────────────────────────
@when('I check if "{key}" is a sensitive key')
def step_check_sensitive_key(context: Context, key: str) -> None:
context.sensitive_result = is_sensitive_key(key)
@then("the sensitivity check should be true")
def step_sensitivity_true(context: Context) -> None:
assert context.sensitive_result is True, "Expected True, got False"
@then("the sensitivity check should be false")
def step_sensitivity_false(context: Context) -> None:
assert context.sensitive_result is False, "Expected False, got True"
# ── Database URL masking ─────────────────────────────────────────
@given('a database URL "{url}"')
def step_given_database_url(context: Context, url: str) -> None:
context.database_url = url
@when("I mask the database URL")
def step_mask_database_url(context: Context) -> None:
context.masked_url = mask_database_url(context.database_url)
@then('the masked URL should be "{expected}"')
def step_masked_url_should_be(context: Context, expected: str) -> None:
assert context.masked_url == expected, (
f"Expected '{expected}', got '{context.masked_url}'"
)
@then('the masked URL should contain "{substring}"')
def step_masked_url_should_contain(context: Context, substring: str) -> None:
assert substring in context.masked_url, (
f"Expected '{substring}' in '{context.masked_url}'"
)
@then('the masked URL should not contain "{substring}"')
def step_masked_url_should_not_contain(
context: Context,
substring: str,
) -> None:
assert substring not in context.masked_url, (
f"Did not expect '{substring}' in '{context.masked_url}'"
)
# ── Global show_secrets flag ─────────────────────────────────────
@when("I check the global show_secrets flag")
def step_check_show_secrets(context: Context) -> None:
# Ensure we start from clean state
_reset_show_secrets()
context._cleanup_handlers.append(_reset_show_secrets)
context.show_secrets_flag = get_show_secrets()
@when("I set the global show_secrets flag to true")
def step_set_show_secrets_true(context: Context) -> None:
set_show_secrets(True)
if _reset_show_secrets not in context._cleanup_handlers:
context._cleanup_handlers.append(_reset_show_secrets)
context.show_secrets_flag = get_show_secrets()
@when("I set the global show_secrets flag to false")
def step_set_show_secrets_false(context: Context) -> None:
set_show_secrets(False)
context.show_secrets_flag = get_show_secrets()
@then("the flag should be false")
def step_flag_false(context: Context) -> None:
assert context.show_secrets_flag is False, "Expected show_secrets to be False"
@then("the flag should be true")
def step_flag_true(context: Context) -> None:
assert context.show_secrets_flag is True, "Expected show_secrets to be True"
# ── structlog processor ──────────────────────────────────────────
@given('a structlog event dict with key "{key}" and value "{value}"')
def step_given_structlog_event_key(
context: Context,
key: str,
value: str,
) -> None:
# Ensure show_secrets is off for processor tests
_reset_show_secrets()
context._cleanup_handlers.append(_reset_show_secrets)
context.event_dict = {key: value}
@given('a structlog event dict with event "{event}"')
def step_given_structlog_event_msg(context: Context, event: str) -> None:
_reset_show_secrets()
context._cleanup_handlers.append(_reset_show_secrets)
context.event_dict = {"event": event}
@when("the masking processor is applied")
def step_apply_masking_processor(context: Context) -> None:
context.processed_event = secrets_masking_processor(
logger=None,
method_name="info",
event_dict=dict(context.event_dict),
)
@then('the event dict key "{key}" should be "{expected}"')
def step_event_key_should_be(context: Context, key: str, expected: str) -> None:
actual = context.processed_event[key]
assert actual == expected, (
f"Event key '{key}': expected '{expected}', got '{actual}'"
)
@then('the event message should contain "{substring}"')
def step_event_message_should_contain(
context: Context,
substring: str,
) -> None:
msg = context.processed_event["event"]
assert substring in msg, f"Expected '{substring}' in event message '{msg}'"
@then('the event message should not contain "{substring}"')
def step_event_message_should_not_contain(
context: Context,
substring: str,
) -> None:
msg = context.processed_event["event"]
assert substring not in msg, (
f"Did not expect '{substring}' in event message '{msg}'"
)
@then('the event message should be "{expected}"')
def step_event_message_should_be(context: Context, expected: str) -> None:
msg = context.processed_event["event"]
assert msg == expected, f"Expected event message '{expected}', got '{msg}'"
# ── CLI error detail redaction ────────────────────────────────────
@given('error details with key "{key}" and value "{value}"')
def step_given_error_details(context: Context, key: str, value: str) -> None:
context.error_details = {key: value}
@when("I redact the error details for CLI display")
def step_redact_error_details(context: Context) -> None:
context.redacted_details = redact_dict(context.error_details)
@then('the redacted details key "{key}" should be "{expected}"')
def step_redacted_details_key(
context: Context,
key: str,
expected: str,
) -> None:
actual = context.redacted_details[key]
assert actual == expected, (
f"Error details key '{key}': expected '{expected}', got '{actual}'"
)
# ── Secret pattern registration ──────────────────────────────────
@given("the default secret patterns")
def step_given_default_patterns(context: Context) -> None:
# Nothing to do — patterns are already loaded at module import.
# We store a marker so we know the Given ran.
context.patterns_initialized = True
@when('I register a custom pattern "{pattern}"')
def step_register_custom_pattern(context: Context, pattern: str) -> None:
register_pattern(pattern)
@when('I redact the string "{value}"')
def step_when_redact_string_inline(context: Context, value: str) -> None:
context.input_string = value
context.redacted_result = redact_value(value)
+74
View File
@@ -0,0 +1,74 @@
*** Settings ***
Documentation SEC5 - Secrets masking integration smoke tests
Library OperatingSystem
Library Collections
Library String
Library Process
Resource ${CURDIR}/common.resource
*** Variables ***
${SRC_DIR} ${CURDIR}/../src/cleveragents
*** Test Cases ***
Redaction Module Exists
[Documentation] Verify the redaction module is present
File Should Exist ${SRC_DIR}/shared/redaction.py
Logging Config Module Exists
[Documentation] Verify the structlog config module is present
File Should Exist ${SRC_DIR}/config/logging.py
Redaction Module Exports Expected Symbols
[Documentation] Verify redaction.py exports the public API
${content}= Get File ${SRC_DIR}/shared/redaction.py
Should Contain ${content} def redact_value
Should Contain ${content} def redact_dict
Should Contain ${content} def is_sensitive_key
Should Contain ${content} def mask_database_url
Should Contain ${content} def register_pattern
Should Contain ${content} def secrets_masking_processor
Should Contain ${content} def get_show_secrets
Should Contain ${content} def set_show_secrets
Should Contain ${content} REDACTED
Main CLI Declares Show Secrets Option
[Documentation] Verify --show-secrets global option is wired in
${content}= Get File ${SRC_DIR}/cli/main.py
Should Contain ${content} --show-secrets
Should Contain ${content} show_secrets_callback
CLI Error Handlers Use Redaction
[Documentation] Verify error handlers in main.py redact details
${content}= Get File ${SRC_DIR}/cli/main.py
Should Contain ${content} redact_value(e.message)
Should Contain ${content} redact_dict(e.details)
Formatting Module Applies Redaction
[Documentation] Verify format_output applies dict redaction
${content}= Get File ${SRC_DIR}/cli/formatting.py
Should Contain ${content} _redact_data
Should Contain ${content} safe_data
Settings Has Show Secrets Field
[Documentation] Verify Settings model includes show_secrets field
${content}= Get File ${SRC_DIR}/config/settings.py
Should Contain ${content} show_secrets: bool
Should Contain ${content} CLEVERAGENTS_SHOW_SECRETS
Settings Has Safe Repr
[Documentation] Verify Settings.__repr__ masks sensitive values
${content}= Get File ${SRC_DIR}/config/settings.py
Should Contain ${content} def __repr__
Should Contain ${content} is_sensitive_key
Redaction Module Does Not Use Eval Or Exec
[Documentation] Security check: no eval/exec in redaction module
${content}= Get File ${SRC_DIR}/shared/redaction.py
Should Not Contain ${content} eval(
Should Not Contain ${content} exec(
Logging Config Integrates Masking Processor
[Documentation] Verify structlog config uses secrets_masking_processor
${content}= Get File ${SRC_DIR}/config/logging.py
Should Contain ${content} secrets_masking_processor
Should Contain ${content} configure_structlog
+5 -2
View File
@@ -244,9 +244,12 @@ def run(
raise typer.Exit(1)
except PlanError as e:
console.print(f"[red]Plan Error:[/red] {e.message}")
from cleveragents.shared.redaction import redact_dict, redact_value
console.print(f"[red]Plan Error:[/red] {redact_value(e.message)}")
if e.details:
for key, value in e.details.items():
safe = redact_dict(e.details)
for key, value in safe.items():
console.print(f" [dim]{key}:[/dim] {value}")
raise typer.Abort() from e
except CleverAgentsError as e:
+5 -2
View File
@@ -401,9 +401,12 @@ def init(
default_filters=default_filters,
)
except ValidationError as e:
err_console.print(f"[red]Validation Error:[/red] {e.message}")
from cleveragents.shared.redaction import redact_dict, redact_value
err_console.print(f"[red]Validation Error:[/red] {redact_value(e.message)}")
if e.details:
for key, value in e.details.items():
safe = redact_dict(e.details)
for key, value in safe.items():
err_console.print(f" {key}: {value}")
raise typer.Abort() from e
except ConfigurationError as e:
+21 -5
View File
@@ -133,12 +133,27 @@ def _format_table(data: dict[str, Any] | list[dict[str, Any]]) -> str:
return buf.getvalue().rstrip("\n")
def _redact_data(
data: dict[str, Any] | list[dict[str, Any]],
) -> dict[str, Any] | list[dict[str, Any]]:
"""Apply secrets redaction to output data before rendering."""
from cleveragents.shared.redaction import redact_dict
if isinstance(data, list):
return [redact_dict(item) if isinstance(item, dict) else item for item in data]
return redact_dict(data)
def format_output(
data: dict[str, Any] | list[dict[str, Any]],
format_type: str,
) -> str:
"""Format *data* according to *format_type*.
All output is passed through :func:`redact_dict` before
rendering so that sensitive values are masked unless the
global ``show_secrets`` flag is enabled.
Parameters
----------
data:
@@ -153,14 +168,15 @@ def format_output(
the domain-specific Rich helper instead; this function returns
the JSON representation as a fallback.
"""
safe_data = _redact_data(data)
fmt = format_type.lower()
if fmt == OutputFormat.JSON.value:
return _format_json(data)
return _format_json(safe_data)
if fmt == OutputFormat.YAML.value:
return _format_yaml(data)
return _format_yaml(safe_data)
if fmt == OutputFormat.PLAIN.value:
return _format_plain(data)
return _format_plain(safe_data)
if fmt == OutputFormat.TABLE.value:
return _format_table(data)
return _format_table(safe_data)
# ``rich`` and any unknown value fall back to JSON
return _format_json(data)
return _format_json(safe_data)
+20 -2
View File
@@ -178,6 +178,14 @@ def version_callback(value: bool) -> None:
raise typer.Exit()
def show_secrets_callback(value: bool) -> None:
"""Handle --show-secrets flag by updating the global redaction state."""
if value:
from cleveragents.shared.redaction import set_show_secrets
set_show_secrets(True)
@app.callback()
def main_callback(
version: bool = typer.Option(
@@ -187,6 +195,13 @@ def main_callback(
is_eager=True,
help="Show version",
),
show_secrets: bool = typer.Option(
False,
"--show-secrets",
callback=show_secrets_callback,
is_eager=True,
help="Reveal secrets in CLI output (default: masked)",
),
) -> None:
"""CleverAgents - AI-powered development assistant."""
_register_subcommands()
@@ -540,11 +555,14 @@ def main(args: list[str] | None = None) -> int:
except typer.Abort:
return 1
except CleverAgentsError as e:
from cleveragents.shared.redaction import redact_dict, redact_value
err_console = get_err_console()
err_console.print(f"[red]Error:[/red] {e.message}")
err_console.print(f"[red]Error:[/red] {redact_value(e.message)}")
if e.details:
safe_details = redact_dict(e.details)
err_console.print("[dim]Details:[/dim]")
for key, value in e.details.items():
for key, value in safe_details.items():
err_console.print(f" {key}: {value}")
return 1
except KeyboardInterrupt:
+94
View File
@@ -0,0 +1,94 @@
"""Structured logging configuration with secrets masking.
Configures ``structlog`` with the :func:`secrets_masking_processor`
from :mod:`cleveragents.shared.redaction` inserted into the processor
chain so that every log event is automatically redacted before
rendering.
Usage::
from cleveragents.config.logging import configure_structlog, get_logger
configure_structlog(env="development", log_level="DEBUG")
logger = get_logger(__name__)
logger.info("connected", api_key="sk-proj-secret123")
# api_key will appear as ***REDACTED*** in output
"""
from __future__ import annotations
import logging
from typing import Any
import structlog
from cleveragents.shared.redaction import secrets_masking_processor
__all__ = ["configure_structlog", "get_logger"]
def configure_structlog(
*,
env: str = "development",
log_level: str = "INFO",
) -> None:
"""Configure structlog with secrets masking.
Args:
env: Runtime environment name. When ``"production"`` the JSON
renderer is used; otherwise the coloured console renderer.
log_level: Minimum log level (e.g. ``"DEBUG"``, ``"INFO"``).
Raises:
ValueError: If *log_level* is not a recognised level name.
"""
numeric_level = getattr(logging, log_level.upper(), None)
if not isinstance(numeric_level, int):
raise ValueError(f"Invalid log level: {log_level!r}")
logging.basicConfig(format="%(message)s", level=numeric_level, force=True)
renderer: structlog.types.Processor
if env.lower() == "production":
renderer = structlog.processors.JSONRenderer()
else:
renderer = structlog.dev.ConsoleRenderer()
shared_processors: list[structlog.types.Processor] = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso"),
secrets_masking_processor,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
]
structlog.configure(
processors=shared_processors,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
formatter = structlog.stdlib.ProcessorFormatter(
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
renderer,
],
)
root_logger = logging.getLogger()
for handler in root_logger.handlers:
handler.setFormatter(formatter)
def get_logger(name: str | None = None) -> Any:
"""Return a structlog bound logger.
Args:
name: Logger name (typically ``__name__``).
Returns:
A structlog ``BoundLogger`` instance.
"""
return structlog.get_logger(name)
+20
View File
@@ -284,6 +284,26 @@ class Settings(BaseSettings):
),
)
# Secrets visibility (SEC5)
show_secrets: bool = Field(
default=False,
validation_alias=AliasChoices("CLEVERAGENTS_SHOW_SECRETS"),
description="When True, secrets are shown in CLI output and logs.",
)
def __repr__(self) -> str:
"""Safe repr that masks sensitive field values."""
from cleveragents.shared.redaction import REDACTED, is_sensitive_key
parts: list[str] = []
for field_name in self.model_fields:
value = getattr(self, field_name, None)
if is_sensitive_key(field_name) and value is not None:
parts.append(f"{field_name}={REDACTED!r}")
else:
parts.append(f"{field_name}={value!r}")
return f"Settings({', '.join(parts)})"
def model_post_init(self, __context: Any) -> None: # type: ignore[override]
# pydantic v1 compatibility: BaseSettings may not define model_post_init
maybe_super = getattr(super(), "model_post_init", None)
+24 -1
View File
@@ -4,10 +4,33 @@ This module contains utilities and helpers that can be used
across all layers without violating architectural boundaries.
Includes:
- Secrets redaction and masking
- Logging utilities
- Metrics collection
- Validation helpers
- Date/time utilities
"""
__all__ = []
from cleveragents.shared.redaction import (
REDACTED,
get_show_secrets,
is_sensitive_key,
mask_database_url,
redact_dict,
redact_value,
register_pattern,
secrets_masking_processor,
set_show_secrets,
)
__all__ = [
"REDACTED",
"get_show_secrets",
"is_sensitive_key",
"mask_database_url",
"redact_dict",
"redact_value",
"register_pattern",
"secrets_masking_processor",
"set_show_secrets",
]
+265
View File
@@ -0,0 +1,265 @@
"""Centralized secrets redaction utility.
Provides pattern-based detection and masking of sensitive values
such as API keys, tokens, and credentials. Integrated into CLI
output, structlog processors, and error detail formatting.
Based on specification section "Secret Management" (line 27427):
- Patterns: ``sk-``, ``sk-ant-``, ``tok_``, bearer tokens
- Replacement: ``***REDACTED***``
- Applied at structlog processor level and CLI output
See Also:
- ``docs/reference/secrets_handling.md``
"""
from __future__ import annotations
import re
import threading
from collections.abc import MutableMapping
from typing import Any
REDACTED = "***REDACTED***"
# ── Sensitive key-name substrings ──────────────────────────────
_SENSITIVE_SUBSTRINGS: set[str] = {
"api_key",
"apikey",
"password",
"passwd",
"secret",
"token",
"credential",
"private_key",
"access_key",
"auth",
}
# Keys whose *entire* name (case-insensitive) is non-sensitive even
# though they contain a sensitive substring (e.g. "token_count").
_FALSE_POSITIVE_KEYS: set[str] = {
"token_count",
"token_limit",
"token_usage",
"max_tokens",
"total_tokens",
"prompt_tokens",
"completion_tokens",
"token_estimate",
"auth_method",
"auth_type",
"auth_enabled",
}
# ── Secret value patterns (compiled) ──────────────────────────
_SECRET_PATTERNS: list[re.Pattern[str]] = [
# OpenAI keys: sk-proj-..., sk-...
re.compile(r"sk-(?:proj-)?[A-Za-z0-9_-]{10,}"),
# Anthropic keys: sk-ant-api03-...
re.compile(r"sk-ant-[A-Za-z0-9_-]{10,}"),
# Token IDs: tok_...
re.compile(r"tok_[A-Za-z0-9]{10,}"),
# Bearer tokens
re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]{20,}"),
# Generic long hex/base64 keys (40+ chars)
re.compile(r"(?:key|KEY)-[A-Za-z0-9]{20,}"),
]
# Lock for thread-safe custom pattern registration
_patterns_lock = threading.Lock()
# ── Global show_secrets flag ──────────────────────────────────
_show_secrets: bool = False
_flag_lock = threading.Lock()
def get_show_secrets() -> bool:
"""Return the current global show_secrets flag."""
with _flag_lock:
return _show_secrets
def set_show_secrets(value: bool) -> None:
"""Set the global show_secrets flag.
Args:
value: Whether to reveal secrets in output.
Raises:
TypeError: If value is not a bool.
"""
if not isinstance(value, bool):
raise TypeError(f"show_secrets must be bool, got {type(value).__name__}")
global _show_secrets
with _flag_lock:
_show_secrets = value
# ── Public API ────────────────────────────────────────────────
def is_sensitive_key(key: str) -> bool:
"""Check whether a key name indicates a secret value.
Args:
key: The field or key name to check.
Returns:
True if the key name suggests a sensitive value.
"""
if not key:
return False
lower = key.lower()
if lower in _FALSE_POSITIVE_KEYS:
return False
return any(sub in lower for sub in _SENSITIVE_SUBSTRINGS)
def redact_value(value: str) -> str:
"""Scan a string for secret patterns and replace matches.
Args:
value: The string to scan and redact.
Returns:
The string with any detected secrets replaced by
``***REDACTED***``.
"""
if not value:
return value
result = value
with _patterns_lock:
patterns = list(_SECRET_PATTERNS)
for pattern in patterns:
result = pattern.sub(REDACTED, result)
return result
def redact_dict(
data: dict[str, Any],
*,
show_secrets: bool | None = None,
) -> dict[str, Any]:
"""Recursively redact sensitive keys and values in a dict.
Args:
data: The dictionary to redact.
show_secrets: Override for the global flag. When ``None``
(default), uses the global ``get_show_secrets()`` value.
Returns:
A new dict with sensitive values replaced by
``***REDACTED***``.
"""
if show_secrets is None:
show_secrets = get_show_secrets()
if show_secrets:
return dict(data)
return _redact_dict_inner(data)
def _redact_dict_inner(data: dict[str, Any]) -> dict[str, Any]:
"""Recursive implementation of dict redaction."""
result: dict[str, Any] = {}
for key, value in data.items():
if is_sensitive_key(key):
result[key] = REDACTED
elif isinstance(value, dict):
result[key] = _redact_dict_inner(value)
elif isinstance(value, str):
result[key] = redact_value(value)
elif isinstance(value, list):
result[key] = [
_redact_dict_inner(item)
if isinstance(item, dict)
else (redact_value(item) if isinstance(item, str) else item)
for item in value
]
else:
result[key] = value
return result
def mask_database_url(url: str) -> str:
"""Mask credentials in a database URL.
SQLite URLs (no credentials) are returned unchanged. URLs
with ``user:password@host`` patterns have the password
replaced.
Args:
url: A SQLAlchemy-style database URL.
Returns:
The URL with any embedded password masked.
"""
if not url:
return url
if url.startswith("sqlite"):
return url
# Pattern: scheme://user:password@host...
masked = re.sub(
r"(://[^:]+:)([^@]+)(@)",
r"\1***\3",
url,
)
return masked
def register_pattern(pattern: str) -> None:
"""Register a custom secret detection regex pattern.
Args:
pattern: A regular expression string to match secrets.
Raises:
ValueError: If the pattern is empty.
re.error: If the pattern is invalid regex.
"""
if not pattern:
raise ValueError("pattern cannot be empty")
compiled = re.compile(pattern)
with _patterns_lock:
_SECRET_PATTERNS.append(compiled)
# ── structlog processor ───────────────────────────────────────
def secrets_masking_processor(
logger: Any,
method_name: str,
event_dict: MutableMapping[str, Any],
) -> MutableMapping[str, Any]:
"""structlog processor that redacts secrets from log events.
Walks all values in the event dict and applies pattern-based
redaction. Intended to be inserted into the structlog
processor chain.
Args:
logger: The wrapped logger object (unused).
method_name: The name of the log method (unused).
event_dict: The structured log event dictionary.
Returns:
The event dict with secrets masked.
"""
if get_show_secrets():
return event_dict
result: dict[str, Any] = {}
for key, value in event_dict.items():
if is_sensitive_key(key):
result[key] = REDACTED
elif isinstance(value, str):
result[key] = redact_value(value)
elif isinstance(value, dict):
result[key] = _redact_dict_inner(value)
else:
result[key] = value
return result