diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index 6952487f9..0142df92a 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -32,6 +32,7 @@ ENV_ACCESS_KEY_NAME, ENV_SECRET_KEY_NAME, INCOMING_DATE_FORMAT, + MAX_SIGNED_URL_EXPIRY, MULTIPART_UPLOAD_CHUNK_SIZE_DEFAULT, NO_ACCESS_ERROR, NO_SCOPES_ERROR, @@ -101,25 +102,45 @@ def generate_url(get_client, args, **kwargs): # pylint: disable=unused-argument "expiry", metavar="EXPIRY", type=str, - help="When this link should expire. Treated as an epoch " - "time if a number. If starts with a '+' treated as " - "an offset.", + help="When this link should expire. Treated as a Unix epoch " + "time in seconds if a number. If it starts with a '+', treated " + "as an offset in seconds from now. The link can be valid for " + f"at most {MAX_SIGNED_URL_EXPIRY} seconds (7 days).", ) - # TODO: - # Add docs for date format and unit for 'expiry' parameter. - parsed = parser.parse_args(args) - client = get_client() now = datetime.now() - if parsed.expiry.startswith("+"): - # this is an offset in seconds - offset = int(parsed.expiry[1:]) - else: - expiry = int(parsed.expiry) - offset = expiry - ceil(now.timestamp()) + try: + if parsed.expiry.startswith("+"): + # this is an offset in seconds + offset = int(parsed.expiry[1:]) + else: + expiry = int(parsed.expiry) + offset = expiry - ceil(now.timestamp()) + except ValueError: + print( + f"Invalid expiry {parsed.expiry!r}: expected an epoch time or " + "an offset like +3600, in seconds", + file=sys.stderr, + ) + sys.exit(ExitCodes.ARGUMENT_ERROR) + + if offset <= 0: + print("The expiry must be in the future", file=sys.stderr) + sys.exit(ExitCodes.ARGUMENT_ERROR) + + if offset > MAX_SIGNED_URL_EXPIRY: + print( + f"The expiry is {offset} seconds from now, but a signed URL " + f"can be valid for at most {MAX_SIGNED_URL_EXPIRY} seconds " + "(7 days)", + file=sys.stderr, + ) + sys.exit(ExitCodes.ARGUMENT_ERROR) + + client = get_client() bucket = parsed.bucket key = parsed.file diff --git a/linodecli/plugins/obj/config.py b/linodecli/plugins/obj/config.py index 0e287f56c..5c835582e 100644 --- a/linodecli/plugins/obj/config.py +++ b/linodecli/plugins/obj/config.py @@ -28,6 +28,10 @@ # for help commands PLUGIN_BASE = "linode-cli obj" +# the longest a signed URL can stay valid, in seconds (7 days); a URL +# signed with Signature Version 4 is refused once its expiry exceeds this +MAX_SIGNED_URL_EXPIRY = 604800 + columns = shutil.get_terminal_size(fallback=(80, 24)).columns PROGRESS_BAR_WIDTH = columns - 20 if columns > 30 else columns diff --git a/tests/unit/test_plugin_obj.py b/tests/unit/test_plugin_obj.py index 608508099..a4f0e9c5b 100644 --- a/tests/unit/test_plugin_obj.py +++ b/tests/unit/test_plugin_obj.py @@ -1,11 +1,14 @@ import time -from unittest.mock import patch +from unittest.mock import Mock, patch +import pytest from pytest import CaptureFixture from linodecli import CLI, plugins +from linodecli.exit_codes import ExitCodes from linodecli.plugins import obj from linodecli.plugins.obj import get_obj_args_parser, helpers, print_help +from linodecli.plugins.obj.config import MAX_SIGNED_URL_EXPIRY def test_print_help(mock_cli: CLI, capsys: CaptureFixture): @@ -336,3 +339,69 @@ def call_operation_side_effect(resource, action, *args, **kwargs): assert ( not delete_calls ), "Cleanup should not be performed when key-cleanup-enabled is False" + + +def test_generate_url_offset_within_limit(capsys: CaptureFixture): + client = Mock() + client.generate_presigned_url.return_value = "https://signed.example" + + obj.generate_url(lambda: client, ["bucket", "key", "+3600"]) + + client.generate_presigned_url.assert_called_once_with( + ClientMethod="get_object", + ExpiresIn=3600, + Params={"Bucket": "bucket", "Key": "key"}, + ) + assert capsys.readouterr().out.strip() == "https://signed.example" + + +def test_generate_url_offset_at_limit(): + client = Mock() + + obj.generate_url( + lambda: client, ["bucket", "key", f"+{MAX_SIGNED_URL_EXPIRY}"] + ) + + assert ( + client.generate_presigned_url.call_args.kwargs["ExpiresIn"] + == MAX_SIGNED_URL_EXPIRY + ) + + +def test_generate_url_epoch_within_limit(): + client = Mock() + + obj.generate_url( + lambda: client, ["bucket", "key", str(int(time.time()) + 3600)] + ) + + assert ( + 3500 + < client.generate_presigned_url.call_args.kwargs["ExpiresIn"] + <= 3600 + ) + + +@pytest.mark.parametrize( + "expiry, message", + [ + (f"+{MAX_SIGNED_URL_EXPIRY + 1}", "at most 604800 seconds"), + ("+31536000", "at most 604800 seconds"), + (str(int(time.time()) + 365 * 24 * 3600), "at most 604800 seconds"), + ("+0", "must be in the future"), + (str(int(time.time()) - 60), "must be in the future"), + ("+an-hour", "Invalid expiry"), + ("tomorrow", "Invalid expiry"), + ], +) +def test_generate_url_rejects_invalid_expiry( + capsys: CaptureFixture, expiry: str, message: str +): + get_client = Mock() + + with pytest.raises(SystemExit) as exc: + obj.generate_url(get_client, ["bucket", "key", expiry]) + + assert exc.value.code == ExitCodes.ARGUMENT_ERROR + assert message in capsys.readouterr().err + get_client.assert_not_called()