|
| 1 | +# Copyright 2017 Google Inc. |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | + |
| 15 | +"""ECDSA verifier and signer that use the ``cryptography`` library. |
| 16 | +""" |
| 17 | + |
| 18 | +from dataclasses import dataclass |
| 19 | +from typing import Any, Dict, Optional, Union |
| 20 | + |
| 21 | +import cryptography.exceptions |
| 22 | +from cryptography.hazmat import backends |
| 23 | +from cryptography.hazmat.primitives import hashes |
| 24 | +from cryptography.hazmat.primitives import serialization |
| 25 | +from cryptography.hazmat.primitives.asymmetric import ec |
| 26 | +from cryptography.hazmat.primitives.asymmetric import padding |
| 27 | +from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature |
| 28 | +from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature |
| 29 | +import cryptography.x509 |
| 30 | + |
| 31 | +from google.auth import _helpers |
| 32 | +from google.auth.crypt import base |
| 33 | + |
| 34 | + |
| 35 | +_CERTIFICATE_MARKER = b"-----BEGIN CERTIFICATE-----" |
| 36 | +_BACKEND = backends.default_backend() |
| 37 | +_PADDING = padding.PKCS1v15() |
| 38 | + |
| 39 | + |
| 40 | +@dataclass |
| 41 | +class _ESAttributes: |
| 42 | + """A class that models ECDSA attributes. |
| 43 | +
|
| 44 | + Attributes: |
| 45 | + rs_size (int): Size for ASN.1 r and s size. |
| 46 | + sha_algo (hashes.HashAlgorithm): Hash algorithm. |
| 47 | + algorithm (str): Algorithm name. |
| 48 | + """ |
| 49 | + |
| 50 | + rs_size: int |
| 51 | + sha_algo: hashes.HashAlgorithm |
| 52 | + algorithm: str |
| 53 | + |
| 54 | + @classmethod |
| 55 | + def from_key( |
| 56 | + cls, key: Union[ec.EllipticCurvePublicKey, ec.EllipticCurvePrivateKey] |
| 57 | + ): |
| 58 | + return cls.from_curve(key.curve) |
| 59 | + |
| 60 | + @classmethod |
| 61 | + def from_curve(cls, curve: ec.EllipticCurve): |
| 62 | + # ECDSA raw signature has (r||s) format where r,s are two |
| 63 | + # integers of size 32 bytes for P-256 curve and 48 bytes |
| 64 | + # for P-384 curve. For P-256 curve, we use SHA256 hash algo, |
| 65 | + # and for P-384 curve we use SHA384 algo. |
| 66 | + if isinstance(curve, ec.SECP384R1): |
| 67 | + return cls(48, hashes.SHA384(), "ES384") |
| 68 | + else: |
| 69 | + # default to ES256 |
| 70 | + return cls(32, hashes.SHA256(), "ES256") |
| 71 | + |
| 72 | + |
| 73 | +class EsVerifier(base.Verifier): |
| 74 | + """Verifies ECDSA cryptographic signatures using public keys. |
| 75 | +
|
| 76 | + Args: |
| 77 | + public_key ( |
| 78 | + cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey): |
| 79 | + The public key used to verify signatures. |
| 80 | + """ |
| 81 | + |
| 82 | + def __init__(self, public_key: ec.EllipticCurvePublicKey) -> None: |
| 83 | + self._pubkey = public_key |
| 84 | + self._attributes = _ESAttributes.from_key(public_key) |
| 85 | + |
| 86 | + @_helpers.copy_docstring(base.Verifier) |
| 87 | + def verify(self, message: bytes, signature: bytes) -> bool: |
| 88 | + # First convert (r||s) raw signature to ASN1 encoded signature. |
| 89 | + sig_bytes = _helpers.to_bytes(signature) |
| 90 | + if len(sig_bytes) != self._attributes.rs_size * 2: |
| 91 | + return False |
| 92 | + r = int.from_bytes(sig_bytes[: self._attributes.rs_size], byteorder="big") |
| 93 | + s = int.from_bytes(sig_bytes[self._attributes.rs_size :], byteorder="big") |
| 94 | + asn1_sig = encode_dss_signature(r, s) |
| 95 | + |
| 96 | + message = _helpers.to_bytes(message) |
| 97 | + try: |
| 98 | + self._pubkey.verify(asn1_sig, message, ec.ECDSA(self._attributes.sha_algo)) |
| 99 | + return True |
| 100 | + except (ValueError, cryptography.exceptions.InvalidSignature): |
| 101 | + return False |
| 102 | + |
| 103 | + @classmethod |
| 104 | + def from_string(cls, public_key: Union[str, bytes]) -> "EsVerifier": |
| 105 | + """Construct an Verifier instance from a public key or public |
| 106 | + certificate string. |
| 107 | +
|
| 108 | + Args: |
| 109 | + public_key (Union[str, bytes]): The public key in PEM format or the |
| 110 | + x509 public key certificate. |
| 111 | +
|
| 112 | + Returns: |
| 113 | + Verifier: The constructed verifier. |
| 114 | +
|
| 115 | + Raises: |
| 116 | + ValueError: If the public key can't be parsed. |
| 117 | + """ |
| 118 | + public_key_data = _helpers.to_bytes(public_key) |
| 119 | + |
| 120 | + if _CERTIFICATE_MARKER in public_key_data: |
| 121 | + cert = cryptography.x509.load_pem_x509_certificate( |
| 122 | + public_key_data, _BACKEND |
| 123 | + ) |
| 124 | + pubkey = cert.public_key() # type: Any |
| 125 | + |
| 126 | + else: |
| 127 | + pubkey = serialization.load_pem_public_key(public_key_data, _BACKEND) |
| 128 | + |
| 129 | + if not isinstance(pubkey, ec.EllipticCurvePublicKey): |
| 130 | + raise TypeError("Expected public key of type EllipticCurvePublicKey") |
| 131 | + |
| 132 | + return cls(pubkey) |
| 133 | + |
| 134 | + |
| 135 | +class EsSigner(base.Signer, base.FromServiceAccountMixin): |
| 136 | + """Signs messages with an ECDSA private key. |
| 137 | +
|
| 138 | + Args: |
| 139 | + private_key ( |
| 140 | + cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey): |
| 141 | + The private key to sign with. |
| 142 | + key_id (str): Optional key ID used to identify this private key. This |
| 143 | + can be useful to associate the private key with its associated |
| 144 | + public key or certificate. |
| 145 | + """ |
| 146 | + |
| 147 | + def __init__( |
| 148 | + self, private_key: ec.EllipticCurvePrivateKey, key_id: Optional[str] = None |
| 149 | + ) -> None: |
| 150 | + self._key = private_key |
| 151 | + self._key_id = key_id |
| 152 | + self._attributes = _ESAttributes.from_key(private_key) |
| 153 | + |
| 154 | + @property |
| 155 | + def algorithm(self) -> str: |
| 156 | + """Name of the algorithm used to sign messages. |
| 157 | + Returns: |
| 158 | + str: The algorithm name. |
| 159 | + """ |
| 160 | + return self._attributes.algorithm |
| 161 | + |
| 162 | + @property # type: ignore |
| 163 | + @_helpers.copy_docstring(base.Signer) |
| 164 | + def key_id(self) -> Optional[str]: |
| 165 | + return self._key_id |
| 166 | + |
| 167 | + @_helpers.copy_docstring(base.Signer) |
| 168 | + def sign(self, message: bytes) -> bytes: |
| 169 | + message = _helpers.to_bytes(message) |
| 170 | + asn1_signature = self._key.sign(message, ec.ECDSA(self._attributes.sha_algo)) |
| 171 | + |
| 172 | + # Convert ASN1 encoded signature to (r||s) raw signature. |
| 173 | + (r, s) = decode_dss_signature(asn1_signature) |
| 174 | + return r.to_bytes(self._attributes.rs_size, byteorder="big") + s.to_bytes( |
| 175 | + self._attributes.rs_size, byteorder="big" |
| 176 | + ) |
| 177 | + |
| 178 | + @classmethod |
| 179 | + def from_string( |
| 180 | + cls, key: Union[bytes, str], key_id: Optional[str] = None |
| 181 | + ) -> "EsSigner": |
| 182 | + """Construct a RSASigner from a private key in PEM format. |
| 183 | +
|
| 184 | + Args: |
| 185 | + key (Union[bytes, str]): Private key in PEM format. |
| 186 | + key_id (str): An optional key id used to identify the private key. |
| 187 | +
|
| 188 | + Returns: |
| 189 | + google.auth.crypt._cryptography_rsa.RSASigner: The |
| 190 | + constructed signer. |
| 191 | +
|
| 192 | + Raises: |
| 193 | + ValueError: If ``key`` is not ``bytes`` or ``str`` (unicode). |
| 194 | + UnicodeDecodeError: If ``key`` is ``bytes`` but cannot be decoded |
| 195 | + into a UTF-8 ``str``. |
| 196 | + ValueError: If ``cryptography`` "Could not deserialize key data." |
| 197 | + """ |
| 198 | + key_bytes = _helpers.to_bytes(key) |
| 199 | + private_key = serialization.load_pem_private_key( |
| 200 | + key_bytes, password=None, backend=_BACKEND |
| 201 | + ) |
| 202 | + |
| 203 | + if not isinstance(private_key, ec.EllipticCurvePrivateKey): |
| 204 | + raise TypeError("Expected private key of type EllipticCurvePrivateKey") |
| 205 | + |
| 206 | + return cls(private_key, key_id=key_id) |
| 207 | + |
| 208 | + def __getstate__(self) -> Dict[str, Any]: |
| 209 | + """Pickle helper that serializes the _key attribute.""" |
| 210 | + state = self.__dict__.copy() |
| 211 | + state["_key"] = self._key.private_bytes( |
| 212 | + encoding=serialization.Encoding.PEM, |
| 213 | + format=serialization.PrivateFormat.PKCS8, |
| 214 | + encryption_algorithm=serialization.NoEncryption(), |
| 215 | + ) |
| 216 | + return state |
| 217 | + |
| 218 | + def __setstate__(self, state: Dict[str, Any]) -> None: |
| 219 | + """Pickle helper that deserializes the _key attribute.""" |
| 220 | + state["_key"] = serialization.load_pem_private_key(state["_key"], None) |
| 221 | + self.__dict__.update(state) |
0 commit comments