fix(api): require inline kubeconfig credentials (#12952)

Co-authored-by: alejandrobailo <alejandrobailo94@gmail.com>

api/changelog.d/kubeconfig-inline-credentials.security.md

@@ -0,0 +1 @@
+Kubernetes kubeconfig validation accepts only inline credentials and known cluster fields, rejecting file-path fields (`tokenFile`, `client-certificate`, `client-key`, `certificate-authority`), command-based authentication (`exec`, `cmd-path`), `auth-provider` directives, proxy URLs and unknown user or cluster keys; stored kubeconfigs are re-validated before each scan and connection test

api/src/backend/api/tests/test_serializers.py

@@ -5,6 +5,8 @@ from api.v1.serializer_utils.integrations import (
 )
 from api.v1.serializer_utils.providers import ProviderSecretField
 from api.v1.serializers import (
+    KUBERNETES_KUBECONFIG_INVALID_ERROR,
+    KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
     ImageProviderSecret,
     IntegrationSerializer,
     IntegrationUpdateSerializer,
@@ -340,6 +342,278 @@ current-context: test-context
         assert not serializer.is_valid()
         assert "kubeconfig_content" in serializer.errors
 
+    @staticmethod
+    def _kubeconfig(user_extra="", cluster_extra=""):
+        return f"""
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+{cluster_extra}
+users:
+  - name: test-user
+    user:
+      token: test-token
+{user_extra}
+contexts:
+  - name: test-context
+    context:
+      cluster: test-cluster
+      user: test-user
+current-context: test-context
+"""
+
+    @staticmethod
+    def _assert_rejected_without_echo(
+        kubeconfig_content, leaked_value, expected_message
+    ):
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid() is False
+        assert serializer.errors["kubeconfig_content"] == [expected_message]
+        assert leaked_value not in str(serializer.errors)
+
+    def test_inline_token_with_certificate_authority_data_is_accepted(self):
+        kubeconfig_content = self._kubeconfig(
+            cluster_extra="      certificate-authority-data: dGVzdA=="
+        )
+
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid()
+
+    def test_kubeconfig_with_token_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      tokenFile: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_client_certificate_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      client-certificate: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_client_key_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      client-key: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_certificate_authority_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      certificate-authority: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_cluster_proxy_url_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      proxy-url: http://proxy.evil.test"),
+            "proxy.evil.test",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    @pytest.mark.parametrize("provider_name", ["gcp", "oidc", "azure"])
+    def test_kubeconfig_with_any_auth_provider_is_rejected(self, provider_name):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(
+                user_extra=(
+                    "      auth-provider:\n"
+                    f"        name: {provider_name}\n"
+                    "        config:\n"
+                    "          idp-issuer-url: https://idp.evil.test"
+                )
+            ),
+            "idp.evil.test",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_unknown_user_key_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      bogus: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_unknown_cluster_key_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      bogus: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_allowlisted_cluster_keys_is_accepted(self):
+        serializer = KubernetesProviderSecret(
+            data={
+                "kubeconfig_content": self._kubeconfig(
+                    cluster_extra=(
+                        "      certificate-authority-data: dGVzdA==\n"
+                        "      insecure-skip-tls-verify: false\n"
+                        "      tls-server-name: kubernetes.example.test\n"
+                        "      disable-compression: true\n"
+                        "      extensions: []"
+                    )
+                )
+            }
+        )
+
+        assert serializer.is_valid(), serializer.errors
+
+    def test_kubeconfig_with_non_list_clusters_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  name: /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    def test_kubeconfig_with_non_dict_cluster_entry_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  - /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    def test_kubeconfig_with_non_dict_cluster_value_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster: /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    @pytest.mark.parametrize(
+        "user_extra",
+        [
+            "      token: test-token",
+            "      username: test-user\n      password: test-password",
+            "      client-certificate-data: dGVzdA==\n      client-key-data: dGVzdA==",
+            "      token: test-token\n      as: other-user",
+            "      token: test-token\n      as-uid: 1234",
+            "      token: test-token\n      as-groups:\n        - group-a",
+            "      token: test-token\n      as-user-extra:\n        scopes:\n          - read",
+        ],
+        ids=[
+            "token",
+            "username-password",
+            "client-cert-data",
+            "as",
+            "as-uid",
+            "as-groups",
+            "as-user-extra",
+        ],
+    )
+    def test_kubeconfig_with_allowlisted_user_keys_is_accepted(self, user_extra):
+        kubeconfig_content = f"""
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+      certificate-authority-data: dGVzdA==
+users:
+  - name: test-user
+    user:
+{user_extra}
+contexts:
+  - name: test-context
+    context:
+      cluster: test-cluster
+      user: test-user
+current-context: test-context
+"""
+
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid()
+
+    @pytest.mark.parametrize(
+        "kubeconfig_content",
+        [
+            """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+users:
+  - name: good-user
+    user:
+      token: test-token
+  - name: bad-user
+    user:
+      tokenFile: /etc/passwd
+""",
+            """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: good-cluster
+    cluster:
+      server: https://kubernetes.example.test
+  - name: bad-cluster
+    cluster:
+      server: https://kubernetes.example.test
+      certificate-authority: /etc/passwd
+users:
+  - name: test-user
+    user:
+      token: test-token
+""",
+        ],
+        ids=["second-user", "second-cluster"],
+    )
+    def test_kubeconfig_with_bad_second_entry_is_rejected(self, kubeconfig_content):
+        self._assert_rejected_without_echo(
+            kubeconfig_content,
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_indented_document_is_rejected(self):
+        kubeconfig_content = (
+            "  wrapper:\n"
+            "  users:\n"
+            "  - name: u\n"
+            "    user:\n"
+            "      tokenFile: /etc/passwd\n"
+            "  clusters:\n"
+            "  - name: c\n"
+            "    cluster:\n"
+            "      server: https://example.test\n"
+        )
+
+        self._assert_rejected_without_echo(
+            kubeconfig_content,
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
     def test_malformed_kubeconfig_is_rejected(self):
         serializer = KubernetesProviderSecret(
             data={"kubeconfig_content": "apiVersion: ["}

api/src/backend/api/tests/test_utils.py

@@ -14,6 +14,7 @@ from api.utils import (
     return_prowler_provider,
     validate_invitation,
 )
+from api.v1.serializers import KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
 from prowler.providers.alibabacloud.alibabacloud_provider import AlibabacloudProvider
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.security_hub.security_hub import SecurityHubConnection
@@ -35,6 +36,34 @@ from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvid
 from prowler.providers.vercel.vercel_provider import VercelProvider
 from rest_framework.exceptions import NotFound, ValidationError
 
+VALID_KUBECONFIG = """\
+apiVersion: v1
+kind: Config
+clusters:
+- name: test-cluster
+  cluster:
+    server: https://example.invalid:6443
+    certificate-authority-data: Y2E=
+users:
+- name: test-user
+  user:
+    token: test-token
+contexts:
+- name: test-context
+  context:
+    cluster: test-cluster
+    user: test-user
+current-context: test-context
+"""
+
+TOKEN_FILE_KUBECONFIG = VALID_KUBECONFIG.replace(
+    "token: test-token", "tokenFile: /etc/passwd"
+)
+
+EXEC_KUBECONFIG = VALID_KUBECONFIG.replace(
+    "token: test-token", "exec:\n      command: test-command"
+)
+
 
 class TestMergeDicts:
     def test_simple_merge(self):
@@ -288,6 +317,40 @@ class TestProwlerProviderConnectionTest:
             raise_on_exception=False,
         )
 
+    @patch("api.utils.return_prowler_provider")
+    def test_kubernetes_connection_test_rejects_invalid_kubeconfig(
+        self, mock_return_prowler_provider
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": TOKEN_FILE_KUBECONFIG}
+
+        connection = prowler_provider_connection_test(provider)
+
+        assert connection.is_connected is False
+        assert (
+            str(connection.error) == KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
+        )
+        mock_return_prowler_provider.return_value.test_connection.assert_not_called()
+
+    @patch("api.utils.return_prowler_provider")
+    def test_kubernetes_connection_test_with_valid_kubeconfig(
+        self, mock_return_prowler_provider
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": VALID_KUBECONFIG}
+
+        prowler_provider_connection_test(provider)
+
+        mock_return_prowler_provider.return_value.test_connection.assert_called_once_with(
+            kubeconfig_content=VALID_KUBECONFIG,
+            provider_id="provider_uid",
+            raise_on_exception=False,
+        )
+
     @pytest.mark.django_db
     @patch("api.utils.return_prowler_provider")
     def test_prowler_provider_connection_test_without_secret(
@@ -405,10 +468,6 @@ class TestGetProwlerProviderKwargs:
                 Provider.ProviderChoices.GOOGLEWORKSPACE.value,
                 {},
             ),
-            (
-                Provider.ProviderChoices.KUBERNETES.value,
-                {"context": "provider_uid"},
-            ),
             (
                 Provider.ProviderChoices.M365.value,
                 {},
@@ -459,6 +518,39 @@ class TestGetProwlerProviderKwargs:
         expected_result = {**secret_dict, **expected_extra_kwargs}
         assert result == expected_result
 
+    def test_get_prowler_provider_kwargs_kubernetes_valid_kubeconfig(self):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": VALID_KUBECONFIG}
+
+        assert get_prowler_provider_kwargs(provider) == {
+            "kubeconfig_content": VALID_KUBECONFIG,
+            "context": "provider_uid",
+        }
+
+    @pytest.mark.parametrize(
+        "secret",
+        [
+            {"kubeconfig_content": TOKEN_FILE_KUBECONFIG},
+            {"kubeconfig_content": EXEC_KUBECONFIG},
+            {"kubeconfig_content": "[]"},
+            {"key": "value"},
+        ],
+    )
+    def test_get_prowler_provider_kwargs_kubernetes_rejects_invalid_kubeconfig(
+        self, secret
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = secret
+
+        with pytest.raises(ValidationError) as exc_info:
+            get_prowler_provider_kwargs(provider)
+
+        assert "/etc/passwd" not in str(exc_info.value)
+
     def test_get_prowler_provider_kwargs_oraclecloud_maps_region_to_home_region(
         self,
     ):

api/src/backend/api/utils.py

@@ -17,7 +17,7 @@ from api.models import (
     Role,
     UserRoleRelationship,
 )
-from api.v1.serializers import FindingMetadataSerializer
+from api.v1.serializers import FindingMetadataSerializer, KubernetesProviderSecret
 from django.contrib.postgres.aggregates import ArrayAgg
 from django.db import transaction
 from django.db.models import Subquery
@@ -201,6 +201,13 @@ def return_prowler_provider(
     return prowler_provider
 
 
+def _validate_kubernetes_secret(secret: dict) -> None:
+    """Re-validate a stored kubeconfig at use time; errors never echo its content."""
+    KubernetesProviderSecret(
+        data={"kubeconfig_content": secret.get("kubeconfig_content")}
+    ).is_valid(raise_exception=True)
+
+
 def get_prowler_provider_kwargs(
     provider: Provider, mutelist_processor: Processor | None = None
 ) -> dict:
@@ -225,6 +232,7 @@ def get_prowler_provider_kwargs(
             "project_ids": [provider.uid],
         }
     elif provider.provider == Provider.ProviderChoices.KUBERNETES.value:
+        _validate_kubernetes_secret(prowler_provider_kwargs)
         prowler_provider_kwargs = {**prowler_provider_kwargs, "context": provider.uid}
     elif provider.provider == Provider.ProviderChoices.GITHUB.value:
         if provider.uid:
@@ -392,6 +400,16 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
     except Provider.secret.RelatedObjectDoesNotExist as secret_error:
         return Connection(is_connected=False, error=secret_error)
 
+    if provider.provider == Provider.ProviderChoices.KUBERNETES.value:
+        # A rejected stored kubeconfig is a failed connection, not a task error.
+        try:
+            _validate_kubernetes_secret(prowler_provider_kwargs)
+        except ValidationError as validation_error:
+            return Connection(
+                is_connected=False,
+                error=Exception(validation_error.detail["kubeconfig_content"][0]),
+            )
+
     # For IaC provider, construct the kwargs properly for test_connection
     if provider.provider == Provider.ProviderChoices.IAC.value:
         # Don't pass repository_url from secret, use scan_repository_url with the UID

api/src/backend/api/v1/serializer_utils/providers.py

@@ -214,7 +214,8 @@ from rest_framework_json_api import serializers
                     "kubeconfig_content": {
                         "type": "string",
                         "description": "The content of the Kubernetes kubeconfig file, encoded as a string. "
-                        "Kubeconfig command-based authentication is not supported in Prowler Cloud for security reasons.",
+                        "Only inline credentials are supported: token, username/password, or client-certificate-data with client-key-data for users, and certificate-authority-data for clusters. "
+                        "File-path and auth-provider kubeconfig fields are not supported.",
                     }
                 },
                 "required": ["kubeconfig_content"],

api/src/backend/api/v1/serializers.py

@@ -1659,14 +1659,43 @@ class FindingMetadataSerializer(BaseSerializerV1):
 
 
 # Provider secrets
-KUBERNETES_KUBECONFIG_UNSUPPORTED_COMMAND_AUTH_ERROR = (
-    "Kubernetes kubeconfig command-based authentication is not supported in "
-    "Prowler Cloud for security reasons."
+KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR = (
+    "Only inline Kubernetes credentials are supported. "
+    "For user authentication use token, username/password, or client-certificate-data with client-key-data; "
+    "clusters may use certificate-authority-data. "
+    "File-path fields (tokenFile, client-certificate, client-key, certificate-authority), "
+    "command-based authentication (exec, cmd-path), auth-provider directives, and proxy URLs are not supported."
 )
 KUBERNETES_KUBECONFIG_INVALID_ERROR = "Invalid Kubernetes kubeconfig content."
 
 
-def kubeconfig_contains_unsupported_command_auth(kubeconfig: dict) -> bool:
+KUBECONFIG_ALLOWED_USER_KEYS = frozenset(
+    {
+        "token",
+        "username",
+        "password",
+        "client-certificate-data",
+        "client-key-data",
+        "as",
+        "as-uid",
+        "as-groups",
+        "as-user-extra",
+    }
+)
+KUBECONFIG_ALLOWED_CLUSTER_KEYS = frozenset(
+    {
+        "server",
+        "certificate-authority-data",
+        "insecure-skip-tls-verify",
+        "tls-server-name",
+        "disable-compression",
+        "extensions",
+    }
+)
+
+
+def kubeconfig_is_inline_only_credentials(kubeconfig: dict) -> bool:
+    """Accept only inline credentials; reject file-reference and auth-provider fields."""
     users = kubeconfig.get("users", [])
     if not isinstance(users, list):
         raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
@@ -1679,21 +1708,25 @@ def kubeconfig_contains_unsupported_command_auth(kubeconfig: dict) -> bool:
         if not isinstance(user, dict):
             raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
 
-        if "exec" in user:
-            return True
+        if any(key not in KUBECONFIG_ALLOWED_USER_KEYS for key in user):
+            return False
 
-        auth_provider = user.get("auth-provider", {})
-        if not isinstance(auth_provider, dict):
-            continue
+    clusters = kubeconfig.get("clusters", [])
+    if not isinstance(clusters, list):
+        raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
 
-        auth_provider_config = auth_provider.get("config", {})
-        if not isinstance(auth_provider_config, dict):
-            continue
+    for cluster_entry in clusters:
+        if not isinstance(cluster_entry, dict):
+            raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
 
-        if "cmd-path" in auth_provider_config:
-            return True
+        cluster = cluster_entry.get("cluster", {})
+        if not isinstance(cluster, dict):
+            raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
 
-    return False
+        if any(key not in KUBECONFIG_ALLOWED_CLUSTER_KEYS for key in cluster):
+            return False
+
+    return True
 
 
 class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
@@ -1876,7 +1909,7 @@ class MongoDBAtlasProviderSecret(serializers.Serializer):
 
 
 class KubernetesProviderSecret(serializers.Serializer):
-    kubeconfig_content = serializers.CharField()
+    kubeconfig_content = serializers.CharField(trim_whitespace=False)
 
     def validate_kubeconfig_content(self, kubeconfig_content):
         try:
@@ -1889,9 +1922,9 @@ class KubernetesProviderSecret(serializers.Serializer):
         if not isinstance(kubeconfig, dict):
             raise serializers.ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
 
-        if kubeconfig_contains_unsupported_command_auth(kubeconfig):
+        if not kubeconfig_is_inline_only_credentials(kubeconfig):
             raise serializers.ValidationError(
-                KUBERNETES_KUBECONFIG_UNSUPPORTED_COMMAND_AUTH_ERROR
+                KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
             )
 
         return kubeconfig_content

api/src/backend/tasks/tests/test_connection.py

@@ -3,7 +3,8 @@ from datetime import UTC, datetime
 from unittest.mock import MagicMock, patch
 
 import pytest
-from api.models import Integration, LighthouseConfiguration, Provider
+from api.models import Integration, LighthouseConfiguration, Provider, ProviderSecret
+from api.v1.serializers import KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
 from tasks.jobs.connection import (
     check_integration_connection,
     check_lighthouse_connection,
@@ -65,17 +66,52 @@ def test_check_provider_connection_exception(
 
     mock_provider_connection_test.return_value = MagicMock()
     mock_provider_connection_test.return_value.is_connected = False
-    mock_provider_connection_test.return_value.error = Exception()
+    mock_provider_connection_test.return_value.error = Exception(
+        "Unable to assume role"
+    )
 
     result = check_provider_connection(provider_id="provider_id")
 
     assert result["connected"] is False
-    assert result["error"] is not None
+    assert result["error"] == "Unable to assume role"
 
     mock_provider_instance.save.assert_called_once()
     assert mock_provider_instance.connected is False
 
 
+@pytest.mark.django_db
+def test_check_provider_connection_with_stored_non_inline_kubeconfig(
+    kubernetes_provider,
+):
+    """A kubeconfig stored before the inline-only rule marks the provider as disconnected."""
+    kubernetes_provider.connected = True
+    kubernetes_provider.save()
+    ProviderSecret.objects.create(
+        tenant_id=kubernetes_provider.tenant_id,
+        provider=kubernetes_provider,
+        secret_type=ProviderSecret.TypeChoices.STATIC,
+        secret={
+            "kubeconfig_content": (
+                "apiVersion: v1\n"
+                "kind: Config\n"
+                "users:\n"
+                "- name: test-user\n"
+                "  user:\n"
+                "    tokenFile: /etc/passwd\n"
+            )
+        },
+    )
+
+    result = check_provider_connection(provider_id=str(kubernetes_provider.id))
+
+    kubernetes_provider.refresh_from_db()
+    assert result == {
+        "connected": False,
+        "error": KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+    }
+    assert kubernetes_provider.connected is False
+
+
 @pytest.mark.parametrize(
     "lighthouse_data",
     [