[pull] master from talkiq:master

.github/CODEOWNERS

@@ -1 +1 @@
-*                   @talkiq/engineering @TheKevJames @cphoward
+* @talkiq/engineering @TheKevJames @caseydialpad

.pre-commit-config.yaml

@@ -99,7 +99,7 @@ repos:
     - cryptography==50.0.1
     - pyjwt==2.13.0
     - tenacity==9.1.4
-    - types-requests==2.33.0.20260712
+    - types-requests==2.33.0.20260906
     args:
     - --show-error-codes
     - --strict
@@ -140,8 +140,8 @@ repos:
     name: mypy-storage
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.4
-    - rsa==4.9.1
+    - cryptography==50.0.1
+    - gcloud-aio-auth==5.5.0
     - types-aiofiles==25.1.0.20251011
     - types-requests==2.32.4.20260107
     files: storage/

storage/gcloud/aio/storage/blob.py

@@ -3,20 +3,19 @@
 import datetime
 import enum
 import hashlib
-import io
 import os
 from typing import Any
 from typing import TYPE_CHECKING
 from urllib.parse import quote
 
-import rsa
+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
 from gcloud.aio.auth import BUILD_GCLOUD_REST  # pylint: disable=no-name-in-module
 from gcloud.aio.auth import decode  # pylint: disable=no-name-in-module
 from gcloud.aio.auth import IamClient  # pylint: disable=no-name-in-module
 from gcloud.aio.auth import Token  # pylint: disable=no-name-in-module
-from pyasn1.codec.der import decoder
-from pyasn1_modules import pem
-from pyasn1_modules.rfc5208 import PrivateKeyInfo
 
 from .constants import DEFAULT_TIMEOUT
 
@@ -32,41 +31,6 @@
 
 HOST = os.environ.get('STORAGE_EMULATOR_HOST', 'storage.googleapis.com')
 
-PKCS1_MARKER = (
-    '-----BEGIN RSA PRIVATE KEY-----',
-    '-----END RSA PRIVATE KEY-----',
-)
-PKCS8_MARKER = (
-    '-----BEGIN PRIVATE KEY-----',
-    '-----END PRIVATE KEY-----',
-)
-PKCS8_SPEC = PrivateKeyInfo()
-
-
-class PemKind(enum.Enum):
-    """
-    Tracks the response of ``pem.readPemBlocksFromFile(key, *args)``>
-
-    Note that the specified method returns ``(marker_id, key_bytes)``, where
-    ``marker_id`` is the integer index of the matching ``arg`` (or -1 if no
-    match was found.
-
-    For example::
-
-        (marker_id, _) = pem.readPemBlocksFromFile(key, PKCS1_MARKER,
-                                                   PCKS8_MARKER)
-        if marker_id == -1:
-            # "key" did not match either type or was invalid
-        if marker_id == 0:
-            # "key" matched the zeroth provided marker arg, eg. PKCS1_MARKER
-        if marker_id == 1:
-            # "key" matched the zeroth provided marker arg, eg. PKCS8_MARKER
-    """
-
-    INVALID = -1
-    PKCS1 = 0
-    PKCS8 = 1
-
 
 class _SignatureMethod(enum.Enum):
     """
@@ -239,35 +203,23 @@ async def get_signed_url(  # pylint: disable=too-many-locals
 
     @staticmethod
     def get_pem_signature(str_to_sign: str, private_key: str) -> bytes:
-        # N.B. see the ``PemKind`` enum
-        marker_id, key_bytes = pem.readPemBlocksFromFile(
-            io.StringIO(private_key), PKCS1_MARKER, PKCS8_MARKER,
-        )
-        if marker_id == PemKind.INVALID.value:
-            raise ValueError('private key is invalid or unsupported')
-
-        if marker_id == PemKind.PKCS8.value:
-            # convert from pkcs8 to pkcs1
-            key_info, remaining = decoder.decode(
-                key_bytes,
-                asn1Spec=PKCS8_SPEC,
+        try:
+            key = serialization.load_pem_private_key(
+                private_key.encode(), password=None,
             )
-            if remaining != b'':
-                raise ValueError(
-                    'could not read PKCS8 key: found extra bytes',
-                    remaining,
-                )
+        except (ValueError, TypeError) as e:
+            # N.B. TypeError is raised when the key is encrypted, ie. when a
+            # password would have been required.
+            raise ValueError('private key is invalid or unsupported') from e
 
-            private_key_info = key_info.getComponentByName('privateKey')
-            key_bytes = private_key_info.asOctets()
+        if not isinstance(key, rsa.RSAPrivateKey):
+            raise ValueError('private key is invalid or unsupported')
 
-        key = rsa.key.PrivateKey.load_pkcs1(key_bytes, format='DER')
-        signed_blob = rsa.pkcs1.sign(
+        return key.sign(
             str_to_sign.encode(),
-            key,
-            'SHA-256',
+            padding.PKCS1v15(),
+            hashes.SHA256(),
         )
-        return signed_blob
 
     @staticmethod
     async def get_iam_api_signature(

storage/poetry.rest.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.4.2 and should not be changed by hand.
 
 [[package]]
 name = "certifi"
@@ -338,7 +338,7 @@ test = ["pytest (>=6)"]
 
 [[package]]
 name = "gcloud-rest-auth"
-version = "5.4.4"
+version = "5.5.0"
 description = "Python Client for Google Cloud Auth"
 optional = false
 python-versions = ">= 3.10, < 4.0"
@@ -348,7 +348,7 @@ develop = false
 
 [package.dependencies]
 chardet = ">= 2.0, < 8.0"
-cryptography = ">= 2.0.0, < 51.0.0"
+cryptography = ">= 2.0.0, < 52.0.0"
 pyjwt = ">= 1.5.3, < 3.0.0"
 requests = ">= 2.2.1, < 3.0.0"
 tenacity = ">= 8.2.0, < 10.0.0"
@@ -412,33 +412,6 @@ files = [
 dev = ["pre-commit", "tox"]
 testing = ["coverage", "pytest", "pytest-benchmark"]
 
-[[package]]
-name = "pyasn1"
-version = "0.6.4"
-description = "Pure-Python implementation of ASN.1 types and DER/BER/CER codecs (X.208)"
-optional = false
-python-versions = ">=3.8"
-groups = ["main"]
-files = [
-    {file = "pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b"},
-    {file = "pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81"},
-]
-
-[[package]]
-name = "pyasn1-modules"
-version = "0.4.2"
-description = "A collection of ASN.1-based protocols modules"
-optional = false
-python-versions = ">=3.8"
-groups = ["main"]
-files = [
-    {file = "pyasn1_modules-0.4.2-py3-none-any.whl", hash = "sha256:29253a9207ce32b64c3ac6600edc75368f98473906e8fd1043bd6b5b1de2c14a"},
-    {file = "pyasn1_modules-0.4.2.tar.gz", hash = "sha256:677091de870a80aae844b1ca6134f54652fa2c8c5a52aa396440ac3106e941e6"},
-]
-
-[package.dependencies]
-pyasn1 = ">=0.6.1,<0.7.0"
-
 [[package]]
 name = "pycparser"
 version = "3.0"
@@ -549,21 +522,6 @@ urllib3 = ">=1.26,<3"
 socks = ["PySocks (>=1.5.6,!=1.5.7)"]
 use-chardet-on-py3 = ["chardet (>=3.0.2,<8)"]
 
-[[package]]
-name = "rsa"
-version = "4.9.1"
-description = "Pure-Python RSA implementation"
-optional = false
-python-versions = "<4,>=3.6"
-groups = ["main"]
-files = [
-    {file = "rsa-4.9.1-py3-none-any.whl", hash = "sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762"},
-    {file = "rsa-4.9.1.tar.gz", hash = "sha256:e7bdbfdb5497da4c07dfd35530e1a902659db6ff241e39d9953cad06ebd0ae75"},
-]
-
-[package.dependencies]
-pyasn1 = ">=0.1.3"
-
 [[package]]
 name = "tenacity"
 version = "9.1.4"
@@ -672,4 +630,4 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""]
 [metadata]
 lock-version = "2.1"
 python-versions = ">= 3.10, < 4.0"
-content-hash = "dc382f06447a84adfdb943b96eedbce1c1733f9d9d55335e9ef70542e7362cc6"
+content-hash = "fc25dc9df0092a3afaf853a61392aa0d3bacd52099993d21d6f31c27773e387c"

storage/pyproject.rest.toml

@@ -22,9 +22,9 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">= 3.10, < 4.0"
 # aiofiles = ">=0.6.0, <26.0.0"
+# See https://cryptography.io/en/latest/api-stability/#deprecation
+cryptography = ">= 3.1.0, < 52.0.0"  # pin max to < (major + 3)
 gcloud-rest-auth = ">= 5.4.4, < 6.0.0"
-pyasn1-modules = ">=0.2.1, <0.5.0"
-rsa = ">= 3.1.4, < 5.0.0"
 
 [tool.poetry.group.dev.dependencies]
 gcloud-rest-auth = { path = "../auth" }

storage/pyproject.toml

@@ -22,9 +22,9 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">= 3.10, < 4.0"
 aiofiles = ">=0.6.0, <26.0.0"
+# See https://cryptography.io/en/latest/api-stability/#deprecation
+cryptography = ">= 3.1.0, < 52.0.0"  # pin max to < (major + 3)
 gcloud-aio-auth = ">= 5.4.4, < 6.0.0"
-pyasn1-modules = ">=0.2.1, <0.5.0"
-rsa = ">= 3.1.4, < 5.0.0"
 
 [tool.poetry.group.dev.dependencies]
 gcloud-aio-auth = { path = "../auth" }

storage/tests/unit/blob_test.py

@@ -1,5 +1,60 @@
-from gcloud.aio.storage import blob  # pylint: disable=unused-import
+import pytest
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric import rsa
+from gcloud.aio.storage import Blob
 
 
-def test_importable():
-    assert True
+def generate_key(encoding_format):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    pem = key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=encoding_format,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode()
+    return key, pem
+
+
+@pytest.mark.parametrize(
+    'encoding_format', [
+        serialization.PrivateFormat.PKCS8,
+        serialization.PrivateFormat.TraditionalOpenSSL,  # PKCS#1
+    ],
+)
+def test_get_pem_signature(encoding_format):
+    key, private_key = generate_key(encoding_format)
+
+    signature = Blob.get_pem_signature('test-payload', private_key)
+
+    key.public_key().verify(
+        signature, b'test-payload', padding.PKCS1v15(), hashes.SHA256(),
+    )
+
+
+@pytest.mark.parametrize('private_key', ['', 'not a pem at all'])
+def test_get_pem_signature_invalid_key(private_key):
+    with pytest.raises(ValueError):
+        Blob.get_pem_signature('test-payload', private_key)
+
+
+def test_get_pem_signature_truncated_key():
+    _, private_key = generate_key(serialization.PrivateFormat.PKCS8)
+    lines = private_key.splitlines()
+    truncated = '\n'.join(lines[:2] + lines[-1:]) + '\n'
+
+    with pytest.raises(ValueError):
+        Blob.get_pem_signature('test-payload', truncated)
+
+
+def test_get_pem_signature_non_rsa_key():
+    key = ec.generate_private_key(ec.SECP256R1())
+    private_key = key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode()
+
+    with pytest.raises(ValueError):
+        Blob.get_pem_signature('test-payload', private_key)