Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions b2/_internal/console_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
1 change: 1 addition & 0 deletions changelog.d/+notification-rule-no-arg-logging.fixed.md
Original file line number Diff line number Diff line change
@@ -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.
100 changes: 100 additions & 0 deletions test/unit/console_tool/test_notification_rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
Loading