Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-04-14 17:36
📄 __init__.py	0 B	2025-04-14 17:36
📁 aio	-	2025-04-14 17:36
📄 compliance.py # Copyright 2016 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. import http.client as http_client import time import flask # type: ignore import pytest # type: ignore from pytest_localserver.http import WSGIServer # type: ignore from google.auth import exceptions # .invalid will never resolve, see https://tools.ietf.org/html/rfc2606 NXDOMAIN = "test.invalid" class RequestResponseTests(object): @pytest.fixture(scope="module") def server(self): """Provides a test HTTP server. The test server is automatically created before a test and destroyed at the end. The server is serving a test application that can be used to verify requests. """ app = flask.Flask(__name__) app.debug = True # pylint: disable=unused-variable # (pylint thinks the flask routes are unusued.) @app.route("/basic") def index(): header_value = flask.request.headers.get("x-test-header", "value") headers = {"X-Test-Header": header_value} return "Basic Content", http_client.OK, headers @app.route("/server_error") def server_error(): return "Error", http_client.INTERNAL_SERVER_ERROR @app.route("/wait") def wait(): time.sleep(3) return "Waited" # pylint: enable=unused-variable server = WSGIServer(application=app.wsgi_app) server.start() yield server server.stop() def test_request_basic(self, server): request = self.make_request() response = request(url=server.url + "/basic", method="GET") assert response.status == http_client.OK assert response.headers["x-test-header"] == "value" assert response.data == b"Basic Content" def test_request_with_timeout_success(self, server): request = self.make_request() response = request(url=server.url + "/basic", method="GET", timeout=2) assert response.status == http_client.OK assert response.headers["x-test-header"] == "value" assert response.data == b"Basic Content" def test_request_with_timeout_failure(self, server): request = self.make_request() with pytest.raises(exceptions.TransportError): request(url=server.url + "/wait", method="GET", timeout=1) def test_request_headers(self, server): request = self.make_request() response = request( url=server.url + "/basic", method="GET", headers={"x-test-header": "hello world"}, ) assert response.status == http_client.OK assert response.headers["x-test-header"] == "hello world" assert response.data == b"Basic Content" def test_request_error(self, server): request = self.make_request() response = request(url=server.url + "/server_error", method="GET") assert response.status == http_client.INTERNAL_SERVER_ERROR assert response.data == b"Error" def test_connection_error(self): request = self.make_request() with pytest.raises(exceptions.TransportError): request(url="http://{}".format(NXDOMAIN), method="GET")	3.65 KB	2025-04-14 17:36
📄 test__custom_tls_signer.py # Copyright 2022 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. import base64 import ctypes import os import mock import pytest # type: ignore from requests.packages.urllib3.util.ssl_ import create_urllib3_context # type: ignore import urllib3.contrib.pyopenssl # type: ignore from google.auth import exceptions from google.auth.transport import _custom_tls_signer urllib3.contrib.pyopenssl.inject_into_urllib3() FAKE_ENTERPRISE_CERT_FILE_PATH = "/path/to/enterprise/cert/file" ENTERPRISE_CERT_FILE = os.path.join( os.path.dirname(__file__), "../data/enterprise_cert_valid.json" ) ENTERPRISE_CERT_FILE_PROVIDER = os.path.join( os.path.dirname(__file__), "../data/enterprise_cert_valid_provider.json" ) INVALID_ENTERPRISE_CERT_FILE = os.path.join( os.path.dirname(__file__), "../data/enterprise_cert_invalid.json" ) def test_load_provider_lib(): with mock.patch("ctypes.CDLL", return_value=mock.MagicMock()): _custom_tls_signer.load_provider_lib("/path/to/provider/lib") def test_load_offload_lib(): with mock.patch("ctypes.CDLL", return_value=mock.MagicMock()): lib = _custom_tls_signer.load_offload_lib("/path/to/offload/lib") assert lib.ConfigureSslContext.argtypes == [ _custom_tls_signer.SIGN_CALLBACK_CTYPE, ctypes.c_char_p, ctypes.c_void_p, ] assert lib.ConfigureSslContext.restype == ctypes.c_int def test_load_signer_lib(): with mock.patch("ctypes.CDLL", return_value=mock.MagicMock()): lib = _custom_tls_signer.load_signer_lib("/path/to/signer/lib") assert lib.SignForPython.restype == ctypes.c_int assert lib.SignForPython.argtypes == [ ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ] assert lib.GetCertPemForPython.restype == ctypes.c_int assert lib.GetCertPemForPython.argtypes == [ ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int, ] def test__compute_sha256_digest(): to_be_signed = ctypes.create_string_buffer(b"foo") sig = _custom_tls_signer._compute_sha256_digest(to_be_signed, 4) assert ( base64.b64encode(sig).decode() == "RG5gyEH8CAAh3lxgbt2PLPAHPO8p6i9+cn5dqHfUUYM=" ) def test_get_sign_callback(): # mock signer lib's SignForPython function mock_sig_len = 10 mock_signer_lib = mock.MagicMock() mock_signer_lib.SignForPython.return_value = mock_sig_len # create a sign callback. The callback calls signer lib's SignForPython method sign_callback = _custom_tls_signer.get_sign_callback( mock_signer_lib, FAKE_ENTERPRISE_CERT_FILE_PATH ) # mock the parameters used to call the sign callback to_be_signed = ctypes.POINTER(ctypes.c_ubyte)() to_be_signed_len = 4 returned_sig_array = ctypes.c_ubyte() mock_sig_array = ctypes.byref(returned_sig_array) returned_sign_len = ctypes.c_ulong() mock_sig_len_array = ctypes.byref(returned_sign_len) # call the callback, make sure the signature len is returned via mock_sig_len_array[0] assert sign_callback( mock_sig_array, mock_sig_len_array, to_be_signed, to_be_signed_len ) assert returned_sign_len.value == mock_sig_len def test_get_sign_callback_failed_to_sign(): # mock signer lib's SignForPython function. Set the sig len to be 0 to # indicate the signing failed. mock_sig_len = 0 mock_signer_lib = mock.MagicMock() mock_signer_lib.SignForPython.return_value = mock_sig_len # create a sign callback. The callback calls signer lib's SignForPython method sign_callback = _custom_tls_signer.get_sign_callback( mock_signer_lib, FAKE_ENTERPRISE_CERT_FILE_PATH ) # mock the parameters used to call the sign callback to_be_signed = ctypes.POINTER(ctypes.c_ubyte)() to_be_signed_len = 4 returned_sig_array = ctypes.c_ubyte() mock_sig_array = ctypes.byref(returned_sig_array) returned_sign_len = ctypes.c_ulong() mock_sig_len_array = ctypes.byref(returned_sign_len) sign_callback(mock_sig_array, mock_sig_len_array, to_be_signed, to_be_signed_len) # sign callback should return 0 assert not sign_callback( mock_sig_array, mock_sig_len_array, to_be_signed, to_be_signed_len ) def test_get_cert_no_cert(): # mock signer lib's GetCertPemForPython function to return 0 to indicts # the cert doesn't exit (cert len = 0) mock_signer_lib = mock.MagicMock() mock_signer_lib.GetCertPemForPython.return_value = 0 # call the get cert method with pytest.raises(exceptions.MutualTLSChannelError) as excinfo: _custom_tls_signer.get_cert(mock_signer_lib, FAKE_ENTERPRISE_CERT_FILE_PATH) assert excinfo.match("failed to get certificate") def test_get_cert(): # mock signer lib's GetCertPemForPython function mock_cert_len = 10 mock_signer_lib = mock.MagicMock() mock_signer_lib.GetCertPemForPython.return_value = mock_cert_len # call the get cert method mock_cert = _custom_tls_signer.get_cert( mock_signer_lib, FAKE_ENTERPRISE_CERT_FILE_PATH ) # make sure the signer lib's GetCertPemForPython is called twice, and the # mock_cert has length mock_cert_len assert mock_signer_lib.GetCertPemForPython.call_count == 2 assert len(mock_cert) == mock_cert_len def test_custom_tls_signer(): offload_lib = mock.MagicMock() signer_lib = mock.MagicMock() # Test load_libraries method with mock.patch( "google.auth.transport._custom_tls_signer.load_signer_lib" ) as load_signer_lib: with mock.patch( "google.auth.transport._custom_tls_signer.load_offload_lib" ) as load_offload_lib: load_offload_lib.return_value = offload_lib load_signer_lib.return_value = signer_lib with mock.patch( "google.auth.transport._custom_tls_signer.get_cert" ) as get_cert: with mock.patch( "google.auth.transport._custom_tls_signer.get_sign_callback" ) as get_sign_callback: get_cert.return_value = b"mock_cert" signer_object = _custom_tls_signer.CustomTlsSigner( ENTERPRISE_CERT_FILE ) signer_object.load_libraries() signer_object.attach_to_ssl_context(create_urllib3_context()) get_cert.assert_called_once() get_sign_callback.assert_called_once() offload_lib.ConfigureSslContext.assert_called_once() assert not signer_object.should_use_provider() assert signer_object._enterprise_cert_file_path == ENTERPRISE_CERT_FILE assert signer_object._offload_lib == offload_lib assert signer_object._signer_lib == signer_lib load_signer_lib.assert_called_with("/path/to/signer/lib") load_offload_lib.assert_called_with("/path/to/offload/lib") def test_custom_tls_signer_provider(): provider_lib = mock.MagicMock() # Test load_libraries method with mock.patch( "google.auth.transport._custom_tls_signer.load_provider_lib" ) as load_provider_lib: load_provider_lib.return_value = provider_lib signer_object = _custom_tls_signer.CustomTlsSigner( ENTERPRISE_CERT_FILE_PROVIDER ) signer_object.load_libraries() signer_object.attach_to_ssl_context(mock.MagicMock()) assert signer_object.should_use_provider() assert signer_object._enterprise_cert_file_path == ENTERPRISE_CERT_FILE_PROVIDER assert signer_object._provider_lib == provider_lib load_provider_lib.assert_called_with("/path/to/provider/lib") def test_custom_tls_signer_failed_to_load_libraries(): with pytest.raises(exceptions.MutualTLSChannelError) as excinfo: signer_object = _custom_tls_signer.CustomTlsSigner(INVALID_ENTERPRISE_CERT_FILE) signer_object.load_libraries() assert excinfo.match("enterprise cert file is invalid") def test_custom_tls_signer_failed_to_attach(): with pytest.raises(exceptions.MutualTLSChannelError) as excinfo: signer_object = _custom_tls_signer.CustomTlsSigner(ENTERPRISE_CERT_FILE) signer_object._offload_lib = mock.MagicMock() signer_object._signer_lib = mock.MagicMock() signer_object._sign_callback = mock.MagicMock() signer_object._cert = b"mock cert" signer_object._offload_lib.ConfigureSslContext.return_value = False signer_object.attach_to_ssl_context(mock.MagicMock()) assert excinfo.match("failed to configure ECP Offload SSL context") def test_custom_tls_signer_failed_to_attach_provider(): with pytest.raises(exceptions.MutualTLSChannelError) as excinfo: signer_object = _custom_tls_signer.CustomTlsSigner( ENTERPRISE_CERT_FILE_PROVIDER ) signer_object._provider_lib = mock.MagicMock() signer_object._provider_lib.ECP_attach_to_ctx.return_value = False signer_object.attach_to_ssl_context(mock.MagicMock()) assert excinfo.match("failed to configure ECP Provider SSL context") def test_custom_tls_signer_failed_to_attach_no_libs(): with pytest.raises(exceptions.MutualTLSChannelError) as excinfo: signer_object = _custom_tls_signer.CustomTlsSigner(ENTERPRISE_CERT_FILE) signer_object._offload_lib = None signer_object._signer_lib = None signer_object.attach_to_ssl_context(mock.MagicMock()) assert excinfo.match("Invalid ECP configuration.")	9.82 KB	2025-04-14 17:36
📄 test__http_client.py # Copyright 2016 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. import pytest # type: ignore from google.auth import exceptions import google.auth.transport._http_client from tests.transport import compliance class TestRequestResponse(compliance.RequestResponseTests): def make_request(self): return google.auth.transport._http_client.Request() def test_non_http(self): request = self.make_request() with pytest.raises(exceptions.TransportError) as excinfo: request(url="https://{}".format(compliance.NXDOMAIN), method="GET") assert excinfo.match("https")	1.1 KB	2025-04-14 17:36
📄 test__mtls_helper.py # Copyright 2020 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. import os import re import mock from OpenSSL import crypto import pytest # type: ignore from google.auth import exceptions from google.auth.transport import _mtls_helper CONTEXT_AWARE_METADATA = {"cert_provider_command": ["some command"]} ENCRYPTED_EC_PRIVATE_KEY = b"""-----BEGIN ENCRYPTED PRIVATE KEY----- MIHkME8GCSqGSIb3DQEFDTBCMCkGCSqGSIb3DQEFDDAcBAgl2/yVgs1h3QICCAAw DAYIKoZIhvcNAgkFADAVBgkrBgEEAZdVAQIECJk2GRrvxOaJBIGQXIBnMU4wmciT uA6yD8q0FxuIzjG7E2S6tc5VRgSbhRB00eBO3jWmO2pBybeQW+zVioDcn50zp2ts wYErWC+LCm1Zg3r+EGnT1E1GgNoODbVQ3AEHlKh1CGCYhEovxtn3G+Fjh7xOBrNB saVVeDb4tHD4tMkiVVUBrUcTZPndP73CtgyGHYEphasYPzEz3+AU -----END ENCRYPTED PRIVATE KEY-----""" EC_PUBLIC_KEY = b"""-----BEGIN PUBLIC KEY----- MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEvCNi1NoDY1oMqPHIgXI8RBbTYGi/ brEjbre1nSiQW11xRTJbVeETdsuP0EAu2tG3PcRhhwDfeJ8zXREgTBurNw== -----END PUBLIC KEY-----""" PASSPHRASE = b"""-----BEGIN PASSPHRASE----- password -----END PASSPHRASE-----""" PASSPHRASE_VALUE = b"password" def check_cert_and_key(content, expected_cert, expected_key): success = True cert_match = re.findall(_mtls_helper._CERT_REGEX, content) success = success and len(cert_match) == 1 and cert_match[0] == expected_cert key_match = re.findall(_mtls_helper._KEY_REGEX, content) success = success and len(key_match) == 1 and key_match[0] == expected_key return success class TestCertAndKeyRegex(object): def test_cert_and_key(self): # Test single cert and single key check_cert_and_key( pytest.public_cert_bytes + pytest.private_key_bytes, pytest.public_cert_bytes, pytest.private_key_bytes, ) check_cert_and_key( pytest.private_key_bytes + pytest.public_cert_bytes, pytest.public_cert_bytes, pytest.private_key_bytes, ) # Test cert chain and single key check_cert_and_key( pytest.public_cert_bytes + pytest.public_cert_bytes + pytest.private_key_bytes, pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.private_key_bytes, ) check_cert_and_key( pytest.private_key_bytes + pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.private_key_bytes, ) def test_key(self): # Create some fake keys for regex check. KEY = b"""-----BEGIN PRIVATE KEY----- MIIBCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZg /fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQAB -----END PRIVATE KEY-----""" RSA_KEY = b"""-----BEGIN RSA PRIVATE KEY----- MIIBCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZg /fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQAB -----END RSA PRIVATE KEY-----""" EC_KEY = b"""-----BEGIN EC PRIVATE KEY----- MIIBCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZg /fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQAB -----END EC PRIVATE KEY-----""" check_cert_and_key( pytest.public_cert_bytes + KEY, pytest.public_cert_bytes, KEY ) check_cert_and_key( pytest.public_cert_bytes + RSA_KEY, pytest.public_cert_bytes, RSA_KEY ) check_cert_and_key( pytest.public_cert_bytes + EC_KEY, pytest.public_cert_bytes, EC_KEY ) class TestCheckConfigPath(object): def test_success(self): metadata_path = os.path.join(pytest.data_dir, "context_aware_metadata.json") returned_path = _mtls_helper._check_config_path(metadata_path) assert returned_path is not None def test_failure(self): metadata_path = os.path.join(pytest.data_dir, "not_exists.json") returned_path = _mtls_helper._check_config_path(metadata_path) assert returned_path is None class TestReadMetadataFile(object): def test_success(self): metadata_path = os.path.join(pytest.data_dir, "context_aware_metadata.json") metadata = _mtls_helper._load_json_file(metadata_path) assert "cert_provider_command" in metadata def test_file_not_json(self): # read a file which is not json format. metadata_path = os.path.join(pytest.data_dir, "privatekey.pem") with pytest.raises(exceptions.ClientCertError): _mtls_helper._load_json_file(metadata_path) class TestRunCertProviderCommand(object): def create_mock_process(self, output, error): # There are two steps to execute a script with subprocess.Popen. # (1) process = subprocess.Popen([comannds]) # (2) stdout, stderr = process.communicate() # This function creates a mock process which can be returned by a mock # subprocess.Popen. The mock process returns the given output and error # when mock_process.communicate() is called. mock_process = mock.Mock() attrs = {"communicate.return_value": (output, error), "returncode": 0} mock_process.configure_mock(**attrs) return mock_process @mock.patch("subprocess.Popen", autospec=True) def test_success(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + pytest.private_key_bytes, b"" ) cert, key, passphrase = _mtls_helper._run_cert_provider_command(["command"]) assert cert == pytest.public_cert_bytes assert key == pytest.private_key_bytes assert passphrase is None mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + ENCRYPTED_EC_PRIVATE_KEY + PASSPHRASE, b"" ) cert, key, passphrase = _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) assert cert == pytest.public_cert_bytes assert key == ENCRYPTED_EC_PRIVATE_KEY assert passphrase == PASSPHRASE_VALUE @mock.patch("subprocess.Popen", autospec=True) def test_success_with_cert_chain(self, mock_popen): PUBLIC_CERT_CHAIN_BYTES = pytest.public_cert_bytes + pytest.public_cert_bytes mock_popen.return_value = self.create_mock_process( PUBLIC_CERT_CHAIN_BYTES + pytest.private_key_bytes, b"" ) cert, key, passphrase = _mtls_helper._run_cert_provider_command(["command"]) assert cert == PUBLIC_CERT_CHAIN_BYTES assert key == pytest.private_key_bytes assert passphrase is None mock_popen.return_value = self.create_mock_process( PUBLIC_CERT_CHAIN_BYTES + ENCRYPTED_EC_PRIVATE_KEY + PASSPHRASE, b"" ) cert, key, passphrase = _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) assert cert == PUBLIC_CERT_CHAIN_BYTES assert key == ENCRYPTED_EC_PRIVATE_KEY assert passphrase == PASSPHRASE_VALUE @mock.patch("subprocess.Popen", autospec=True) def test_missing_cert(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.private_key_bytes, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) mock_popen.return_value = self.create_mock_process( ENCRYPTED_EC_PRIVATE_KEY + PASSPHRASE, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) @mock.patch("subprocess.Popen", autospec=True) def test_missing_key(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + PASSPHRASE, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) @mock.patch("subprocess.Popen", autospec=True) def test_missing_passphrase(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + ENCRYPTED_EC_PRIVATE_KEY, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) @mock.patch("subprocess.Popen", autospec=True) def test_passphrase_not_expected(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + pytest.private_key_bytes + PASSPHRASE, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) @mock.patch("subprocess.Popen", autospec=True) def test_encrypted_key_expected(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + pytest.private_key_bytes + PASSPHRASE, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command( ["command"], expect_encrypted_key=True ) @mock.patch("subprocess.Popen", autospec=True) def test_unencrypted_key_expected(self, mock_popen): mock_popen.return_value = self.create_mock_process( pytest.public_cert_bytes + ENCRYPTED_EC_PRIVATE_KEY, b"" ) with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) @mock.patch("subprocess.Popen", autospec=True) def test_cert_provider_returns_error(self, mock_popen): mock_popen.return_value = self.create_mock_process(b"", b"some error") mock_popen.return_value.returncode = 1 with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) @mock.patch("subprocess.Popen", autospec=True) def test_popen_raise_exception(self, mock_popen): mock_popen.side_effect = OSError() with pytest.raises(exceptions.ClientCertError): _mtls_helper._run_cert_provider_command(["command"]) class TestGetClientSslCredentials(object): @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key", autospec=True ) @mock.patch( "google.auth.transport._mtls_helper._run_cert_provider_command", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_success_with_context_aware_metadata( self, mock_check_config_path, mock_load_json_file, mock_run_cert_provider_command, mock_get_workload_cert_and_key, ): mock_check_config_path.return_value = "/path/to/config" mock_load_json_file.return_value = {"cert_provider_command": ["command"]} mock_run_cert_provider_command.return_value = (b"cert", b"key", None) mock_get_workload_cert_and_key.return_value = (None, None) has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials() assert has_cert assert cert == b"cert" assert key == b"key" assert passphrase is None @mock.patch( "google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True ) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_success_with_certificate_config( self, mock_check_config_path, mock_load_json_file, mock_get_cert_config_path, mock_read_cert_and_key_files, ): cert_config_path = "/path/to/config" mock_check_config_path.return_value = cert_config_path mock_load_json_file.return_value = { "cert_configs": { "workload": {"cert_path": "cert/path", "key_path": "key/path"} } } mock_get_cert_config_path.return_value = cert_config_path mock_read_cert_and_key_files.return_value = ( pytest.public_cert_bytes, pytest.private_key_bytes, ) has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials() assert has_cert assert cert == pytest.public_cert_bytes assert key == pytest.private_key_bytes assert passphrase is None @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_success_without_metadata(self, mock_check_config_path): mock_check_config_path.return_value = False has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials() assert not has_cert assert cert is None assert key is None assert passphrase is None @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key", autospec=True ) @mock.patch( "google.auth.transport._mtls_helper._run_cert_provider_command", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_success_with_encrypted_key( self, mock_check_config_path, mock_load_json_file, mock_run_cert_provider_command, mock_get_workload_cert_and_key, ): mock_check_config_path.return_value = "/path/to/config" mock_load_json_file.return_value = {"cert_provider_command": ["command"]} mock_run_cert_provider_command.return_value = (b"cert", b"key", b"passphrase") mock_get_workload_cert_and_key.return_value = (None, None) has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials( generate_encrypted_key=True ) assert has_cert assert cert == b"cert" assert key == b"key" assert passphrase == b"passphrase" mock_run_cert_provider_command.assert_called_once_with( ["command", "--with_passphrase"], expect_encrypted_key=True ) @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_missing_cert_command( self, mock_check_config_path, mock_load_json_file, mock_get_workload_cert_and_key, ): mock_check_config_path.return_value = "/path/to/config" mock_load_json_file.return_value = {} mock_get_workload_cert_and_key.return_value = (None, None) with pytest.raises(exceptions.ClientCertError): _mtls_helper.get_client_ssl_credentials() @mock.patch( "google.auth.transport._mtls_helper._run_cert_provider_command", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_customize_context_aware_metadata_path( self, mock_check_config_path, mock_load_json_file, mock_run_cert_provider_command, ): context_aware_metadata_path = "/path/to/metata/data" mock_check_config_path.return_value = context_aware_metadata_path mock_load_json_file.return_value = {"cert_provider_command": ["command"]} mock_run_cert_provider_command.return_value = (b"cert", b"key", None) has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials( context_aware_metadata_path=context_aware_metadata_path ) assert has_cert assert cert == b"cert" assert key == b"key" assert passphrase is None mock_check_config_path.assert_called_with(context_aware_metadata_path) mock_load_json_file.assert_called_with(context_aware_metadata_path) class TestGetWorkloadCertAndKey(object): @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) @mock.patch( "google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True ) def test_success( self, mock_read_cert_and_key_files, mock_get_cert_config_path, mock_load_json_file, ): cert_config_path = "/path/to/cert" mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { "cert_configs": { "workload": {"cert_path": "cert/path", "key_path": "key/path"} } } mock_read_cert_and_key_files.return_value = ( pytest.public_cert_bytes, pytest.private_key_bytes, ) actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key( cert_config_path ) assert actual_cert == pytest.public_cert_bytes assert actual_key == pytest.private_key_bytes @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) def test_file_not_found_returns_none(self, mock_get_cert_config_path): mock_get_cert_config_path.return_value = None actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key() assert actual_cert is None assert actual_key is None @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) def test_no_cert_configs(self, mock_get_cert_config_path, mock_load_json_file): mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = {} with pytest.raises(exceptions.ClientCertError): _mtls_helper._get_workload_cert_and_key("") @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) def test_no_workload(self, mock_get_cert_config_path, mock_load_json_file): mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = {"cert_configs": {}} with pytest.raises(exceptions.ClientCertError): _mtls_helper._get_workload_cert_and_key("") @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) def test_no_cert_file(self, mock_get_cert_config_path, mock_load_json_file): mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { "cert_configs": {"workload": {"key_path": "path/to/key"}} } with pytest.raises(exceptions.ClientCertError): _mtls_helper._get_workload_cert_and_key("") @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) def test_no_key_file(self, mock_get_cert_config_path, mock_load_json_file): mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { "cert_configs": {"workload": {"cert_path": "path/to/key"}} } with pytest.raises(exceptions.ClientCertError): _mtls_helper._get_workload_cert_and_key("") class TestReadCertAndKeyFile(object): def test_success(self): cert_path = os.path.join(pytest.data_dir, "public_cert.pem") key_path = os.path.join(pytest.data_dir, "privatekey.pem") actual_cert, actual_key = _mtls_helper._read_cert_and_key_files( cert_path, key_path ) assert actual_cert == pytest.public_cert_bytes assert actual_key == pytest.private_key_bytes def test_no_cert_file(self): cert_path = "fake/file/path" key_path = os.path.join(pytest.data_dir, "privatekey.pem") with pytest.raises(FileNotFoundError): _mtls_helper._read_cert_and_key_files(cert_path, key_path) def test_no_key_file(self): cert_path = os.path.join(pytest.data_dir, "public_cert.pem") key_path = "fake/file/path" with pytest.raises(FileNotFoundError): _mtls_helper._read_cert_and_key_files(cert_path, key_path) def test_invalid_cert_file(self): cert_path = os.path.join(pytest.data_dir, "service_account.json") key_path = os.path.join(pytest.data_dir, "privatekey.pem") with pytest.raises(exceptions.ClientCertError): _mtls_helper._read_cert_and_key_files(cert_path, key_path) def test_invalid_key_file(self): cert_path = os.path.join(pytest.data_dir, "public_cert.pem") key_path = os.path.join(pytest.data_dir, "public_cert.pem") with pytest.raises(exceptions.ClientCertError): _mtls_helper._read_cert_and_key_files(cert_path, key_path) class TestGetCertConfigPath(object): def test_success_with_override(self): config_path = os.path.join(pytest.data_dir, "service_account.json") returned_path = _mtls_helper._get_cert_config_path(config_path) assert returned_path == config_path def test_override_does_not_exist(self): config_path = "fake/file/path" returned_path = _mtls_helper._get_cert_config_path(config_path) assert returned_path is None @mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}) @mock.patch("os.path.exists", autospec=True) def test_default(self, mock_path_exists): mock_path_exists.return_value = True returned_path = _mtls_helper._get_cert_config_path() expected_path = os.path.expanduser( _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH ) assert returned_path == expected_path @mock.patch.dict( os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": "path/to/config/file"} ) @mock.patch("os.path.exists", autospec=True) def test_env_variable(self, mock_path_exists): mock_path_exists.return_value = True returned_path = _mtls_helper._get_cert_config_path() expected_path = "path/to/config/file" assert returned_path == expected_path @mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}) @mock.patch("os.path.exists", autospec=True) def test_env_variable_file_does_not_exist(self, mock_path_exists): mock_path_exists.return_value = False returned_path = _mtls_helper._get_cert_config_path() assert returned_path is None @mock.patch.dict( os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": "path/to/config/file"} ) @mock.patch("os.path.exists", autospec=True) def test_default_file_does_not_exist(self, mock_path_exists): mock_path_exists.return_value = False returned_path = _mtls_helper._get_cert_config_path() assert returned_path is None class TestGetClientCertAndKey(object): def test_callback_success(self): callback = mock.Mock() callback.return_value = (pytest.public_cert_bytes, pytest.private_key_bytes) found_cert_key, cert, key = _mtls_helper.get_client_cert_and_key(callback) assert found_cert_key assert cert == pytest.public_cert_bytes assert key == pytest.private_key_bytes @mock.patch( "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True ) def test_use_metadata(self, mock_get_client_ssl_credentials): mock_get_client_ssl_credentials.return_value = ( True, pytest.public_cert_bytes, pytest.private_key_bytes, None, ) found_cert_key, cert, key = _mtls_helper.get_client_cert_and_key() assert found_cert_key assert cert == pytest.public_cert_bytes assert key == pytest.private_key_bytes class TestDecryptPrivateKey(object): def test_success(self): decrypted_key = _mtls_helper.decrypt_private_key( ENCRYPTED_EC_PRIVATE_KEY, PASSPHRASE_VALUE ) private_key = crypto.load_privatekey(crypto.FILETYPE_PEM, decrypted_key) public_key = crypto.load_publickey(crypto.FILETYPE_PEM, EC_PUBLIC_KEY) x509 = crypto.X509() x509.set_pubkey(public_key) # Test the decrypted key works by signing and verification. signature = crypto.sign(private_key, b"data", "sha256") crypto.verify(x509, signature, b"data", "sha256") def test_crypto_error(self): with pytest.raises(crypto.Error): _mtls_helper.decrypt_private_key( ENCRYPTED_EC_PRIVATE_KEY, b"wrong_password" )	25.54 KB	2025-04-14 17:36
📄 test_grpc.py # Copyright 2016 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. import datetime import os import time import mock import pytest # type: ignore from google.auth import _helpers from google.auth import credentials from google.auth import environment_vars from google.auth import exceptions from google.auth import transport from google.oauth2 import service_account try: # pylint: disable=ungrouped-imports import grpc # type: ignore import google.auth.transport.grpc HAS_GRPC = True except ImportError: # pragma: NO COVER HAS_GRPC = False DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data") METADATA_PATH = os.path.join(DATA_DIR, "context_aware_metadata.json") with open(os.path.join(DATA_DIR, "privatekey.pem"), "rb") as fh: PRIVATE_KEY_BYTES = fh.read() with open(os.path.join(DATA_DIR, "public_cert.pem"), "rb") as fh: PUBLIC_CERT_BYTES = fh.read() pytestmark = pytest.mark.skipif(not HAS_GRPC, reason="gRPC is unavailable.") class CredentialsStub(credentials.Credentials): def __init__(self, token="token"): super(CredentialsStub, self).__init__() self.token = token self.expiry = None def refresh(self, request): self.token += "1" def with_quota_project(self, quota_project_id): raise NotImplementedError() class TestAuthMetadataPlugin(object): def test_call_no_refresh(self): credentials = CredentialsStub() request = mock.create_autospec(transport.Request) plugin = google.auth.transport.grpc.AuthMetadataPlugin(credentials, request) context = mock.create_autospec(grpc.AuthMetadataContext, instance=True) context.method_name = mock.sentinel.method_name context.service_url = mock.sentinel.service_url callback = mock.create_autospec(grpc.AuthMetadataPluginCallback) plugin(context, callback) time.sleep(2) callback.assert_called_once_with( [("authorization", "Bearer {}".format(credentials.token))], None ) def test_call_refresh(self): credentials = CredentialsStub() credentials.expiry = datetime.datetime.min + _helpers.REFRESH_THRESHOLD request = mock.create_autospec(transport.Request) plugin = google.auth.transport.grpc.AuthMetadataPlugin(credentials, request) context = mock.create_autospec(grpc.AuthMetadataContext, instance=True) context.method_name = mock.sentinel.method_name context.service_url = mock.sentinel.service_url callback = mock.create_autospec(grpc.AuthMetadataPluginCallback) plugin(context, callback) time.sleep(2) assert credentials.token == "token1" callback.assert_called_once_with( [("authorization", "Bearer {}".format(credentials.token))], None ) def test__get_authorization_headers_with_service_account(self): credentials = mock.create_autospec(service_account.Credentials) request = mock.create_autospec(transport.Request) plugin = google.auth.transport.grpc.AuthMetadataPlugin(credentials, request) context = mock.create_autospec(grpc.AuthMetadataContext, instance=True) context.method_name = "methodName" context.service_url = "https://pubsub.googleapis.com/methodName" plugin._get_authorization_headers(context) credentials._create_self_signed_jwt.assert_called_once_with(None) def test__get_authorization_headers_with_service_account_and_default_host(self): credentials = mock.create_autospec(service_account.Credentials) request = mock.create_autospec(transport.Request) default_host = "pubsub.googleapis.com" plugin = google.auth.transport.grpc.AuthMetadataPlugin( credentials, request, default_host=default_host ) context = mock.create_autospec(grpc.AuthMetadataContext, instance=True) context.method_name = "methodName" context.service_url = "https://pubsub.googleapis.com/methodName" plugin._get_authorization_headers(context) credentials._create_self_signed_jwt.assert_called_once_with( "https://{}/".format(default_host) ) @mock.patch( "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True ) @mock.patch("grpc.composite_channel_credentials", autospec=True) @mock.patch("grpc.metadata_call_credentials", autospec=True) @mock.patch("grpc.ssl_channel_credentials", autospec=True) @mock.patch("grpc.secure_channel", autospec=True) class TestSecureAuthorizedChannel(object): @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_secure_authorized_channel_adc( self, check_config_path, load_json_file, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): credentials = CredentialsStub() request = mock.create_autospec(transport.Request) target = "example.com:80" # Mock the context aware metadata and client cert/key so mTLS SSL channel # will be used. check_config_path.return_value = METADATA_PATH load_json_file.return_value = {"cert_provider_command": ["some command"]} get_client_ssl_credentials.return_value = ( True, PUBLIC_CERT_BYTES, PRIVATE_KEY_BYTES, None, ) channel = None with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): channel = google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, options=mock.sentinel.options ) # Check the auth plugin construction. auth_plugin = metadata_call_credentials.call_args[0][0] assert isinstance(auth_plugin, google.auth.transport.grpc.AuthMetadataPlugin) assert auth_plugin._credentials == credentials assert auth_plugin._request == request # Check the ssl channel call. ssl_channel_credentials.assert_called_once_with( certificate_chain=PUBLIC_CERT_BYTES, private_key=PRIVATE_KEY_BYTES ) # Check the composite credentials call. composite_channel_credentials.assert_called_once_with( ssl_channel_credentials.return_value, metadata_call_credentials.return_value ) # Check the channel call. secure_channel.assert_called_once_with( target, composite_channel_credentials.return_value, options=mock.sentinel.options, ) assert channel == secure_channel.return_value @mock.patch("google.auth.transport.grpc.SslCredentials", autospec=True) def test_secure_authorized_channel_adc_without_client_cert_env( self, ssl_credentials_adc_method, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): # Test client cert won't be used if GOOGLE_API_USE_CLIENT_CERTIFICATE # environment variable is not set. credentials = CredentialsStub() request = mock.create_autospec(transport.Request) target = "example.com:80" channel = google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, options=mock.sentinel.options ) # Check the auth plugin construction. auth_plugin = metadata_call_credentials.call_args[0][0] assert isinstance(auth_plugin, google.auth.transport.grpc.AuthMetadataPlugin) assert auth_plugin._credentials == credentials assert auth_plugin._request == request # Check the ssl channel call. ssl_channel_credentials.assert_called_once() ssl_credentials_adc_method.assert_not_called() # Check the composite credentials call. composite_channel_credentials.assert_called_once_with( ssl_channel_credentials.return_value, metadata_call_credentials.return_value ) # Check the channel call. secure_channel.assert_called_once_with( target, composite_channel_credentials.return_value, options=mock.sentinel.options, ) assert channel == secure_channel.return_value def test_secure_authorized_channel_explicit_ssl( self, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): credentials = mock.Mock() request = mock.Mock() target = "example.com:80" ssl_credentials = mock.Mock() google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, ssl_credentials=ssl_credentials ) # Since explicit SSL credentials are provided, get_client_ssl_credentials # shouldn't be called. assert not get_client_ssl_credentials.called # Check the ssl channel call. assert not ssl_channel_credentials.called # Check the composite credentials call. composite_channel_credentials.assert_called_once_with( ssl_credentials, metadata_call_credentials.return_value ) def test_secure_authorized_channel_mutual_exclusive( self, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): credentials = mock.Mock() request = mock.Mock() target = "example.com:80" ssl_credentials = mock.Mock() client_cert_callback = mock.Mock() with pytest.raises(ValueError): google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, ssl_credentials=ssl_credentials, client_cert_callback=client_cert_callback, ) def test_secure_authorized_channel_with_client_cert_callback_success( self, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): credentials = mock.Mock() request = mock.Mock() target = "example.com:80" client_cert_callback = mock.Mock() client_cert_callback.return_value = (PUBLIC_CERT_BYTES, PRIVATE_KEY_BYTES) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, client_cert_callback=client_cert_callback ) client_cert_callback.assert_called_once() # Check we are using the cert and key provided by client_cert_callback. ssl_channel_credentials.assert_called_once_with( certificate_chain=PUBLIC_CERT_BYTES, private_key=PRIVATE_KEY_BYTES ) # Check the composite credentials call. composite_channel_credentials.assert_called_once_with( ssl_channel_credentials.return_value, metadata_call_credentials.return_value ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_secure_authorized_channel_with_client_cert_callback_failure( self, check_config_path, load_json_file, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): credentials = mock.Mock() request = mock.Mock() target = "example.com:80" client_cert_callback = mock.Mock() client_cert_callback.side_effect = Exception("callback exception") with pytest.raises(Exception) as excinfo: with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, client_cert_callback=client_cert_callback, ) assert str(excinfo.value) == "callback exception" def test_secure_authorized_channel_cert_callback_without_client_cert_env( self, secure_channel, ssl_channel_credentials, metadata_call_credentials, composite_channel_credentials, get_client_ssl_credentials, ): # Test client cert won't be used if GOOGLE_API_USE_CLIENT_CERTIFICATE # environment variable is not set. credentials = mock.Mock() request = mock.Mock() target = "example.com:80" client_cert_callback = mock.Mock() google.auth.transport.grpc.secure_authorized_channel( credentials, request, target, client_cert_callback=client_cert_callback ) # Check client_cert_callback is not called because GOOGLE_API_USE_CLIENT_CERTIFICATE # is not set. client_cert_callback.assert_not_called() ssl_channel_credentials.assert_called_once() # Check the composite credentials call. composite_channel_credentials.assert_called_once_with( ssl_channel_credentials.return_value, metadata_call_credentials.return_value ) @mock.patch("grpc.ssl_channel_credentials", autospec=True) @mock.patch( "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) class TestSslCredentials(object): def test_no_context_aware_metadata( self, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): # Mock that the metadata file doesn't exist. mock_check_config_path.return_value = None with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): ssl_credentials = google.auth.transport.grpc.SslCredentials() # Since no context aware metadata is found, we wouldn't call # get_client_ssl_credentials, and the SSL channel credentials created is # non mTLS. assert ssl_credentials.ssl_credentials is not None assert not ssl_credentials.is_mtls mock_get_client_ssl_credentials.assert_not_called() mock_ssl_channel_credentials.assert_called_once_with() def test_get_client_ssl_credentials_failure( self, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): mock_check_config_path.return_value = METADATA_PATH mock_load_json_file.return_value = {"cert_provider_command": ["some command"]} # Mock that client cert and key are not loaded and exception is raised. mock_get_client_ssl_credentials.side_effect = exceptions.ClientCertError() with pytest.raises(exceptions.MutualTLSChannelError): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): assert google.auth.transport.grpc.SslCredentials().ssl_credentials def test_get_client_ssl_credentials_success( self, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): mock_check_config_path.return_value = METADATA_PATH mock_load_json_file.return_value = {"cert_provider_command": ["some command"]} mock_get_client_ssl_credentials.return_value = ( True, PUBLIC_CERT_BYTES, PRIVATE_KEY_BYTES, None, ) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): ssl_credentials = google.auth.transport.grpc.SslCredentials() assert ssl_credentials.ssl_credentials is not None assert ssl_credentials.is_mtls mock_get_client_ssl_credentials.assert_called_once() mock_ssl_channel_credentials.assert_called_once_with( certificate_chain=PUBLIC_CERT_BYTES, private_key=PRIVATE_KEY_BYTES ) def test_get_client_ssl_credentials_without_client_cert_env( self, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): # Test client cert won't be used if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. ssl_credentials = google.auth.transport.grpc.SslCredentials() assert ssl_credentials.ssl_credentials is not None assert not ssl_credentials.is_mtls mock_check_config_path.assert_not_called() mock_load_json_file.assert_not_called() mock_get_client_ssl_credentials.assert_not_called() mock_ssl_channel_credentials.assert_called_once()	17.74 KB	2025-04-14 17:36
📄 test_mtls.py # Copyright 2020 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. import mock import pytest # type: ignore from google.auth import exceptions from google.auth.transport import _mtls_helper from google.auth.transport import mtls @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_has_default_client_cert_source(check_config_path): def return_path_for_metadata(path): return mock.Mock() if path == _mtls_helper.CONTEXT_AWARE_METADATA_PATH else None check_config_path.side_effect = return_path_for_metadata assert mtls.has_default_client_cert_source() def return_path_for_cert_config(path): return ( mock.Mock() if path == _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH else None ) check_config_path.side_effect = return_path_for_cert_config assert mtls.has_default_client_cert_source() check_config_path.side_effect = None check_config_path.return_value = None assert not mtls.has_default_client_cert_source() @mock.patch("google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True) @mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True) def test_default_client_cert_source( has_default_client_cert_source, get_client_cert_and_key ): # Test default client cert source doesn't exist. has_default_client_cert_source.return_value = False with pytest.raises(exceptions.MutualTLSChannelError): mtls.default_client_cert_source() # The following tests will assume default client cert source exists. has_default_client_cert_source.return_value = True # Test good callback. get_client_cert_and_key.return_value = (True, b"cert", b"key") callback = mtls.default_client_cert_source() assert callback() == (b"cert", b"key") # Test bad callback which throws exception. get_client_cert_and_key.side_effect = ValueError() callback = mtls.default_client_cert_source() with pytest.raises(exceptions.MutualTLSChannelError): callback() @mock.patch( "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True ) @mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True) def test_default_client_encrypted_cert_source( has_default_client_cert_source, get_client_ssl_credentials ): # Test default client cert source doesn't exist. has_default_client_cert_source.return_value = False with pytest.raises(exceptions.MutualTLSChannelError): mtls.default_client_encrypted_cert_source("cert_path", "key_path") # The following tests will assume default client cert source exists. has_default_client_cert_source.return_value = True # Test good callback. get_client_ssl_credentials.return_value = (True, b"cert", b"key", b"passphrase") callback = mtls.default_client_encrypted_cert_source("cert_path", "key_path") with mock.patch("{}.open".format(__name__), return_value=mock.MagicMock()): assert callback() == ("cert_path", "key_path", b"passphrase") # Test bad callback which throws exception. get_client_ssl_credentials.side_effect = exceptions.ClientCertError() callback = mtls.default_client_encrypted_cert_source("cert_path", "key_path") with pytest.raises(exceptions.MutualTLSChannelError): callback()	3.81 KB	2025-04-14 17:36
📄 test_requests.py # Copyright 2016 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. import datetime import functools import http.client as http_client import os import sys import freezegun import mock import OpenSSL import pytest # type: ignore import requests import requests.adapters from google.auth import environment_vars from google.auth import exceptions import google.auth.credentials import google.auth.transport._custom_tls_signer import google.auth.transport._mtls_helper import google.auth.transport.requests from google.oauth2 import service_account from tests.transport import compliance @pytest.fixture def frozen_time(): with freezegun.freeze_time("1970-01-01 00:00:00", tick=False) as frozen: yield frozen class TestRequestResponse(compliance.RequestResponseTests): def make_request(self): return google.auth.transport.requests.Request() def test_timeout(self): http = mock.create_autospec(requests.Session, instance=True) request = google.auth.transport.requests.Request(http) request(url="http://example.com", method="GET", timeout=5) assert http.request.call_args[1]["timeout"] == 5 def test_session_closed_on_del(self): http = mock.create_autospec(requests.Session, instance=True) request = google.auth.transport.requests.Request(http) request.__del__() http.close.assert_called_with() http = mock.create_autospec(requests.Session, instance=True) http.close.side_effect = TypeError("test injected TypeError") request = google.auth.transport.requests.Request(http) request.__del__() http.close.assert_called_with() class TestTimeoutGuard(object): def make_guard(self, args, kwargs): return google.auth.transport.requests.TimeoutGuard(args, kwargs) def test_tracks_elapsed_time_w_numeric_timeout(self, frozen_time): with self.make_guard(timeout=10) as guard: frozen_time.tick(delta=datetime.timedelta(seconds=3.8)) assert guard.remaining_timeout == 6.2 def test_tracks_elapsed_time_w_tuple_timeout(self, frozen_time): with self.make_guard(timeout=(16, 19)) as guard: frozen_time.tick(delta=datetime.timedelta(seconds=3.8)) assert guard.remaining_timeout == (12.2, 15.2) def test_noop_if_no_timeout(self, frozen_time): with self.make_guard(timeout=None) as guard: frozen_time.tick(delta=datetime.timedelta(days=3650)) # NOTE: no timeout error raised, despite years have passed assert guard.remaining_timeout is None def test_timeout_error_w_numeric_timeout(self, frozen_time): with pytest.raises(requests.exceptions.Timeout): with self.make_guard(timeout=10) as guard: frozen_time.tick(delta=datetime.timedelta(seconds=10.001)) assert guard.remaining_timeout == pytest.approx(-0.001) def test_timeout_error_w_tuple_timeout(self, frozen_time): with pytest.raises(requests.exceptions.Timeout): with self.make_guard(timeout=(11, 10)) as guard: frozen_time.tick(delta=datetime.timedelta(seconds=10.001)) assert guard.remaining_timeout == pytest.approx((0.999, -0.001)) def test_custom_timeout_error_type(self, frozen_time): class FooError(Exception): pass with pytest.raises(FooError): with self.make_guard(timeout=1, timeout_error_type=FooError): frozen_time.tick(delta=datetime.timedelta(seconds=2)) def test_lets_suite_errors_bubble_up(self, frozen_time): with pytest.raises(IndexError): with self.make_guard(timeout=1): [1, 2, 3][3] class CredentialsStub(google.auth.credentials.Credentials): def __init__(self, token="token"): super(CredentialsStub, self).__init__() self.token = token def apply(self, headers, token=None): headers["authorization"] = self.token def before_request(self, request, method, url, headers): self.apply(headers) def refresh(self, request): self.token += "1" def with_quota_project(self, quota_project_id): raise NotImplementedError() class TimeTickCredentialsStub(CredentialsStub): """Credentials that spend some (mocked) time when refreshing a token.""" def __init__(self, time_tick, token="token"): self._time_tick = time_tick super(TimeTickCredentialsStub, self).__init__(token=token) def refresh(self, request): self._time_tick() super(TimeTickCredentialsStub, self).refresh(requests) class AdapterStub(requests.adapters.BaseAdapter): def __init__(self, responses, headers=None): super(AdapterStub, self).__init__() self.responses = responses self.requests = [] self.headers = headers or {} def send(self, request, kwargs): # pylint: disable=arguments-differ # request is the only required argument here and the only argument # we care about. self.requests.append(request) return self.responses.pop(0) def close(self): # pragma: NO COVER # pylint wants this to be here because it's abstract in the base # class, but requests never actually calls it. return class TimeTickAdapterStub(AdapterStub): """Adapter that spends some (mocked) time when making a request.""" def __init__(self, time_tick, responses, headers=None): self._time_tick = time_tick super(TimeTickAdapterStub, self).__init__(responses, headers=headers) def send(self, request, kwargs): self._time_tick() return super(TimeTickAdapterStub, self).send(request, kwargs) class TestMutualTlsAdapter(object): @mock.patch.object(requests.adapters.HTTPAdapter, "init_poolmanager") @mock.patch.object(requests.adapters.HTTPAdapter, "proxy_manager_for") def test_success(self, mock_proxy_manager_for, mock_init_poolmanager): adapter = google.auth.transport.requests._MutualTlsAdapter( pytest.public_cert_bytes, pytest.private_key_bytes ) adapter.init_poolmanager() mock_init_poolmanager.assert_called_with(ssl_context=adapter._ctx_poolmanager) adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager) def test_invalid_cert_or_key(self): with pytest.raises(OpenSSL.crypto.Error): google.auth.transport.requests._MutualTlsAdapter( b"invalid cert", b"invalid key" ) @mock.patch.dict("sys.modules", {"OpenSSL.crypto": None}) def test_import_error(self): with pytest.raises(ImportError): google.auth.transport.requests._MutualTlsAdapter( pytest.public_cert_bytes, pytest.private_key_bytes ) def make_response(status=http_client.OK, data=None): response = requests.Response() response.status_code = status response._content = data return response class TestAuthorizedSession(object): TEST_URL = "http://example.com/" def test_constructor(self): authed_session = google.auth.transport.requests.AuthorizedSession( mock.sentinel.credentials ) assert authed_session.credentials == mock.sentinel.credentials def test_constructor_with_auth_request(self): http = mock.create_autospec(requests.Session) auth_request = google.auth.transport.requests.Request(http) authed_session = google.auth.transport.requests.AuthorizedSession( mock.sentinel.credentials, auth_request=auth_request ) assert authed_session._auth_request is auth_request def test_request_default_timeout(self): credentials = mock.Mock(wraps=CredentialsStub()) response = make_response() adapter = AdapterStub([response]) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) authed_session.mount(self.TEST_URL, adapter) patcher = mock.patch("google.auth.transport.requests.requests.Session.request") with patcher as patched_request: authed_session.request("GET", self.TEST_URL) expected_timeout = google.auth.transport.requests._DEFAULT_TIMEOUT assert patched_request.call_args[1]["timeout"] == expected_timeout def test_request_no_refresh(self): credentials = mock.Mock(wraps=CredentialsStub()) response = make_response() adapter = AdapterStub([response]) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) authed_session.mount(self.TEST_URL, adapter) result = authed_session.request("GET", self.TEST_URL) assert response == result assert credentials.before_request.called assert not credentials.refresh.called assert len(adapter.requests) == 1 assert adapter.requests[0].url == self.TEST_URL assert adapter.requests[0].headers["authorization"] == "token" def test_request_refresh(self): credentials = mock.Mock(wraps=CredentialsStub()) final_response = make_response(status=http_client.OK) # First request will 401, second request will succeed. adapter = AdapterStub( [make_response(status=http_client.UNAUTHORIZED), final_response] ) authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) authed_session.mount(self.TEST_URL, adapter) result = authed_session.request("GET", self.TEST_URL) assert result == final_response assert credentials.before_request.call_count == 2 assert credentials.refresh.called assert len(adapter.requests) == 2 assert adapter.requests[0].url == self.TEST_URL assert adapter.requests[0].headers["authorization"] == "token" assert adapter.requests[1].url == self.TEST_URL assert adapter.requests[1].headers["authorization"] == "token1" def test_request_max_allowed_time_timeout_error(self, frozen_time): tick_one_second = functools.partial( frozen_time.tick, delta=datetime.timedelta(seconds=1.0) ) credentials = mock.Mock( wraps=TimeTickCredentialsStub(time_tick=tick_one_second) ) adapter = TimeTickAdapterStub( time_tick=tick_one_second, responses=[make_response(status=http_client.OK)] ) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) authed_session.mount(self.TEST_URL, adapter) # Because a request takes a full mocked second, max_allowed_time shorter # than that will cause a timeout error. with pytest.raises(requests.exceptions.Timeout): authed_session.request("GET", self.TEST_URL, max_allowed_time=0.9) def test_request_max_allowed_time_w_transport_timeout_no_error(self, frozen_time): tick_one_second = functools.partial( frozen_time.tick, delta=datetime.timedelta(seconds=1.0) ) credentials = mock.Mock( wraps=TimeTickCredentialsStub(time_tick=tick_one_second) ) adapter = TimeTickAdapterStub( time_tick=tick_one_second, responses=[ make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.OK), ], ) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) authed_session.mount(self.TEST_URL, adapter) # A short configured transport timeout does not affect max_allowed_time. # The latter is not adjusted to it and is only concerned with the actual # execution time. The call below should thus not raise a timeout error. authed_session.request("GET", self.TEST_URL, timeout=0.5, max_allowed_time=3.1) def test_request_max_allowed_time_w_refresh_timeout_no_error(self, frozen_time): tick_one_second = functools.partial( frozen_time.tick, delta=datetime.timedelta(seconds=1.0) ) credentials = mock.Mock( wraps=TimeTickCredentialsStub(time_tick=tick_one_second) ) adapter = TimeTickAdapterStub( time_tick=tick_one_second, responses=[ make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.OK), ], ) authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=1.1 ) authed_session.mount(self.TEST_URL, adapter) # A short configured refresh timeout does not affect max_allowed_time. # The latter is not adjusted to it and is only concerned with the actual # execution time. The call below should thus not raise a timeout error # (and `timeout` does not come into play either, as it's very long). authed_session.request("GET", self.TEST_URL, timeout=60, max_allowed_time=3.1) def test_request_timeout_w_refresh_timeout_timeout_error(self, frozen_time): tick_one_second = functools.partial( frozen_time.tick, delta=datetime.timedelta(seconds=1.0) ) credentials = mock.Mock( wraps=TimeTickCredentialsStub(time_tick=tick_one_second) ) adapter = TimeTickAdapterStub( time_tick=tick_one_second, responses=[ make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.OK), ], ) authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=100 ) authed_session.mount(self.TEST_URL, adapter) # An UNAUTHORIZED response triggers a refresh (an extra request), thus # the final request that otherwise succeeds results in a timeout error # (all three requests together last 3 mocked seconds). with pytest.raises(requests.exceptions.Timeout): authed_session.request( "GET", self.TEST_URL, timeout=60, max_allowed_time=2.9 ) def test_authorized_session_without_default_host(self): credentials = mock.create_autospec(service_account.Credentials) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) authed_session.credentials._create_self_signed_jwt.assert_called_once_with(None) def test_authorized_session_with_default_host(self): default_host = "pubsub.googleapis.com" credentials = mock.create_autospec(service_account.Credentials) authed_session = google.auth.transport.requests.AuthorizedSession( credentials, default_host=default_host ) authed_session.credentials._create_self_signed_jwt.assert_called_once_with( "https://{}/".format(default_host) ) def test_configure_mtls_channel_with_callback(self): mock_callback = mock.Mock() mock_callback.return_value = ( pytest.public_cert_bytes, pytest.private_key_bytes, ) auth_session = google.auth.transport.requests.AuthorizedSession( credentials=mock.Mock() ) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): auth_session.configure_mtls_channel(mock_callback) assert auth_session.is_mtls assert isinstance( auth_session.adapters["https://"], google.auth.transport.requests._MutualTlsAdapter, ) @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_with_metadata(self, mock_get_client_cert_and_key): mock_get_client_cert_and_key.return_value = ( True, pytest.public_cert_bytes, pytest.private_key_bytes, ) auth_session = google.auth.transport.requests.AuthorizedSession( credentials=mock.Mock() ) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): auth_session.configure_mtls_channel() assert auth_session.is_mtls assert isinstance( auth_session.adapters["https://"], google.auth.transport.requests._MutualTlsAdapter, ) @mock.patch.object(google.auth.transport.requests._MutualTlsAdapter, "__init__") @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_non_mtls( self, mock_get_client_cert_and_key, mock_adapter_ctor ): mock_get_client_cert_and_key.return_value = (False, None, None) auth_session = google.auth.transport.requests.AuthorizedSession( credentials=mock.Mock() ) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): auth_session.configure_mtls_channel() assert not auth_session.is_mtls # Assert _MutualTlsAdapter constructor is not called. mock_adapter_ctor.assert_not_called() @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_exceptions(self, mock_get_client_cert_and_key): mock_get_client_cert_and_key.side_effect = exceptions.ClientCertError() auth_session = google.auth.transport.requests.AuthorizedSession( credentials=mock.Mock() ) with pytest.raises(exceptions.MutualTLSChannelError): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): auth_session.configure_mtls_channel() mock_get_client_cert_and_key.return_value = (False, None, None) with mock.patch.dict("sys.modules"): sys.modules["OpenSSL"] = None with pytest.raises(exceptions.MutualTLSChannelError): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}, ): auth_session.configure_mtls_channel() @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_without_client_cert_env( self, get_client_cert_and_key ): # Test client cert won't be used if GOOGLE_API_USE_CLIENT_CERTIFICATE # environment variable is not set. auth_session = google.auth.transport.requests.AuthorizedSession( credentials=mock.Mock() ) auth_session.configure_mtls_channel() assert not auth_session.is_mtls get_client_cert_and_key.assert_not_called() mock_callback = mock.Mock() auth_session.configure_mtls_channel(mock_callback) assert not auth_session.is_mtls mock_callback.assert_not_called() def test_close_wo_passed_in_auth_request(self): authed_session = google.auth.transport.requests.AuthorizedSession( mock.sentinel.credentials ) authed_session._auth_request_session = mock.Mock(spec=["close"]) authed_session.close() authed_session._auth_request_session.close.assert_called_once_with() def test_close_w_passed_in_auth_request(self): http = mock.create_autospec(requests.Session) auth_request = google.auth.transport.requests.Request(http) authed_session = google.auth.transport.requests.AuthorizedSession( mock.sentinel.credentials, auth_request=auth_request ) authed_session.close() # no raise class TestMutualTlsOffloadAdapter(object): @mock.patch.object(requests.adapters.HTTPAdapter, "init_poolmanager") @mock.patch.object(requests.adapters.HTTPAdapter, "proxy_manager_for") @mock.patch.object( google.auth.transport._custom_tls_signer.CustomTlsSigner, "load_libraries" ) @mock.patch.object( google.auth.transport._custom_tls_signer.CustomTlsSigner, "attach_to_ssl_context", ) def test_success( self, mock_attach_to_ssl_context, mock_load_libraries, mock_proxy_manager_for, mock_init_poolmanager, ): enterprise_cert_file_path = "/path/to/enterprise/cert/json" adapter = google.auth.transport.requests._MutualTlsOffloadAdapter( enterprise_cert_file_path ) mock_load_libraries.assert_called_once() assert mock_attach_to_ssl_context.call_count == 2 adapter.init_poolmanager() mock_init_poolmanager.assert_called_with(ssl_context=adapter._ctx_poolmanager) adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager) @mock.patch.object(requests.adapters.HTTPAdapter, "init_poolmanager") @mock.patch.object(requests.adapters.HTTPAdapter, "proxy_manager_for") @mock.patch.object( google.auth.transport._custom_tls_signer.CustomTlsSigner, "should_use_provider" ) @mock.patch.object( google.auth.transport._custom_tls_signer.CustomTlsSigner, "load_libraries" ) @mock.patch.object( google.auth.transport._custom_tls_signer.CustomTlsSigner, "attach_to_ssl_context", ) def test_success_should_use_provider( self, mock_attach_to_ssl_context, mock_load_libraries, mock_should_use_provider, mock_proxy_manager_for, mock_init_poolmanager, ): enterprise_cert_file_path = "/path/to/enterprise/cert/json" adapter = google.auth.transport.requests._MutualTlsOffloadAdapter( enterprise_cert_file_path ) mock_should_use_provider.side_effect = True mock_load_libraries.assert_called_once() assert mock_attach_to_ssl_context.call_count == 2 adapter.init_poolmanager() mock_init_poolmanager.assert_called_with(ssl_context=adapter._ctx_poolmanager) adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager)	22.49 KB	2025-04-14 17:36
📄 test_urllib3.py # Copyright 2016 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. import http.client as http_client import os import sys import mock import OpenSSL import pytest # type: ignore import urllib3 # type: ignore from google.auth import environment_vars from google.auth import exceptions import google.auth.credentials import google.auth.transport._mtls_helper import google.auth.transport.urllib3 from google.oauth2 import service_account from tests.transport import compliance class TestRequestResponse(compliance.RequestResponseTests): def make_request(self): http = urllib3.PoolManager() return google.auth.transport.urllib3.Request(http) def test_timeout(self): http = mock.create_autospec(urllib3.PoolManager) request = google.auth.transport.urllib3.Request(http) request(url="http://example.com", method="GET", timeout=5) assert http.request.call_args[1]["timeout"] == 5 def test__make_default_http_with_certifi(): http = google.auth.transport.urllib3._make_default_http() assert "cert_reqs" in http.connection_pool_kw @mock.patch.object(google.auth.transport.urllib3, "certifi", new=None) def test__make_default_http_without_certifi(): http = google.auth.transport.urllib3._make_default_http() assert "cert_reqs" not in http.connection_pool_kw class CredentialsStub(google.auth.credentials.Credentials): def __init__(self, token="token"): super(CredentialsStub, self).__init__() self.token = token def apply(self, headers, token=None): headers["authorization"] = self.token def before_request(self, request, method, url, headers): self.apply(headers) def refresh(self, request): self.token += "1" def with_quota_project(self, quota_project_id): raise NotImplementedError() class HttpStub(object): def __init__(self, responses, headers=None): self.responses = responses self.requests = [] self.headers = headers or {} def urlopen(self, method, url, body=None, headers=None, **kwargs): self.requests.append((method, url, body, headers, kwargs)) return self.responses.pop(0) def clear(self): pass class ResponseStub(object): def __init__(self, status=http_client.OK, data=None): self.status = status self.data = data class TestMakeMutualTlsHttp(object): def test_success(self): http = google.auth.transport.urllib3._make_mutual_tls_http( pytest.public_cert_bytes, pytest.private_key_bytes ) assert isinstance(http, urllib3.PoolManager) def test_crypto_error(self): with pytest.raises(OpenSSL.crypto.Error): google.auth.transport.urllib3._make_mutual_tls_http( b"invalid cert", b"invalid key" ) @mock.patch.dict("sys.modules", {"OpenSSL.crypto": None}) def test_import_error(self): with pytest.raises(ImportError): google.auth.transport.urllib3._make_mutual_tls_http( pytest.public_cert_bytes, pytest.private_key_bytes ) class TestAuthorizedHttp(object): TEST_URL = "http://example.com" def test_authed_http_defaults(self): authed_http = google.auth.transport.urllib3.AuthorizedHttp( mock.sentinel.credentials ) assert authed_http.credentials == mock.sentinel.credentials assert isinstance(authed_http.http, urllib3.PoolManager) def test_urlopen_no_refresh(self): credentials = mock.Mock(wraps=CredentialsStub()) response = ResponseStub() http = HttpStub([response]) authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http ) result = authed_http.urlopen("GET", self.TEST_URL) assert result == response assert credentials.before_request.called assert not credentials.refresh.called assert http.requests == [ ("GET", self.TEST_URL, None, {"authorization": "token"}, {}) ] def test_urlopen_refresh(self): credentials = mock.Mock(wraps=CredentialsStub()) final_response = ResponseStub(status=http_client.OK) # First request will 401, second request will succeed. http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), final_response]) authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http ) authed_http = authed_http.urlopen("GET", "http://example.com") assert authed_http == final_response assert credentials.before_request.call_count == 2 assert credentials.refresh.called assert http.requests == [ ("GET", self.TEST_URL, None, {"authorization": "token"}, {}), ("GET", self.TEST_URL, None, {"authorization": "token1"}, {}), ] def test_urlopen_no_default_host(self): credentials = mock.create_autospec(service_account.Credentials) authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials) authed_http.credentials._create_self_signed_jwt.assert_called_once_with(None) def test_urlopen_with_default_host(self): default_host = "pubsub.googleapis.com" credentials = mock.create_autospec(service_account.Credentials) authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, default_host=default_host ) authed_http.credentials._create_self_signed_jwt.assert_called_once_with( "https://{}/".format(default_host) ) def test_proxies(self): http = mock.create_autospec(urllib3.PoolManager) authed_http = google.auth.transport.urllib3.AuthorizedHttp(None, http=http) with authed_http: pass assert http.__enter__.called assert http.__exit__.called authed_http.headers = mock.sentinel.headers assert authed_http.headers == http.headers @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) def test_configure_mtls_channel_with_callback(self, mock_make_mutual_tls_http): callback = mock.Mock() callback.return_value = (pytest.public_cert_bytes, pytest.private_key_bytes) authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials=mock.Mock(), http=mock.Mock() ) with pytest.warns(UserWarning): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): is_mtls = authed_http.configure_mtls_channel(callback) assert is_mtls mock_make_mutual_tls_http.assert_called_once_with( cert=pytest.public_cert_bytes, key=pytest.private_key_bytes ) @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_with_metadata( self, mock_get_client_cert_and_key, mock_make_mutual_tls_http ): authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials=mock.Mock() ) mock_get_client_cert_and_key.return_value = ( True, pytest.public_cert_bytes, pytest.private_key_bytes, ) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): is_mtls = authed_http.configure_mtls_channel() assert is_mtls mock_get_client_cert_and_key.assert_called_once() mock_make_mutual_tls_http.assert_called_once_with( cert=pytest.public_cert_bytes, key=pytest.private_key_bytes ) @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_non_mtls( self, mock_get_client_cert_and_key, mock_make_mutual_tls_http ): authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials=mock.Mock() ) mock_get_client_cert_and_key.return_value = (False, None, None) with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): is_mtls = authed_http.configure_mtls_channel() assert not is_mtls mock_get_client_cert_and_key.assert_called_once() mock_make_mutual_tls_http.assert_not_called() @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_exceptions(self, mock_get_client_cert_and_key): authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials=mock.Mock() ) mock_get_client_cert_and_key.side_effect = exceptions.ClientCertError() with pytest.raises(exceptions.MutualTLSChannelError): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} ): authed_http.configure_mtls_channel() mock_get_client_cert_and_key.return_value = (False, None, None) with mock.patch.dict("sys.modules"): sys.modules["OpenSSL"] = None with pytest.raises(exceptions.MutualTLSChannelError): with mock.patch.dict( os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}, ): authed_http.configure_mtls_channel() @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) def test_configure_mtls_channel_without_client_cert_env( self, get_client_cert_and_key ): callback = mock.Mock() authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials=mock.Mock(), http=mock.Mock() ) # Test the callback is not called if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. is_mtls = authed_http.configure_mtls_channel(callback) assert not is_mtls callback.assert_not_called() # Test ADC client cert is not used if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. is_mtls = authed_http.configure_mtls_channel(callback) assert not is_mtls get_client_cert_and_key.assert_not_called() def test_clear_pool_on_del(self): http = mock.create_autospec(urllib3.PoolManager) authed_http = google.auth.transport.urllib3.AuthorizedHttp( mock.sentinel.credentials, http=http ) authed_http.__del__() http.clear.assert_called_with() authed_http.http = None authed_http.__del__() # Expect it to not crash	11.23 KB	2025-04-14 17:36

Terminal