Add ECDSA signing support for CloudFront URLs

.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."
+}

awscli/botocore/signers.py

@@ -358,15 +358,15 @@ 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):
             private_key = open('private_key.pem', 'r').read()
             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

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)

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 = (