Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions packages/google-auth/google/auth/_service_account_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,11 +52,14 @@ def from_dict(data, require=None, use_rsa_signer=True):
"fields {}.".format(", ".join(missing))
)

# Create a signer.
# Create a signer with automatic fallback to EC and PQC (ML-DSA) signers.
if use_rsa_signer:
signer = crypt.RSASigner.from_service_account_info(data)
try:
signer = crypt.RSASigner.from_service_account_info(data)
except (ValueError, TypeError):
signer = crypt.from_service_account_info(data)
else:
signer = crypt.EsSigner.from_service_account_info(data)
signer = crypt.from_service_account_info(data)

return signer

Expand Down
26 changes: 26 additions & 0 deletions packages/google-auth/google/auth/crypt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,31 @@
RSAVerifier = rsa.RSAVerifier


def from_service_account_info(info):
"""Create a Signer instance from a service account info dictionary.

Automatically detects whether the private key is RSA or ECDSA (or other non-RSA)
and returns the appropriate Signer instance.

Args:
info (Mapping[str, str]): Service account info dictionary.

Returns:
google.auth.crypt.Signer: The constructed signer.
"""
private_key = info.get("private_key")
key_id = info.get("private_key_id")
if not private_key:
raise ValueError("The private_key field is missing from service account info.")

try:
return RSASigner.from_service_account_info(info)
except (ValueError, TypeError):
pass

return EsSigner.from_string(private_key, key_id=key_id)


def verify_signature(message, signature, certs, verifier_cls=rsa.RSAVerifier):
"""Verify an RSA or ECDSA cryptographic signature.

Expand Down Expand Up @@ -93,4 +118,5 @@ class to use for verification. This can be used to select different
"RSAVerifier",
"Signer",
"Verifier",
"from_service_account_info",
]
3 changes: 3 additions & 0 deletions packages/google-auth/google/auth/crypt/_cryptography_rsa.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric import rsa
import cryptography.x509

from google.auth import _helpers
Expand Down Expand Up @@ -133,6 +134,8 @@ def from_string(cls, key, key_id=None):
private_key = serialization.load_pem_private_key(
key, password=None, backend=_BACKEND
)
if not isinstance(private_key, rsa.RSAPrivateKey):
raise TypeError("Expected RSAPrivateKey, got {}".format(type(private_key)))
return cls(private_key, key_id=key_id)

def __getstate__(self):
Expand Down
8 changes: 8 additions & 0 deletions packages/google-auth/tests/test__service_account_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,14 @@ def test_from_dict_es384_signer():
assert signer.algorithm == "ES384"


def test_from_dict_ec_auto_detect():
# Pass an EC key without use_rsa_signer=False (default use_rsa_signer=True).
# It should automatically detect the EC key and return an EsSigner without error.
signer = _service_account_info.from_dict(GDCH_SERVICE_ACCOUNT_ES256_INFO)
assert isinstance(signer, crypt.EsSigner)
assert signer.key_id == GDCH_SERVICE_ACCOUNT_ES256_INFO["private_key_id"]


def test_from_dict_bad_private_key():
info = SERVICE_ACCOUNT_INFO.copy()
info["private_key"] = "garbage"
Expand Down
Loading