1# Copyright 2016 Google LLC
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"""Pure-Python RSA cryptography implementation.
16
17Uses the ``rsa``, ``pyasn1`` and ``pyasn1_modules`` packages
18to parse PEM files storing PKCS#1 or PKCS#8 keys as well as
19certificates. There is no support for p12 files.
20"""
21
22from __future__ import absolute_import
23
24import io
25
26from pyasn1.codec.der import decoder # type: ignore
27from pyasn1_modules import pem # type: ignore
28from pyasn1_modules.rfc2459 import Certificate # type: ignore
29from pyasn1_modules.rfc5208 import PrivateKeyInfo # type: ignore
30import rsa # type: ignore
31
32from google.auth import _helpers
33from google.auth import exceptions
34from google.auth.crypt import base
35
36_POW2 = (128, 64, 32, 16, 8, 4, 2, 1)
37_CERTIFICATE_MARKER = b"-----BEGIN CERTIFICATE-----"
38_PKCS1_MARKER = ("-----BEGIN RSA PRIVATE KEY-----", "-----END RSA PRIVATE KEY-----")
39_PKCS8_MARKER = ("-----BEGIN PRIVATE KEY-----", "-----END PRIVATE KEY-----")
40_PKCS8_SPEC = PrivateKeyInfo()
41
42
43def _bit_list_to_bytes(bit_list):
44 """Converts an iterable of 1s and 0s to bytes.
45
46 Combines the list 8 at a time, treating each group of 8 bits
47 as a single byte.
48
49 Args:
50 bit_list (Sequence): Sequence of 1s and 0s.
51
52 Returns:
53 bytes: The decoded bytes.
54 """
55 num_bits = len(bit_list)
56 byte_vals = bytearray()
57 for start in range(0, num_bits, 8):
58 curr_bits = bit_list[start : start + 8]
59 char_val = sum(val * digit for val, digit in zip(_POW2, curr_bits))
60 byte_vals.append(char_val)
61 return bytes(byte_vals)
62
63
64class RSAVerifier(base.Verifier):
65 """Verifies RSA cryptographic signatures using public keys.
66
67 Args:
68 public_key (rsa.key.PublicKey): The public key used to verify
69 signatures.
70 """
71
72 def __init__(self, public_key):
73 self._pubkey = public_key
74
75 @_helpers.copy_docstring(base.Verifier)
76 def verify(self, message, signature):
77 message = _helpers.to_bytes(message)
78 try:
79 return rsa.pkcs1.verify(message, signature, self._pubkey)
80 except (ValueError, rsa.pkcs1.VerificationError):
81 return False
82
83 @classmethod
84 def from_string(cls, public_key):
85 """Construct an Verifier instance from a public key or public
86 certificate string.
87
88 Args:
89 public_key (Union[str, bytes]): The public key in PEM format or the
90 x509 public key certificate.
91
92 Returns:
93 google.auth.crypt._python_rsa.RSAVerifier: The constructed verifier.
94
95 Raises:
96 ValueError: If the public_key can't be parsed.
97 """
98 public_key = _helpers.to_bytes(public_key)
99 is_x509_cert = _CERTIFICATE_MARKER in public_key
100
101 # If this is a certificate, extract the public key info.
102 if is_x509_cert:
103 der = rsa.pem.load_pem(public_key, "CERTIFICATE")
104 asn1_cert, remaining = decoder.decode(der, asn1Spec=Certificate())
105 if remaining != b"":
106 raise exceptions.InvalidValue("Unused bytes", remaining)
107
108 cert_info = asn1_cert["tbsCertificate"]["subjectPublicKeyInfo"]
109 key_bytes = _bit_list_to_bytes(cert_info["subjectPublicKey"])
110 pubkey = rsa.PublicKey.load_pkcs1(key_bytes, "DER")
111 else:
112 pubkey = rsa.PublicKey.load_pkcs1(public_key, "PEM")
113 return cls(pubkey)
114
115
116class RSASigner(base.Signer, base.FromServiceAccountMixin):
117 """Signs messages with an RSA private key.
118
119 Args:
120 private_key (rsa.key.PrivateKey): The private key to sign with.
121 key_id (str): Optional key ID used to identify this private key. This
122 can be useful to associate the private key with its associated
123 public key or certificate.
124 """
125
126 def __init__(self, private_key, key_id=None):
127 self._key = private_key
128 self._key_id = key_id
129
130 @property # type: ignore
131 @_helpers.copy_docstring(base.Signer)
132 def key_id(self):
133 return self._key_id
134
135 @_helpers.copy_docstring(base.Signer)
136 def sign(self, message):
137 message = _helpers.to_bytes(message)
138 return rsa.pkcs1.sign(message, self._key, "SHA-256")
139
140 @classmethod
141 def from_string(cls, key, key_id=None):
142 """Construct an Signer instance from a private key in PEM format.
143
144 Args:
145 key (str): Private key in PEM format.
146 key_id (str): An optional key id used to identify the private key.
147
148 Returns:
149 google.auth.crypt.Signer: The constructed signer.
150
151 Raises:
152 ValueError: If the key cannot be parsed as PKCS#1 or PKCS#8 in
153 PEM format.
154 """
155 key = _helpers.from_bytes(key) # PEM expects str in Python 3
156 marker_id, key_bytes = pem.readPemBlocksFromFile(
157 io.StringIO(key), _PKCS1_MARKER, _PKCS8_MARKER
158 )
159
160 # Key is in pkcs1 format.
161 if marker_id == 0:
162 private_key = rsa.key.PrivateKey.load_pkcs1(key_bytes, format="DER")
163 # Key is in pkcs8.
164 elif marker_id == 1:
165 key_info, remaining = decoder.decode(key_bytes, asn1Spec=_PKCS8_SPEC)
166 if remaining != b"":
167 raise exceptions.InvalidValue("Unused bytes", remaining)
168 private_key_info = key_info.getComponentByName("privateKey")
169 private_key = rsa.key.PrivateKey.load_pkcs1(
170 private_key_info.asOctets(), format="DER"
171 )
172 else:
173 raise exceptions.MalformedError("No key could be detected.")
174
175 return cls(private_key, key_id=key_id)