"""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)