feat: add SslCredentials class for mTLS ADC

google/auth/transport/_mtls_helper.py

@@ -0,0 +1,107 @@
+# 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.
+
+"""Helper functions for getting mTLS cert and key, for internal use only."""
+
+import json
+import logging
+from os import path
+import subprocess
+
+CONTEXT_AWARE_METADATA_PATH = "~/.secureConnect/context_aware_metadata.json"
+_CERT_PROVIDER_COMMAND = "cert_provider_command"
+_CERTIFICATE_SURFFIX = b"-----END CERTIFICATE-----\n"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def read_metadata_file(metadata_path):
+    """Function to load context aware metadata from the given path.
+
+    Args:
+        metadata_path (str): context aware metadata path.
+
+    Returns:
+        Dict[str]:
+            The metadata. If metadata reading or parsing fails, return None.
+    """
+    metadata_path = path.expanduser(metadata_path)
+    if not path.exists(metadata_path):
+        _LOGGER.debug("%s is not found, skip client SSL authentication.", metadata_path)
+        return None
+
+    with open(metadata_path) as f:
+        try:
+            metadata = json.load(f)
+        except json.decoder.JSONDecodeError as e:
+            _LOGGER.debug(
+                "Failed to decode context_aware_metadata.json with error: %s", str(e)
+            )
+            return None
+
+    return metadata
+
+
+def get_client_ssl_credentials(metadata_json, platform):
+    """Function to get mTLS client side cert and key.
+
+    Args:
+        metadata_json (Dict[str]): metadata JSON file which contains the cert
+            provider command.
+        platform (str): The OS.
+
+    Returns:
+        Tuple[bool, bytes, bytes, bytes, bytes]:
+            The tuple contains the following in order:
+            (1) boolean to show if client cert and key is obtained successfully
+            (2) client certificate in PEM forma if successful, otherwise None
+            (3) client key in PEM format if successful, otherwise None
+            (4) stdout from cert provider command execution
+            (5) stderr from cert provider command execution
+    """
+
+    # Check the system. For now only Linux is supported.
+    if not platform.startswith("linux"):
+        _LOGGER.debug("mTLS for platform: %s is not supported.", platform)
+        return False, None, None, None, None
+
+    # Execute the cert provider command in the metadata json file.
+    if _CERT_PROVIDER_COMMAND not in metadata_json:
+        _LOGGER.debug("cert_provider_command missing, skip client SSL authentication")
+        return False, None, None, None, None
+    try:
+        command = metadata_json[_CERT_PROVIDER_COMMAND]
+        process = subprocess.Popen(
+            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+    except OSError as e:
+        _LOGGER.debug("Failed to run cert provider command with error: %s", str(e))
+        return False, None, None, None, None
+
+    # Check cert provider command execution error.
+    if stderr != b"":
+        _LOGGER.debug("Cert provider command failed with error: %s", stderr)
+        return False, None, None, stdout, stderr
+
+    # Parse stdout, it should be a cert followed by a key, both in PEM format.
+    cert_end = stdout.find(_CERTIFICATE_SURFFIX)
+    if cert_end == -1:
+        _LOGGER.debug("Client SSL certificate is missing")
+        return False, None, None, stdout, stderr
+    private_key_start = cert_end + len(_CERTIFICATE_SURFFIX)
+    if private_key_start >= len(stdout):
+        _LOGGER.debug("Client SSL private key is missing")
+        return False, None, None, stdout, stderr
+    return True, stdout[0:private_key_start], stdout[private_key_start:], stdout, stderr

google/auth/transport/grpc.py

@@ -17,9 +17,12 @@
 from __future__ import absolute_import
 
 from concurrent import futures
+from sys import platform
 
 import six
 
+from google.auth.transport import _mtls_helper
+
 try:
     import grpc
 except ImportError as caught_exc:  # pragma: NO COVER
@@ -149,3 +152,37 @@ def secure_authorized_channel(
     )
 
     return grpc.secure_channel(target, composite_credentials, **kwargs)
+
+
+class SslCredentials:
+    """Class for application default SSL credentials. For Linux with endpoint
+    verification support, device certificate will be automatically loaded if
+    available and mutual TLS will be established.
+    """
+
+    def __init__(self):
+        self._is_mtls = False
+
+        # Load client SSL credentials.
+        context_aware_metadata = _mtls_helper.read_metadata_file(
+            _mtls_helper.CONTEXT_AWARE_METADATA_PATH
+        )
+        if context_aware_metadata:
+            self._is_mtls, cert, key, _, _ = _mtls_helper.get_client_ssl_credentials(
+                context_aware_metadata, platform
+            )
+
+        if self._is_mtls:
+            self._ssl_credentials = grpc.ssl_channel_credentials(
+                certificate_chain=cert, private_key=key
+            )
+        else:
+            self._ssl_credentials = grpc.ssl_channel_credentials()
+
+    @property
+    def ssl_credentials(self):
+        return self._ssl_credentials
+
+    @property
+    def is_mtls(self):
+        return self._is_mtls

tests/data/context_aware_metadata.json

@@ -0,0 +1,6 @@
+{
+  "cert_provider_command":[
+    "/opt/google/endpoint-verification/bin/SecureConnectHelper",
+    "--print_certificate"],
+  "device_resource_ids":["11111111-1111-1111"]
+}

tests/transport/test__mtls_helper.py

@@ -0,0 +1,188 @@
+# 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 mock
+
+from google.auth.transport import _mtls_helper
+
+DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data")
+
+with open(os.path.join(DATA_DIR, "privatekey.pub"), "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()
+
+CLIENT_SSL_CREDENTIALS = PUBLIC_CERT_BYTES + PRIVATE_KEY_BYTES
+
+CONTEXT_AWARE_METADATA = {"cert_provider_command": ["some command"]}
+
+CONTEXT_AWARE_METADATA_NO_CERT_PROVIDER_COMMAND = {}
+
+
+class TestReadMetadataFile(object):
+    def test_success(self):
+        metadata_path = os.path.join(DATA_DIR, "context_aware_metadata.json")
+        metadata = _mtls_helper.read_metadata_file(metadata_path)
+
+        assert "cert_provider_command" in metadata
+
+    def test_file_not_exist(self):
+        metadata_path = os.path.join(DATA_DIR, "not_exist.json")
+        metadata = _mtls_helper.read_metadata_file(metadata_path)
+
+        assert metadata is None
+
+    def test_file_not_json(self):
+        # read a file which is not json format.
+        metadata_path = os.path.join(DATA_DIR, "privatekey.pem")
+        metadata = _mtls_helper.read_metadata_file(metadata_path)
+
+        assert metadata is None
+
+
+class TestGetClientSslCredentials(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)}
+        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(CLIENT_SSL_CREDENTIALS, b"")
+        success, cert, key, output, error = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, output, error),
+                    (
+                        True,
+                        PUBLIC_CERT_BYTES,
+                        PRIVATE_KEY_BYTES,
+                        CLIENT_SSL_CREDENTIALS,
+                        b"",
+                    ),
+                )
+            ]
+        )
+
+    def test_not_linux_platform(self):
+        success, cert, key, stdout, stderr = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "win32"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, stdout, stderr),
+                    (False, None, None, None, None),
+                )
+            ]
+        )
+
+    def test_missing_cert_provider_command(self):
+        success, cert, key, stdout, stderr = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA_NO_CERT_PROVIDER_COMMAND, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, stdout, stderr),
+                    (False, None, None, None, None),
+                )
+            ]
+        )
+
+    @mock.patch("subprocess.Popen", autospec=True)
+    def test_missing_cert(self, mock_popen):
+        mock_popen.return_value = self.create_mock_process(PRIVATE_KEY_BYTES, b"")
+        success, cert, key, output, error = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, output, error),
+                    (False, None, None, PRIVATE_KEY_BYTES, b""),
+                )
+            ]
+        )
+
+    @mock.patch("subprocess.Popen", autospec=True)
+    def test_missing_key(self, mock_popen):
+        mock_popen.return_value = self.create_mock_process(PUBLIC_CERT_BYTES, b"")
+        success, cert, key, output, error = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, output, error),
+                    (False, None, None, PUBLIC_CERT_BYTES, b""),
+                )
+            ]
+        )
+
+    @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")
+        success, cert, key, output, error = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, output, error),
+                    (False, None, None, b"", b"some error"),
+                )
+            ]
+        )
+
+    @mock.patch("subprocess.Popen", autospec=True)
+    def test_popen_raise_exception(self, mock_popen):
+        mock_popen.side_effect = OSError()
+        success, cert, key, output, error = _mtls_helper.get_client_ssl_credentials(
+            CONTEXT_AWARE_METADATA, "linux"
+        )
+
+        assert all(
+            [
+                a == b
+                for a, b in zip(
+                    (success, cert, key, output, error), (False, None, None, None, None)
+                )
+            ]
+        )