From 1358d50b4a4d5e7be964f46a7f1bf647ab6a63fc Mon Sep 17 00:00:00 2001 From: luigi liu Date: Thu, 17 Sep 2026 12:02:54 -0400 Subject: [PATCH] Add ECDSA (P-256) signing support for CloudFront URLs --- .../enhancement-cloudfront-ecdsa.json | 5 + awscli/botocore/signers.py | 10 +- awscli/customizations/cloudfront.py | 76 ++++++++- tests/functional/cloudfront/test_sign.py | 160 ++++++++++++++++++ 4 files changed, 243 insertions(+), 8 deletions(-) create mode 100644 .changes/next-release/enhancement-cloudfront-ecdsa.json diff --git a/.changes/next-release/enhancement-cloudfront-ecdsa.json b/.changes/next-release/enhancement-cloudfront-ecdsa.json new file mode 100644 index 000000000000..c243cf7b031b --- /dev/null +++ b/.changes/next-release/enhancement-cloudfront-ecdsa.json @@ -0,0 +1,5 @@ +{ + "type": "enhancement", + "category": "``cloudfront``", + "description": "Add support for signing CloudFront URLs with ECDSA (P-256) private keys in addition to RSA. The key type is auto-detected from the PEM." +} diff --git a/awscli/botocore/signers.py b/awscli/botocore/signers.py index 8cd7b37215fb..bd4783826100 100644 --- a/awscli/botocore/signers.py +++ b/awscli/botocore/signers.py @@ -358,7 +358,7 @@ def generate_presigned_url( class CloudFrontSigner: '''A signer to create a signed CloudFront URL. - First you create a cloudfront signer based on a normalized RSA signer:: + First you create a cloudfront signer based on a normalized signer:: import rsa def rsa_signer(message): @@ -366,7 +366,7 @@ def rsa_signer(message): return rsa.sign( message, rsa.PrivateKey.load_pkcs1(private_key.encode('utf8')), - 'SHA-1') # CloudFront requires SHA-1 hash + 'SHA-1') # RSA uses SHA-1; ECDSA (P-256) uses SHA-256 cf_signer = CloudFrontSigner(key_id, rsa_signer) To sign with a canned policy:: @@ -386,10 +386,12 @@ def __init__(self, key_id, rsa_signer): :param key_id: The CloudFront Key Pair ID :type rsa_signer: callable - :param rsa_signer: An RSA signer. + :param rsa_signer: An RSA or ECDSA signer. Its only input parameter will be the message to be signed, and its output will be the signed content as a binary string. - The hash algorithm needed by CloudFront is SHA-1. + CloudFront requires a SHA-1 hash for RSA keys and a SHA-256 + hash for ECDSA keys. Name is kept as ``rsa_signer`` for backward + compatibility. """ self.key_id = key_id self.rsa_signer = rsa_signer diff --git a/awscli/customizations/cloudfront.py b/awscli/customizations/cloudfront.py index f785dd4d6a90..37cea5f6113c 100644 --- a/awscli/customizations/cloudfront.py +++ b/awscli/customizations/cloudfront.py @@ -10,6 +10,7 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. +import base64 import hashlib import random import sys @@ -18,7 +19,7 @@ from botocore.signers import CloudFrontSigner from botocore.utils import parse_to_aware_datetime -from awscrt.crypto import RSA, RSASignatureAlgorithm +from awscrt.crypto import EC, RSA, RSASignatureAlgorithm from awscli.arguments import CustomArgument from awscli.customizations.commands import BasicCommand @@ -226,7 +227,10 @@ def _add_sign(command_table, session, **kwargs): class SignCommand(BasicCommand): NAME = 'sign' - DESCRIPTION = 'Sign a given url.' + DESCRIPTION = ( + 'Sign a given url. Supports RSA and ECDSA private keys. ' + 'The key type is auto-detected from the PEM header.' + ) DATE_FORMAT = """Supported formats include: YYYY-MM-DD (which means 0AM UTC of that day), YYYY-MM-DDThh:mm:ss (with default timezone as UTC), @@ -251,7 +255,11 @@ class SignCommand(BasicCommand): { 'name': 'private-key', 'required': True, - 'help_text': 'file://path/to/your/private-key.pem', + 'help_text': ( + 'file://path/to/your/private-key.pem. Both RSA and ECDSA ' + '(P-256) keys are supported; the key type is detected ' + 'automatically.' + ), }, { 'name': 'date-less-than', @@ -275,7 +283,7 @@ class SignCommand(BasicCommand): def _run_main(self, args, parsed_globals): signer = CloudFrontSigner( - args.key_pair_id, RSASigner(args.private_key).sign + args.key_pair_id, _build_signer(args.private_key).sign ) date_less_than = parse_to_aware_datetime(args.date_less_than) date_greater_than = args.date_greater_than @@ -300,6 +308,35 @@ def _run_main(self, args, parsed_globals): return 0 +def _build_signer(private_key): + """Return the appropriate signer based on the private key type.""" + if 'BEGIN EC PRIVATE KEY' in private_key: + return ECDSASigner(private_key) + if 'BEGIN RSA PRIVATE KEY' in private_key: + return RSASigner(private_key) + if 'BEGIN PRIVATE KEY' in private_key: + return _create_signer_from_pkcs8(private_key) + raise ValueError( + "Unsupported key type. Supported formats: " + "RSA (PKCS#1 or PKCS#8) and EC (SEC1 or PKCS#8). " + "Check that your key file has a valid PEM header." + ) + +def _create_signer_from_pkcs8(private_key): + try: + return RSASigner(private_key) + except (RuntimeError, ValueError): + pass + try: + return ECDSASigner(private_key) + except (RuntimeError, ValueError): + pass + raise ValueError( + "Failed to load PKCS#8 private key as either RSA or EC. " + "Check that your key file is a valid private key in PKCS#8 format." + ) + + class RSASigner: def __init__(self, private_key): key_bytes = private_key.encode('utf8') @@ -310,3 +347,34 @@ def sign(self, message): RSASignatureAlgorithm.PKCS1_5_SHA1, hashlib.sha1(message).digest() ) + + +class ECDSASigner: + _P256_COORDINATE_LENGTH = 32 + + def __init__(self, private_key): + try: + der_bytes = _pem_to_der(private_key) + self.priv_key = EC.new_key_from_der_data(der_bytes) + except (RuntimeError, ValueError) as e: + raise ValueError( + "Failed to load EC private key. Ensure the key is a valid " + "EC private key in SEC1 or PKCS#8 PEM format." + ) from e + coords = self.priv_key.get_public_coords() + if len(coords.x) > self._P256_COORDINATE_LENGTH: + raise ValueError( + "Only P-256 EC keys are supported for CloudFront signing. " + "The provided key appears to use a different curve." + ) + + def sign(self, message): + return self.priv_key.sign(hashlib.sha256(message).digest()) + + +def _pem_to_der(pem): + """Strip the PEM armor and base64-decode the body to raw DER bytes.""" + body = ''.join( + line for line in pem.splitlines() if '-----' not in line + ) + return base64.b64decode(body) diff --git a/tests/functional/cloudfront/test_sign.py b/tests/functional/cloudfront/test_sign.py index 94183e3f980c..708b78f72e46 100644 --- a/tests/functional/cloudfront/test_sign.py +++ b/tests/functional/cloudfront/test_sign.py @@ -10,11 +10,26 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. +import base64 +import hashlib + from botocore.compat import parse_qs, urlparse +from botocore.signers import CloudFrontSigner +from botocore.utils import parse_to_aware_datetime + +from awscrt.crypto import EC +from awscli.customizations.cloudfront import _pem_to_der from awscli.testutils import BaseAWSCommandParamsTest, FileCreator, mock +def _url_b64decode(value): + # Reverse the CloudFront-specific base64 substitutions applied by + # CloudFrontSigner._url_b64encode. + restored = value.replace('-', '+').replace('_', '=').replace('~', '/') + return base64.b64decode(restored) + + class TestSign(BaseAWSCommandParamsTest): # A private key only for testing purpose. private_key = ( @@ -109,6 +124,151 @@ def test_custom_policy(self): ) +class BaseECDSASignTest(BaseAWSCommandParamsTest): + # Abstract base; concrete subclasses supply an EC private key. Prevents + # pytest from collecting this base class directly. + __test__ = False + # Overridden by subclasses with an EC private key in a specific PEM format. + private_key = None + url = 'http://example.com/hi' + prefix = 'cloudfront sign --key-pair-id my_id --url http://example.com/hi ' + + def setUp(self): + files = FileCreator() + self.private_key_file = files.create_file('foo.pem', self.private_key) + self.addCleanup(files.remove_all) + super().setUp() + + def _run_and_parse(self, cmdline): + url = self.run_cmd(cmdline)[0].strip() + self.assertEqual(len(url.splitlines()), 1, "Expects only 1 line") + self.assertTrue(url.startswith(self.url), "URL mismatch") + return parse_qs(urlparse(url).query) + + def _assert_signature_verifies(self, params, policy): + # ECDSA signatures are non-deterministic (a random nonce is used), so + # rather than comparing against a fixed value we verify the signature + # cryptographically against the policy that was signed. + self.assertEqual(params['Key-Pair-Id'], ['my_id']) + key = EC.new_key_from_der_data(_pem_to_der(self.private_key)) + signature = _url_b64decode(params['Signature'][0]) + digest = hashlib.sha256(policy.encode('utf8')).digest() + self.assertTrue( + key.verify(digest, signature), + "ECDSA signature failed to verify", + ) + + def test_canned_policy(self): + cmdline = ( + self.prefix + + '--private-key file://' + + self.private_key_file + + ' --date-less-than 2016-1-1' + ) + params = self._run_and_parse(cmdline) + self.assertEqual(params['Expires'], ['1451606400']) + self.assertNotIn('Policy', params) + # For a canned policy the signed payload is the canned policy that + # CloudFrontSigner builds internally from the expiration date. + policy = CloudFrontSigner('my_id', None).build_policy( + self.url, parse_to_aware_datetime('2016-1-1') + ) + self._assert_signature_verifies(params, policy) + + def test_custom_policy(self): + cmdline = ( + self.prefix + + '--private-key file://' + + self.private_key_file + + ' --date-less-than 2016-1-1 --ip-address 12.34.56.78' + ) + params = self._run_and_parse(cmdline) + self.assertNotIn('Expires', params) + # The custom policy is emitted (base64url encoded) in the URL, so the + # exact signed payload can be recovered and verified against. + policy = _url_b64decode(params['Policy'][0]).decode('utf8') + self._assert_signature_verifies(params, policy) + + +class TestSignECDSASEC1(BaseECDSASignTest): + __test__ = True + # An EC (P-256) private key in SEC1 format, only for testing purpose. + private_key = ( + '-----BEGIN EC PRIVATE KEY-----\n' + 'MHcCAQEEIEJv7Bciy04Q7+wqRyaA2xSCsaHtqPmDIQ5msTzcH1xNoAoGCCqGSM49\n' + 'AwEHoUQDQgAEdPNT3OyY+yjo4dOMWcnmKSeIUzrfH2WHkcfKFm32D9B0/DNP9Coj\n' + 'qIXILIjVsmvtp0ULy/ICJEeZbKxUv1/OjA==\n' + '-----END EC PRIVATE KEY-----\n' + ) + + +class TestSignECDSAPKCS8(BaseECDSASignTest): + __test__ = True + # The same EC (P-256) key in PKCS#8 format, only for testing purpose. + private_key = ( + '-----BEGIN PRIVATE KEY-----\n' + 'MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgQm/sFyLLThDv7CpH\n' + 'JoDbFIKxoe2o+YMhDmaxPNwfXE2hRANCAAR081Pc7Jj7KOjh04xZyeYpJ4hTOt8f\n' + 'ZYeRx8oWbfYP0HT8M0/0KiOohcgsiNWya+2nRQvL8gIkR5lsrFS/X86M\n' + '-----END PRIVATE KEY-----\n' + ) + + +class TestSignECDSAUnsupportedCurve(BaseAWSCommandParamsTest): + # An EC private key on the P-384 curve, which CloudFront does not support. + private_key = ( + '-----BEGIN EC PRIVATE KEY-----\n' + 'MIGkAgEBBDCIoGBIXHIpvlHWVTT+jka5Jpj1YR5rWIncoxf6VUxxhlHjEI7hqDto\n' + 'FajvDTKH5jSgBwYFK4EEACKhZANiAAT1i0QFJOMXeKxMx4VpZHw6OoKhEOB4nOXk\n' + 'h+Z9dhiQ4H6O2D84WS6ql+iyNIH2qux8jBUju3fc8NdbVwIqyfQZWRRo/Lg5ekDp\n' + 'M7re404ay7JYpiJXlCZP+RBCBn23NZU=\n' + '-----END EC PRIVATE KEY-----\n' + ) + prefix = 'cloudfront sign --key-pair-id my_id --url http://example.com/hi ' + + def setUp(self): + files = FileCreator() + self.private_key_file = files.create_file('foo.pem', self.private_key) + self.addCleanup(files.remove_all) + super().setUp() + + def test_non_p256_curve_raises_error(self): + cmdline = ( + self.prefix + + '--private-key file://' + + self.private_key_file + + ' --date-less-than 2016-1-1' + ) + _, stderr, _ = self.run_cmd(cmdline, expected_rc=255) + self.assertIn('Only P-256 EC keys are supported', stderr) + + +class TestSignUnsupportedKeyType(BaseAWSCommandParamsTest): + # A key whose PEM header is neither RSA, EC, nor PKCS#8 "PRIVATE KEY". + private_key = ( + '-----BEGIN OPENSSH PRIVATE KEY-----\n' + 'b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtz\n' + '-----END OPENSSH PRIVATE KEY-----\n' + ) + prefix = 'cloudfront sign --key-pair-id my_id --url http://example.com/hi ' + + def setUp(self): + files = FileCreator() + self.private_key_file = files.create_file('foo.pem', self.private_key) + self.addCleanup(files.remove_all) + super().setUp() + + def test_unsupported_key_type_raises_error(self): + cmdline = ( + self.prefix + + '--private-key file://' + + self.private_key_file + + ' --date-less-than 2016-1-1' + ) + _, stderr, _ = self.run_cmd(cmdline, expected_rc=255) + self.assertIn('Unsupported key type', stderr) + + class TestSignPKCS8(BaseAWSCommandParamsTest): # A private key only for testing purpose. private_key = (