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
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
google.auth.crypt.pqc module
============================

.. automodule:: google.auth.crypt.pqc
:members:
:inherited-members:
:show-inheritance:
Original file line number Diff line number Diff line change
Expand Up @@ -14,4 +14,5 @@ Submodules

google.auth.crypt.base
google.auth.crypt.es256
google.auth.crypt.pqc
google.auth.crypt.rsa
14 changes: 12 additions & 2 deletions packages/google-auth/google/auth/crypt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,17 +35,24 @@
The code above also works for :class:`ES256Signer` and :class:`ES256Verifier`.
Note that these two classes are only available if your `cryptography` dependency
version is at least 1.4.0.

Post-quantum ML-DSA signing and verification is available via :class:`PqcSigner`
and :class:`PqcVerifier` when `cryptography` version is at least 47.0.0.
"""

from google.auth.crypt import base
from google.auth.crypt import es
from google.auth.crypt import es256
from google.auth.crypt import pqc
from google.auth.crypt import rsa

EsSigner = es.EsSigner
EsVerifier = es.EsVerifier
ES256Signer = es256.ES256Signer
ES256Verifier = es256.ES256Verifier
PqcSigner = pqc.PqcSigner
PqcVerifier = pqc.PqcVerifier
is_mldsa_key = pqc.is_mldsa_key


# Aliases to maintain the v1.0.0 interface, as the crypt module was split
Expand All @@ -57,7 +64,7 @@


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

Checks that the provided ``signature`` was generated from ``bytes`` using
the private key associated with the ``cert``.
Expand All @@ -69,7 +76,7 @@ def verify_signature(message, signature, certs, verifier_cls=rsa.RSAVerifier):
to use to check the signature.
verifier_cls (Optional[~google.auth.crypt.base.Signer]): Which verifier
class to use for verification. This can be used to select different
algorithms, such as RSA or ECDSA. Default value is :class:`RSAVerifier`.
algorithms, such as RSA, ECDSA, or ML-DSA. Default value is :class:`RSAVerifier`.

Returns:
bool: True if the signature is valid, otherwise False.
Expand All @@ -89,8 +96,11 @@ class to use for verification. This can be used to select different
"EsVerifier",
"ES256Signer",
"ES256Verifier",
"PqcSigner",
"PqcVerifier",
"RSASigner",
"RSAVerifier",
"Signer",
"Verifier",
"is_mldsa_key",
]
285 changes: 285 additions & 0 deletions packages/google-auth/google/auth/crypt/pqc.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,285 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is 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.

"""Post-quantum ML-DSA verifier and signer that use the ``cryptography`` library.
"""

from typing import Any, Dict, Optional, Union

import cryptography.exceptions
from cryptography.hazmat import backends
from cryptography.hazmat.primitives import serialization
import cryptography.x509

from google.auth import _helpers
from google.auth.crypt import base

# ==============================================================================
# Module-level imports & NIST FIPS 204 OID definitions
# ==============================================================================

try:
from cryptography.hazmat.primitives.asymmetric import mldsa
except ImportError:
mldsa = None # type: ignore

_CERTIFICATE_MARKER = b"-----BEGIN CERTIFICATE-----"
_BACKEND = backends.default_backend()

_UPGRADE_ERROR = (
"Post-Quantum ML-DSA Service Account keys require cryptography>=47.0.0. "
"Please upgrade your cryptography library (pip install 'cryptography>=47.0.0')."
)


def _get_mldsa_types(*attr_names: str) -> tuple:
if mldsa is None:
return ()
types = []
for name in attr_names:
t = getattr(mldsa, name, None)
if isinstance(t, type):
types.append(t)
return tuple(types)


# ==============================================================================
# Key detection helpers
# ==============================================================================


def is_mldsa_key(key: Union[str, bytes]) -> bool:
"""Determines whether a key is an ML-DSA private or public key."""
if mldsa is None:
return False

expected_types = _get_mldsa_types(
"MLDSA44PrivateKey",
"MLDSA65PrivateKey",
"MLDSA87PrivateKey",
"MLDSA44PublicKey",
"MLDSA65PublicKey",
"MLDSA87PublicKey",
)
if not expected_types:
return False

try:
key_bytes = _helpers.to_bytes(key)
except (TypeError, ValueError, AttributeError):
return False

try:
priv_key = serialization.load_pem_private_key(
key_bytes, password=None, backend=_BACKEND
)
return isinstance(priv_key, expected_types)
except Exception:
pass

try:
pub_key = serialization.load_pem_public_key(key_bytes, _BACKEND)
return isinstance(pub_key, expected_types)
except Exception:
pass

return False


def _get_mldsa_algorithm_name(key: Any) -> str:
"""Determines the ML-DSA algorithm string ("ML-DSA-44", "ML-DSA-65", or "ML-DSA-87") for a key."""
cls_name = key.__class__.__name__
t44 = _get_mldsa_types("MLDSA44PrivateKey", "MLDSA44PublicKey")
t65 = _get_mldsa_types("MLDSA65PrivateKey", "MLDSA65PublicKey")
t87 = _get_mldsa_types("MLDSA87PrivateKey", "MLDSA87PublicKey")

if "87" in cls_name or (t87 and isinstance(key, t87)):
return "ML-DSA-87"
if "65" in cls_name or (t65 and isinstance(key, t65)):
return "ML-DSA-65"
if "44" in cls_name or (t44 and isinstance(key, t44)):
return "ML-DSA-44"

raise TypeError(
"Expected key of type ML-DSA (e.g. MLDSA44PrivateKey, MLDSA65PrivateKey, or MLDSA87PrivateKey), got: {}".format(
cls_name
)
)


# ==============================================================================
# PQC Verifier
# ==============================================================================


class PqcVerifier(base.Verifier):
"""Verifies ML-DSA cryptographic signatures using public keys.

ML-DSA-65 (3,309-byte signature) is set as the recommended default PQC key
type over ML-DSA-87 (4,627 bytes) to minimize HTTP request header overhead.

Args:
public_key: The public key used to verify signatures.
"""

def __init__(self, public_key: Any) -> None:
self._pubkey = public_key

@_helpers.copy_docstring(base.Verifier)
def verify(self, message: bytes, signature: bytes) -> bool:
message = _helpers.to_bytes(message)
sig_bytes = _helpers.to_bytes(signature)
try:
self._pubkey.verify(sig_bytes, message)
return True
except (ValueError, TypeError, cryptography.exceptions.InvalidSignature):
return False

@classmethod
def from_string(cls, public_key: Union[str, bytes]) -> "PqcVerifier":
"""Construct a Verifier instance from a public key or certificate string.

Args:
public_key (Union[bytes, str]): Public key or certificate in PEM format.

Returns:
google.auth.crypt.pqc.PqcVerifier: The constructed verifier.

Raises:
RuntimeError: If ``cryptography`` is less than version 47.0.0.
ValueError: If ``public_key`` cannot be parsed.
TypeError: If ``public_key`` is not an ML-DSA public key.
"""
if mldsa is None:
raise RuntimeError(_UPGRADE_ERROR)

public_key_data = _helpers.to_bytes(public_key)

if _CERTIFICATE_MARKER in public_key_data:
cert = cryptography.x509.load_pem_x509_certificate(
public_key_data, _BACKEND
)
pubkey: Any = cert.public_key()
else:
pubkey = serialization.load_pem_public_key(public_key_data, _BACKEND)

expected_types = _get_mldsa_types(
"MLDSA44PublicKey", "MLDSA65PublicKey", "MLDSA87PublicKey"
)
if not expected_types or not isinstance(pubkey, expected_types):
raise TypeError(
"Expected public key of type ML-DSA (e.g. MLDSA44PublicKey, MLDSA65PublicKey, or MLDSA87PublicKey)"
)

return cls(pubkey)


# ==============================================================================
# PQC Signer
# ==============================================================================


class PqcSigner(base.Signer, base.FromServiceAccountMixin):
"""Signs messages with a post-quantum ML-DSA (Module-Lattice-Based Digital
Signature Algorithm) private key.

ML-DSA-65 (3,309-byte signature) is set as the recommended default PQC key
type over ML-DSA-87 (4,627 bytes) to minimize HTTP request header overhead.

Args:
private_key: The ML-DSA private key to sign with.
key_id (Optional[str]): Optional key ID used to identify this private key.
"""

RECOMMENDED_DEFAULT_ALGORITHM = "ML-DSA-65"

def __init__(self, private_key: Any, key_id: Optional[str] = None) -> None:
self._key = private_key
self._key_id = key_id
self._algorithm = _get_mldsa_algorithm_name(private_key)

@property
def algorithm(self) -> str:
"""Name of the algorithm used to sign messages.

Returns:
str: The algorithm name (e.g., "ML-DSA-65" or "ML-DSA-87").
"""
return self._algorithm

@property # type: ignore
@_helpers.copy_docstring(base.Signer)
def key_id(self) -> Optional[str]:
return self._key_id

@_helpers.copy_docstring(base.Signer)
def sign(self, message: bytes) -> bytes:
message = _helpers.to_bytes(message)
return self._key.sign(message)

@classmethod
def from_string(
cls, key: Union[bytes, str], key_id: Optional[str] = None
) -> "PqcSigner":
"""Construct a PqcSigner from a private key in PEM format.

Args:
key (Union[bytes, str]): Private key in PEM format.
key_id (Optional[str]): An optional key id used to identify the private key.

Returns:
google.auth.crypt.pqc.PqcSigner: The constructed signer.

Raises:
RuntimeError: If ``cryptography`` is less than version 47.0.0.
ValueError: If ``cryptography`` "Could not deserialize key data."
TypeError: If ``key`` is not an ML-DSA private key.
"""
if mldsa is None:
raise RuntimeError(_UPGRADE_ERROR)

key_bytes = _helpers.to_bytes(key)
private_key = serialization.load_pem_private_key(
key_bytes, password=None, backend=_BACKEND
)

expected_types = _get_mldsa_types(
"MLDSA44PrivateKey", "MLDSA65PrivateKey", "MLDSA87PrivateKey"
)
if not expected_types or not isinstance(private_key, expected_types):
raise TypeError(
"Expected private key of type ML-DSA (e.g. MLDSA44PrivateKey, MLDSA65PrivateKey, or MLDSA87PrivateKey)"
)

return cls(private_key, key_id=key_id)

def __getstate__(self) -> Dict[str, Any]:
"""Pickle helper that serializes the _key attribute."""
state = self.__dict__.copy()
state["_key"] = self._key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
return state

def __setstate__(self, state: Dict[str, Any]) -> None:
"""Pickle helper that deserializes the _key attribute."""
if mldsa is None:
raise RuntimeError(_UPGRADE_ERROR)
state = state.copy()
state["_key"] = serialization.load_pem_private_key(
state["_key"], password=None, backend=_BACKEND
)
self.__dict__.update(state)
Comment on lines +277 to +285

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Improve consistency and safety in __setstate__ by copying the state dictionary to avoid mutating the unpickled state in-place, and passing explicit keyword arguments (password=None and backend=_BACKEND) to load_pem_private_key to match the implementation in from_string.

Suggested change
def __setstate__(self, state: Dict[str, Any]) -> None:
"""Pickle helper that deserializes the _key attribute."""
if mldsa is None:
raise RuntimeError(_UPGRADE_ERROR)
state["_key"] = serialization.load_pem_private_key(state["_key"], None)
self.__dict__.update(state)
def __setstate__(self, state: Dict[str, Any]) -> None:
"""Pickle helper that deserializes the _key attribute."""
if mldsa is None:
raise RuntimeError(_UPGRADE_ERROR)
state = state.copy()
state["_key"] = serialization.load_pem_private_key(
state["_key"], password=None, backend=_BACKEND
)
self.__dict__.update(state)

Loading
Loading