Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/crypt/es.py: 49%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

91 statements  

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 

18from dataclasses import dataclass 

19from typing import Any, Dict, Optional, Union 

20 

21import cryptography.exceptions 

22from cryptography.hazmat import backends 

23from cryptography.hazmat.primitives import hashes 

24from cryptography.hazmat.primitives import serialization 

25from cryptography.hazmat.primitives.asymmetric import ec 

26from cryptography.hazmat.primitives.asymmetric import padding 

27from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature 

28from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature 

29import cryptography.x509 

30 

31from google.auth import _helpers 

32from 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 

41class _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 

73class 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 

135class 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)