diff --git a/packages/google-auth/google/auth/identity_pool.py b/packages/google-auth/google/auth/identity_pool.py index 41b312e4f841..69b7f9bfa53c 100644 --- a/packages/google-auth/google/auth/identity_pool.py +++ b/packages/google-auth/google/auth/identity_pool.py @@ -200,6 +200,9 @@ def _read_trust_chain(self): trust_chain_data = f.read() # Split PEM data into individual certificates. cert_blocks = trust_chain_data.split(b"-----BEGIN CERTIFICATE-----") + if len(cert_blocks) > 1: + # Text before the first PEM header is not a certificate. + cert_blocks = cert_blocks[1:] for cert_block in cert_blocks: # Skip empty blocks. if cert_block.strip(): diff --git a/packages/google-auth/tests/test_identity_pool.py b/packages/google-auth/tests/test_identity_pool.py index 7d6a7c1b68d8..6e552f2bd4d9 100644 --- a/packages/google-auth/tests/test_identity_pool.py +++ b/packages/google-auth/tests/test_identity_pool.py @@ -1061,6 +1061,78 @@ def test_retrieve_subject_token_certificate_trust_chain_without_leaf( subject_token = credentials.retrieve_subject_token(None) assert subject_token == json.dumps([CERT_FILE_CONTENT, OTHER_CERT_FILE_CONTENT]) + @pytest.mark.parametrize( + "preamble", + [ + b"# Generated trust chain\n", + b"Bag Attributes\n friendlyName: example\n", + b" \t\r\n", + ], + ) + @pytest.mark.parametrize( + "chain_file", [TRUST_CHAIN_WITH_LEAF_FILE, TRUST_CHAIN_WITHOUT_LEAF_FILE] + ) + @mock.patch( + "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", + return_value=(CERT_FILE, KEY_FILE), + ) + def test_retrieve_subject_token_certificate_trust_chain_preamble( + self, mock_get_workload_cert_and_key_paths, tmp_path, preamble, chain_file + ): + trust_chain_path = tmp_path / "trust_chain.pem" + with open(chain_file, "rb") as chain: + trust_chain_path.write_bytes(preamble + chain.read()) + credentials = self.make_credentials( + credential_source={ + "certificate": { + "use_default_certificate_config": "true", + "trust_chain_path": str(trust_chain_path), + } + } + ) + + subject_token = credentials.retrieve_subject_token(None) + + assert subject_token == json.dumps([CERT_FILE_CONTENT, OTHER_CERT_FILE_CONTENT]) + + @pytest.mark.parametrize( + "trust_chain_data", + [ + b"# No certificates in this file\n", + b"# Generated trust chain\n-----BEGIN CERTIFICATE-----\ninvalid\n-----END CERTIFICATE-----\n", + ], + ) + @mock.patch( + "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", + return_value=(CERT_FILE, KEY_FILE), + ) + def test_retrieve_subject_token_certificate_invalid_trust_chain_preamble( + self, mock_get_workload_cert_and_key_paths, tmp_path, trust_chain_data + ): + trust_chain_path = tmp_path / "trust_chain.pem" + trust_chain_path.write_bytes(trust_chain_data) + credentials = self.make_credentials( + credential_source={ + "certificate": { + "use_default_certificate_config": "true", + "trust_chain_path": str(trust_chain_path), + } + } + ) + + with pytest.raises( + exceptions.RefreshError, + match="Error loading PEM certificates from the trust chain file", + ): + credentials.retrieve_subject_token(None) + + @pytest.mark.parametrize("trust_chain_data", [b"", b" \t\r\n"]) + def test_read_empty_trust_chain(self, trust_chain_data): + supplier = identity_pool._X509Supplier("trust_chain.pem", None) + + with mock.patch("builtins.open", mock.mock_open(read_data=trust_chain_data)): + assert supplier._read_trust_chain() == [] + @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", return_value=(CERT_FILE, KEY_FILE),