LCORE-511: Fix type checking issues in k8s authentication module

Fix type checking errors in `src/authentication/k8s.py` caused by incomplete
type annotations in the Kubernetes Python client library.

The Kubernetes Python client library (`kubernetes-client`) has incomplete type
annotations for many of its model classes and API responses. This PR adds
targeted type ignore comments to suppress false positives while maintaining type
safety for the rest of the codebase.

Changes:
- Added explicit type conversions for `k8s_config.host` and `k8s_config.ssl_ca_cert` to ensure proper string types
- Added `# type: ignore` directives for Kubernetes client objects that lack proper type stubs:
  - `V1TokenReview.status` and `.user` attributes
  - `V1SubjectAccessReview.status.allowed` attribute
  - Custom object API responses (dict-like objects with dynamic structure)
- Added type assertions to help Pyright understand singleton initialization pattern

Note: There is no functional changes to authentication logic being introduced in this PR,
code behaviour remains identical.

Signed-off-by: Anik Bhattacharjee <anbhatta@redhat.com>

Author: anik120

pyproject.toml

@@ -74,10 +74,6 @@ dependencies = [
 
 [tool.pyright]
 exclude = [
-  # TODO(lucasagomes): This module was copied from road-core
-  # service/ols/src/auth/k8s.py and currently has 58 Pyright issues. It
-  # might need to be rewritten down the line.
-  "src/authentication/k8s.py",
   # Agent API v1 endpoints - deprecated API but still supported
   # Type errors due to llama-stack-client not exposing Agent API types
   "src/app/endpoints/conversations.py",

src/authentication/k8s.py

@@ -80,19 +80,19 @@ def __new__(cls: type[Self]) -> Self:
                             ce,
                         )
 
-                k8s_config.host = (
-                    configuration.authentication_configuration.k8s_cluster_api
-                    or k8s_config.host
-                )
+                k8s_api_url = configuration.authentication_configuration.k8s_cluster_api
+                if k8s_api_url:
+                    k8s_config.host = str(k8s_api_url)
                 k8s_config.verify_ssl = (
                     not configuration.authentication_configuration.skip_tls_verification
                 )
-                k8s_config.ssl_ca_cert = (
+                ca_cert_path = (
                     configuration.authentication_configuration.k8s_ca_cert_path
-                    if configuration.authentication_configuration.k8s_ca_cert_path
-                    not in {None, Path()}
-                    else k8s_config.ssl_ca_cert
                 )
+                if ca_cert_path and ca_cert_path != Path():
+                    # Kubernetes client library has incomplete type stubs for ssl_ca_cert
+                    k8s_config.ssl_ca_cert = str(ca_cert_path)  # type: ignore[assignment]
+                # else keep the default k8s_config.ssl_ca_cert
                 api_client = kubernetes.client.ApiClient(k8s_config)
                 cls._api_client = api_client
                 cls._custom_objects_api = kubernetes.client.CustomObjectsApi(api_client)
@@ -101,7 +101,10 @@ def __new__(cls: type[Self]) -> Self:
             except Exception as e:
                 logger.info("Failed to initialize Kubernetes client: %s", e)
                 raise
-        return cls._instance
+        # At this point _instance is guaranteed to be initialized
+        # but we need this anyway for type checker
+        assert cls._instance is not None
+        return cls._instance  # type: ignore[return-value]
 
     @classmethod
     def get_authn_api(cls) -> kubernetes.client.AuthenticationV1Api:
@@ -159,20 +162,24 @@ def _get_cluster_id(cls) -> str:
             ClusterIDUnavailableError: If the cluster ID cannot be obtained due
             to missing keys, an API error, or any unexpected error.
         """
+        version_data = None
         try:
             custom_objects_api = cls.get_custom_objects_api()
             version_data = custom_objects_api.get_cluster_custom_object(
                 "config.openshift.io", "v1", "clusterversions", "version"
             )
-            cluster_id = version_data["spec"]["clusterID"]
+            # Type assertion: we know this returns a dict-like object
+            assert isinstance(version_data, dict), "version_data must be a dict"
+            # Kubernetes client library returns untyped dict-like objects
+            cluster_id = str(version_data["spec"]["clusterID"])  # type: ignore[index]
             cls._cluster_id = cluster_id
             return cluster_id
         except KeyError as e:
             logger.error(
                 "Failed to get cluster_id from cluster, missing keys in version object"
             )
             raise ClusterIDUnavailableError("Failed to get cluster ID") from e
-        except TypeError as e:
+        except (TypeError, AssertionError) as e:
             logger.error(
                 "Failed to get cluster_id, version object is: %s", version_data
             )
@@ -239,8 +246,9 @@ def get_user_info(token: str) -> Optional[kubernetes.client.V1TokenReview]:
     )
     try:
         response = auth_api.create_token_review(token_review)
-        if response.status.authenticated:
-            return response.status
+        # Kubernetes client library has incomplete type stubs
+        if response.status.authenticated:  # type: ignore[union-attr]
+            return response.status  # type: ignore[union-attr]
         return None
     except Exception as e:  # pylint: disable=broad-exception-caught
         logger.error("API exception during TokenReview: %s", e)
@@ -307,9 +315,10 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
             response = UnauthorizedResponse(cause="Invalid or expired Kubernetes token")
             raise HTTPException(**response.model_dump())
 
-        if user_info.user.username == "kube:admin":
+        # Kubernetes client library has incomplete type stubs
+        if user_info.user.username == "kube:admin":  # type: ignore[union-attr]
             try:
-                user_info.user.uid = K8sClientSingleton.get_cluster_id()
+                user_info.user.uid = K8sClientSingleton.get_cluster_id()  # type: ignore[union-attr]
             except ClusterIDUnavailableError as e:
                 logger.error("Failed to get cluster ID: %s", e)
                 response = InternalServerErrorResponse(
@@ -322,8 +331,8 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
             authorization_api = K8sClientSingleton.get_authz_api()
             sar = kubernetes.client.V1SubjectAccessReview(
                 spec=kubernetes.client.V1SubjectAccessReviewSpec(
-                    user=user_info.user.username,
-                    groups=user_info.user.groups,
+                    user=user_info.user.username,  # type: ignore[union-attr]
+                    groups=user_info.user.groups,  # type: ignore[union-attr]
                     non_resource_attributes=kubernetes.client.V1NonResourceAttributes(
                         path=self.virtual_path, verb="get"
                     ),
@@ -339,13 +348,14 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
             )
             raise HTTPException(**response.model_dump()) from e
 
-        if not response.status.allowed:
-            response = ForbiddenResponse.endpoint(user_id=user_info.user.uid)
+        # Kubernetes client library has incomplete type stubs
+        if not response.status.allowed:  # type: ignore[union-attr]
+            response = ForbiddenResponse.endpoint(user_id=user_info.user.uid)  # type: ignore[union-attr]
             raise HTTPException(**response.model_dump())
 
         return (
-            user_info.user.uid,
-            user_info.user.username,
+            user_info.user.uid,  # type: ignore[union-attr]
+            user_info.user.username,  # type: ignore[union-attr]
             self.skip_userid_check,
             token,
         )