441 lines
17 KiB
Python
441 lines
17 KiB
Python
# 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
|
|
|
|
from google.auth import exceptions
|
|
from google.auth.transport import _mtls_helper
|
|
|
|
CONTEXT_AWARE_METADATA = {"cert_provider_command": ["some command"]}
|
|
|
|
CONTEXT_AWARE_METADATA_NO_CERT_PROVIDER_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 TestCheckaMetadataPath(object):
|
|
def test_success(self):
|
|
metadata_path = os.path.join(pytest.data_dir, "context_aware_metadata.json")
|
|
returned_path = _mtls_helper._check_dca_metadata_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_dca_metadata_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._read_dca_metadata_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._read_dca_metadata_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._run_cert_provider_command", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._read_dca_metadata_file", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
|
|
)
|
|
def test_success(
|
|
self,
|
|
mock_check_dca_metadata_path,
|
|
mock_read_dca_metadata_file,
|
|
mock_run_cert_provider_command,
|
|
):
|
|
mock_check_dca_metadata_path.return_value = True
|
|
mock_read_dca_metadata_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()
|
|
assert has_cert
|
|
assert cert == b"cert"
|
|
assert key == b"key"
|
|
assert passphrase is None
|
|
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
|
|
)
|
|
def test_success_without_metadata(self, mock_check_dca_metadata_path):
|
|
mock_check_dca_metadata_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._run_cert_provider_command", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._read_dca_metadata_file", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
|
|
)
|
|
def test_success_with_encrypted_key(
|
|
self,
|
|
mock_check_dca_metadata_path,
|
|
mock_read_dca_metadata_file,
|
|
mock_run_cert_provider_command,
|
|
):
|
|
mock_check_dca_metadata_path.return_value = True
|
|
mock_read_dca_metadata_file.return_value = {
|
|
"cert_provider_command": ["command"]
|
|
}
|
|
mock_run_cert_provider_command.return_value = (b"cert", b"key", b"passphrase")
|
|
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._read_dca_metadata_file", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
|
|
)
|
|
def test_missing_cert_command(
|
|
self, mock_check_dca_metadata_path, mock_read_dca_metadata_file
|
|
):
|
|
mock_check_dca_metadata_path.return_value = True
|
|
mock_read_dca_metadata_file.return_value = {}
|
|
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._read_dca_metadata_file", autospec=True
|
|
)
|
|
@mock.patch(
|
|
"google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
|
|
)
|
|
def test_customize_context_aware_metadata_path(
|
|
self,
|
|
mock_check_dca_metadata_path,
|
|
mock_read_dca_metadata_file,
|
|
mock_run_cert_provider_command,
|
|
):
|
|
context_aware_metadata_path = "/path/to/metata/data"
|
|
mock_check_dca_metadata_path.return_value = context_aware_metadata_path
|
|
mock_read_dca_metadata_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_dca_metadata_path.assert_called_with(context_aware_metadata_path)
|
|
mock_read_dca_metadata_file.assert_called_with(context_aware_metadata_path)
|
|
|
|
|
|
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"
|
|
)
|