khyari hamza
d815339c52
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)
321 lines
10 KiB
Python
321 lines
10 KiB
Python
"""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)
|