diff --git a/b2/_internal/console_tool.py b/b2/_internal/console_tool.py index 889d26ec5..5c7410e14 100644 --- a/b2/_internal/console_tool.py +++ b/b2/_internal/console_tool.py @@ -4591,6 +4591,18 @@ def _run(self, args): class BucketNotificationRuleCreateBase( JSONOptionMixin, BucketNotificationRuleWarningMixin, Command ): + FORBID_LOGGING_ARGUMENTS = True + + def run(self, args): + # SDK bucket DEBUG call traces include signing secrets and custom header values. + sdk_logger = logging.getLogger('b2sdk._internal.bucket') + previous_level = sdk_logger.level + sdk_logger.setLevel(max(logging.INFO, sdk_logger.getEffectiveLevel())) + try: + return super().run(args) + finally: + sdk_logger.setLevel(previous_level) + @classmethod def _validate_secret(cls, value: str) -> str: if not re.match(r'^[a-zA-Z0-9]{32}$', value): diff --git a/changelog.d/+notification-rule-no-arg-logging.fixed.md b/changelog.d/+notification-rule-no-arg-logging.fixed.md new file mode 100644 index 000000000..f96a4ab3b --- /dev/null +++ b/changelog.d/+notification-rule-no-arg-logging.fixed.md @@ -0,0 +1 @@ +The `bucket notification-rule create` and `bucket notification-rule update` commands no longer log signing secrets or custom header values in command arguments or SDK call traces. diff --git a/test/unit/console_tool/test_notification_rules.py b/test/unit/console_tool/test_notification_rules.py index 83d017107..4af8c44b0 100644 --- a/test/unit/console_tool/test_notification_rules.py +++ b/test/unit/console_tool/test_notification_rules.py @@ -8,9 +8,12 @@ # ###################################################################### import json +import logging import pytest +from b2._internal.console_tool import CommandError + @pytest.fixture() def bucket_notification_rule(b2_cli, bucket): @@ -279,3 +282,100 @@ def test_notification_rules__sign_secret(b2_cli, bucket, bucket_notification_rul [*command, 'list', '--json', f'b2://{bucket}'], )[1] ) == [bucket_notification_rule] + + +@pytest.mark.parametrize('command', [['bucket', 'notification-rule'], ['notification-rules']]) +@pytest.mark.parametrize('subcommand', ['create', 'update']) +@pytest.mark.parametrize( + 'logging_flags', [['--verbose'], ['--debug-logs'], ['--verbose', '--debug-logs']] +) +def test_notification_rules__sensitive_arguments_are_not_logged( + b2_cli, + bucket, + bucket_notification_rule, + b2_cli_log_fix, + cwd_path, + caplog, + capsys, + command, + subcommand, + logging_flags, +): + # Deliberately synthetic values, not credentials. + sign_secret = 'a' * 32 + header_value = 'test-only-header-value' + caplog.set_level(logging.DEBUG) + caplog.set_level(logging.DEBUG, logger='b2sdk._internal.bucket') + caplog.clear() + capsys.readouterr() + original_handlers = logging.root.handlers[:] + + try: + b2_cli.run( + [ + *command, + subcommand, + *logging_flags, + f'b2://{bucket}', + 'new-rule' if subcommand == 'create' else bucket_notification_rule['name'], + '--event-type', + 'b2:ObjectCreated:*', + '--webhook-url', + 'https://example.com/webhook', + '--sign-secret', + sign_secret, + '--custom-header', + f'X-Test-Header={header_value}', + ] + ) + + outputs = [caplog.text, capsys.readouterr().err] + if '--debug-logs' in logging_flags: + outputs.append((cwd_path / 'b2_cli.log').read_text()) + for output in outputs: + assert sign_secret not in output + assert header_value not in output + assert 'arguments hidden' in caplog.text + + caplog.clear() + control_args = ['bucket', 'create', *logging_flags, 'control-log-bucket', 'allPrivate'] + b2_cli.run(control_args) + assert f"with arguments: {['b2', *control_args]}" in caplog.text + assert 'calling Bucket(' in caplog.text + finally: + for handler in logging.root.handlers[:]: + if handler not in original_handlers: + logging.root.removeHandler(handler) + handler.close() + + +@pytest.mark.parametrize('sdk_log_level', [logging.DEBUG, logging.WARNING]) +def test_notification_rules__sdk_logging_restored_after_error( + b2_cli, bucket, bucket_notification_rule, caplog, monkeypatch, sdk_log_level +): + sdk_logger = logging.getLogger('b2sdk._internal.bucket') + caplog.set_level(logging.DEBUG) + caplog.set_level(sdk_log_level, logger=sdk_logger.name) + + def reject_rules(*args, **kwargs): + sdk_logger.debug('test-only SDK request trace') + sdk_logger.warning('test-only SDK warning') + raise CommandError('test-only request failure') + + monkeypatch.setattr(b2_cli.raw_api, 'set_bucket_notification_rules', reject_rules) + b2_cli.run( + [ + 'bucket', + 'notification-rule', + 'update', + f'b2://{bucket}', + bucket_notification_rule['name'], + '--disable', + ], + expected_status=1, + expected_stderr='ERROR: test-only request failure\n', + ) + + assert 'test-only SDK request trace' not in caplog.text + assert 'test-only SDK warning' in caplog.text + assert sdk_logger.level == sdk_log_level