Retrieve desktop app client secret from gcp secret manager

Signed-off-by: Fabio Grätz <fabiogratz@googlemail.com>

plugins/flytekit-identity-aware-proxy/flytekitplugins/identity_aware_proxy/cli.py

@@ -2,6 +2,8 @@
 import typing
 
 import click
+from google.api_core.exceptions import NotFound
+from google.cloud import secretmanager
 
 from flytekit.clients.auth.auth_client import AuthorizationClient
 from flytekit.clients.auth.authenticator import Authenticator
@@ -85,6 +87,18 @@ def refresh_credentials(self):
         KeyringStore.store(self._creds)
 
 
+def get_gcp_secret_manager_secret(project_id: str, secret_id: str, version: typing.Optional[str] = "latest"):
+    """Retrieve secret from GCP secret manager."""
+    client = secretmanager.SecretManagerServiceClient()
+    name = f"projects/{project_id}/secrets/{secret_id}/versions/{version}"
+    try:
+        response = client.access_secret_version(name=name)
+    except NotFound as e:
+        raise click.BadParameter(e.message)
+    payload = response.payload.data.decode("UTF-8")
+    return payload
+
+
 @click.command()
 @click.option(
     "--desktop_client_id",
@@ -117,9 +131,18 @@ def refresh_credentials(self):
         "https://cloud.google.com/iap/docs/enabling-kubernetes-howto#oauth-credentials"
     ),
 )
-def flyte_iap_token(desktop_client_id: str, desktop_client_secret_gcp_secret_name: str, webapp_client_id: str):
+@click.option(
+    "--project",
+    type=str,
+    default=None,
+    required=True,
+    help="GCP project ID (in which `desktop_client_secret_gcp_secret_name` is saved).",
+)
+def flyte_iap_token(
+    desktop_client_id: str, desktop_client_secret_gcp_secret_name: str, webapp_client_id: str, project: str
+):
     """Generate an ID token for proxy-authentication/authorization with GCP Identity Aware Proxy."""
-    desktop_client_secret = desktop_client_secret_gcp_secret_name  # TODO
+    desktop_client_secret = get_gcp_secret_manager_secret(project, desktop_client_secret_gcp_secret_name)
 
     iap_authenticator = GCPIdentityAwareProxyAuthenticator(
         audience=webapp_client_id,

plugins/flytekit-identity-aware-proxy/setup.py

@@ -4,7 +4,7 @@
 
 microlib_name = f"flytekitplugins-{PLUGIN_NAME}"
 
-plugin_requires = ["click"]
+plugin_requires = ["click", "google-cloud-secret-manager"]
 
 __version__ = "0.0.0+develop"
 

plugins/flytekit-identity-aware-proxy/tests/test_flytekitplugins_iap.py

@@ -0,0 +1,46 @@
+from unittest.mock import MagicMock, patch
+
+import click
+import pytest
+from click.testing import CliRunner
+from flytekitplugins.identity_aware_proxy.cli import flyte_iap_token, get_gcp_secret_manager_secret
+from google.api_core.exceptions import NotFound
+
+
+def test_help() -> None:
+    """Smoke test external command IAP ID token generator cli by printing help message."""
+    runner = CliRunner()
+    result = runner.invoke(flyte_iap_token, "--help")
+    assert "Generate an ID token" in result.output
+    assert result.exit_code == 0
+
+
+def test_get_gcp_secret_manager_secret():
+    """Test retrieval of GCP secret manager secret."""
+    project_id = "test_project"
+    secret_id = "test_secret"
+    version = "latest"
+    expected_payload = "test_payload"
+
+    mock_client = MagicMock()
+    mock_client.access_secret_version.return_value.payload.data.decode.return_value = expected_payload
+    with patch("google.cloud.secretmanager.SecretManagerServiceClient", return_value=mock_client):
+        payload = get_gcp_secret_manager_secret(project_id, secret_id, version)
+        assert payload == expected_payload
+
+        name = f"projects/{project_id}/secrets/{secret_id}/versions/{version}"
+        mock_client.access_secret_version.assert_called_once_with(name=name)
+
+
+def test_get_gcp_secret_manager_secret_not_found():
+    """Test retrieving non-existing secret from GCP secret manager."""
+    project_id = "test_project"
+    secret_id = "test_secret"
+    version = "latest"
+
+    mock_client = MagicMock()
+    mock_client.access_secret_version.side_effect = NotFound("Secret not found")
+    with patch("google.cloud.secretmanager.SecretManagerServiceClient", return_value=mock_client):
+        # Call the get_gcp_secret_manager_secret function and assert that it raises a BadParameter exception
+        with pytest.raises(click.BadParameter):
+            get_gcp_secret_manager_secret(project_id, secret_id, version)