feat: add private eks handling

**Returns:** Plain text URL to access the forwarded port (e.g., `http://aws-auth-provider:14240`)

#### Query Parameters

| Name          | Description                                                   | Required              |

|---------------|---------------------------------------------------------------|-----------------------|

| `instance_id` | EC2 instance ID to connect to                                 | yes                   |

| `remote_port` | Port on the remote instance                                   | yes                   |

| `remote_host` | Remote host for advanced port forwarding scenarios           | no                    |

| `local_port`  | Local port to bind (auto-allocated if not specified)          | no                    |

| `protocol`    | URL protocol (`http` or `https`, default: `http`)            | no                    |

| `region`      | AWS region to use                                             | no _(can be retrieved from env)_ |

| `env_ctx`     | the [environment context to consider](#the-notion-of-env_ctx) | no _(can be guessed)_ |

| `role_arn`    | AWS IAM role ARN to assume                                    | no _(can be retrieved from env)_ |

### `GET /kubeconfig`

This API generates a complete kubeconfig file for an AWS EKS cluster with a valid authentication token.

- Cluster endpoint and certificate authority

- Pre-authenticated token (matches `aws eks get-token` format)

- Configurable namespace and user name

- **Automatic private endpoint handling** via SSM port forwarding

The authentication token is generated using AWS STS and is compatible with EKS authentication requirements.

#### Private Endpoint Support

This endpoint automatically detects private EKS clusters (clusters with `endpointPublicAccess: false` and `endpointPrivateAccess: true`) and establishes an SSM port forwarding session to provide connectivity.

**How it works:**

1. The API checks the cluster's VPC configuration and automatically determines if it's a private cluster

2. Then it establishes an SSM Session Manager tunnel through the specified EC2 instance to forward traffic to the private Kubernetes API server

3. Then it configures the kubeconfig with appropriate TLS settings (insecure-skip-tls-verify for proxied connections)

4. And validates the connection by making a test request to the Kubernetes API `/version` endpoint to verify the certificate

**Requirements for private clusters:**

- You must provide an `instance_id` parameter pointing to a running EC2 instance in the EKS cluster's VPC with SSM agent installed and configured

- Appropriate IAM permissions for SSM Session Manager and Cluster Access

- Instance must be reachable from the cluster's subnets

You can also **force SSM tunneling** for public clusters by providing the `instance_id` parameter.

#### Query Parameters

| Name           | Description                                                   | Required              | Default         |

| `ttl_minutes`  | Time-to-live for the token in minutes (max: 15)               | no                    | `15`            |

| `namespace`    | Default namespace for kubectl context                         | no                    | `default`       |

| `user_name`    | Username in the kubeconfig                                    | no                    | `kubectl-user`  |

| `instance_id`  | EC2 instance ID for SSM port forwarding                       | **yes** _(for private clusters only)_ | -               |

#### Example Usage

```bash

# Generate kubeconfig for a cluster

# Generate kubeconfig for a public cluster

curl -s "http://aws-auth-provider/kubeconfig?cluster_name=my-cluster&region=us-east-1" > kubeconfig.yaml

# Generate kubeconfig for a private cluster (requires instance_id)

curl -s "http://aws-auth-provider/kubeconfig?cluster_name=my-private-cluster&region=us-east-1&instance_id=i-1234567890abcdef0" > kubeconfig.yaml

# Force SSM tunneling for a public cluster with a specific instance

curl -s "http://aws-auth-provider/kubeconfig?cluster_name=my-cluster&instance_id=i-1234567890abcdef0" > kubeconfig.yaml

# Use the kubeconfig

export KUBECONFIG=kubeconfig.yaml

kubectl get nodes

#### Example Response

**Public Cluster:**

```yaml

apiVersion: v1

kind: Config

  cluster-version: '1.33'

  cluster-status: ACTIVE

  authentication-mode: API

  endpoint-public-access: true

  endpoint-private-access: false

  ssm-forwarded: false

  note: If authentication fails, ensure your AWS IAM user/role is mapped in the cluster's access entries (for API mode) or aws-auth ConfigMap (for CONFIG_MAP mode)

```

**Private Cluster (with SSM Port Forwarding):**

```yaml

apiVersion: v1

kind: Config

clusters:

- name: my-private-cluster

  cluster:

    server: https://localhost:12345

    insecure-skip-tls-verify: true

contexts:

- name: my-private-cluster-context

  context:

    cluster: my-private-cluster

    user: kubectl-user

    namespace: default

current-context: my-private-cluster-context

users:

- name: kubectl-user

  user:

    token: k8s-aws-v1.aHR0cHM6Ly9zdHMuYW1hem9u...

preferences: {}

metadata:

  token-ttl-minutes: 15

  token-expires-at: '2025-10-05T12:15:00Z'

  cluster-arn: arn:aws:eks:us-east-1:123456789012:cluster/my-private-cluster

  cluster-version: '1.33'

  cluster-status: ACTIVE

  authentication-mode: API

  endpoint-public-access: false

  endpoint-private-access: true

  ssm-forwarded: true

  original-endpoint: https://ABC123.gr7.us-east-1.eks.amazonaws.com

  tls-hostname-verification: disabled

  note: This is a PRIVATE cluster. SSM port forwarding has been established. TLS hostname verification is disabled (insecure-skip-tls-verify: true) because traffic is proxied through localhost. Ensure your AWS IAM user/role is mapped in the cluster's access entries (for API mode) or aws-auth ConfigMap (for CONFIG_MAP mode). The SSM session will remain active as long as the aws-auth-provider service is running.

  authentication-mode: API

```

#### Notes

- Ensure your AWS credentials have `eks:DescribeCluster` permission for the cluster and permissions to access the cluster

| Name          | Description                                                   | Required              |

|---------------|---------------------------------------------------------------|-----------------------|

| `instance_id` | EC2 instance ID to connect to                                 | yes                   |

| `remote_port` | Port on the remote instance                                   | yes                   |

| `remote_host` | Remote host for advanced port forwarding scenarios           | no                    |

| `local_port`  | Local port to bind (auto-allocated if not specified)          | no                    |

| `protocol`    | URL protocol (`http` or `https`, default: `http`)            | no                    |

| `region`      | AWS region to use                                             | no _(can be retrieved from env)_ |

| `env_ctx`     | the [environment context to consider](#the-notion-of-env_ctx) | no _(can be guessed)_ |

| `role_arn`    | AWS IAM role ARN to assume                                    | no _(can be retrieved from env)_ |

**Environment Variables:**

- `API_HOST`: Host for generated URLs (default: `aws-auth-provider`)

import base64

import json

import logging

import ssl

import tempfile

from datetime import datetime, timedelta

from urllib.parse import urlparse

import boto3

import urllib3

import yaml

from botocore.signers import RequestSigner

from fastapi import HTTPException, Query

from . import ssm

from .utils import configure_boto

logger = logging.getLogger("aws-auth-provider.kubeconfig")

def verify_proxy_connection(

    proxy_url: str, expected_hostname: str, ca_data: str, timeout: int = 10

) -> dict:

    """

    Verify the proxy connection by making a request to the Kubernetes API

    and validating the TLS certificate.

    Args:

        proxy_url: The proxy URL to test (e.g., https://localhost:12345)

        expected_hostname: The expected hostname in the certificate

        ca_data: Base64-encoded CA certificate data

        timeout: Request timeout in seconds

    Returns:

        Dictionary with verification results

    Raises:

        HTTPException: If verification fails

    """

    try:

        logger.info(f"Verifying proxy connection to {proxy_url}")

        # Decode the CA certificate and write to a temporary file

        ca_cert_pem = base64.b64decode(ca_data).decode("utf-8")

        # Write CA cert to a temporary file

        with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f:

            f.write(ca_cert_pem)

            ca_cert_path = f.name

        try:

            # Create SSL context with the CA cert and enable hostname verification

            ssl_context = ssl.create_default_context(cafile=ca_cert_path)

            # Enable hostname verification to validate against expected hostname

            ssl_context.check_hostname = True

            ssl_context.verify_mode = ssl.CERT_REQUIRED

            # Create HTTP pool with custom SSL context

            # We need to use server_hostname to set SNI for proper cert validation

            http = urllib3.PoolManager(

                ssl_context=ssl_context,

                timeout=urllib3.Timeout(connect=timeout, read=timeout),

                cert_reqs="CERT_REQUIRED",

                ca_certs=ca_cert_path,

                # Set the expected hostname for SNI

                assert_hostname=expected_hostname,

                server_hostname=expected_hostname,

            )

            # parse the proxy URL to extract host and port

            parsed_url = urlparse(proxy_url)

            proxy_url_protocol = parsed_url.scheme

            proxy_url_port = parsed_url.port or (

                443 if proxy_url_protocol == "https" else 80

            )

            # Make a request to /version endpoint

            # Use the proxy URL but with server_hostname set for SNI

            response = http.request(

                "GET",

                f"{proxy_url_protocol}://localhost:{proxy_url_port}/version",

                headers={"Host": expected_hostname},  # Set Host header

            )

            logger.info(

                f"Proxy verification successful: status={response.status}, "

                f"response_size={len(response.data)} bytes, "

                f"validated hostname={expected_hostname}"

            )

            return {

                "success": True,

                "message": "Proxy connection verified successfully",

            }

        finally:

            # Clean up the temporary file

            import os

            try:

                os.unlink(ca_cert_path)

            except Exception:

                pass

    except urllib3.exceptions.SSLError as e:

        logger.error(f"SSL verification failed: {str(e)}")

        raise HTTPException(

            status_code=503,

            detail=f"SSL certificate verification failed through proxy: {str(e)}",

        )

    except urllib3.exceptions.TimeoutError:

        logger.error(f"Connection timeout after {timeout}s")

        raise HTTPException(

            status_code=503,

            detail=f"Timeout connecting to Kubernetes API through proxy after {timeout}s",

        )

    except urllib3.exceptions.HTTPError as e:

        logger.error(f"HTTP error: {str(e)}")

        raise HTTPException(

            status_code=503,

            detail=f"Failed to connect to Kubernetes API through proxy: {str(e)}",

        )

    except Exception as e:

        logger.error(f"Unexpected error during verification: {str(e)}")

        raise HTTPException(

            status_code=503,

            detail=f"Unexpected error verifying proxy connection: {str(e)}",

        )

def get_eks_token(cluster_name, ttl_minutes=15) -> str:

    """

    Generate an EKS authentication token that matches the format of 'aws eks get-token'.

    Args:

        cluster_name: Name of the EKS cluster

        ttl_minutes: Time-to-live in minutes (default: 15, max: 15 for security best practices)

    Returns:

        EKS authentication token in k8s-aws-v1 format

    """

    # Cap TTL at 15 minutes as recommended by AWS for security

    # STS presigned URLs support longer TTLs, but EKS recommends short-lived tokens

    ttl_minutes = min(ttl_minutes, 15)

    session = boto3.session.Session()

    client = session.client(

        "sts", region_name="us-east-1"

    )  # Always use us-east-1 for global endpoint

    service_id = client.meta.service_model.service_id

    signer = RequestSigner(

        service_id,

        "us-east-1",  # Global STS endpoint uses us-east-1

        "sts",

        "v4",

        session.get_credentials(),

        session.events,

    )

    params = {

        "method": "GET",

        "url": "https://sts.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15",  # Global endpoint

        "body": {},

        "headers": {

            "x-k8s-aws-id": cluster_name,

        },

        "context": {},

    }

    # Generate presigned URL with the header included in the signature

    signed_url = signer.generate_presigned_url(

        params,

        region_name="us-east-1",

        expires_in=ttl_minutes * 60,  # Convert minutes to seconds

        operation_name="",

    )

    # Kubernetes expects the token prefixed with "k8s-aws-v1."

    token = "k8s-aws-v1." + base64.urlsafe_b64encode(signed_url.encode("utf-8")).decode(

        "utf-8"

    ).rstrip("=")

    return token

def generate_kubeconfig(

    cluster_name: str = Query(alias="cluster_name"),

    env_ctx: str = Query(default=None, alias="env_ctx"),

    region: str = Query(default=None, alias="region"),

    role_arn: str = Query(default=None, alias="role_arn"),

    ttl_minutes: int = Query(default=15, ge=1, le=15, alias="ttl_minutes"),

    namespace: str = Query(default="default", alias="namespace"),

    user_name: str = Query(default="kubectl-user", alias="user_name"),

    instance_id: str = Query(default=None, alias="instance_id"),

) -> str:

    """

    Generate a kubeconfig file by retrieving cluster information from AWS EKS.

    The kubeconfig will be AWS-agnostic with a token that has the specified TTL.

    Args:

        cluster_name: Name of the EKS cluster

        env_ctx: Environment context for AWS authentication

        region: AWS region where the cluster is located

        role_arn: AWS role ARN for authentication

        ttl_minutes: Time-to-live for the token in minutes (default: 15, max: 15 for security)

        namespace: Default namespace (default: "default")

        user_name: Username for the kubeconfig (default: "kubectl-user")

        instance_id: EC2 instance ID for SSM port forwarding (required for private clusters, optional for public clusters)

    Returns:

        YAML-formatted kubeconfig as a string

    """

    configure_boto(env_ctx, region, role_arn)

    # Get EKS cluster information

    eks_client = boto3.client("eks")

    try:

        cluster_response = eks_client.describe_cluster(name=cluster_name)

        cluster = cluster_response["cluster"]

        logger.info(f"Retrieved cluster info for {cluster_name}")

        logger.debug(f"Cluster structure: {json.dumps(cluster, indent=2, default=str)}")

        # Check cluster status

        cluster_status = cluster.get("status", "UNKNOWN")

        if cluster_status != "ACTIVE":

            logger.warning(

                f"Cluster {cluster_name} is not ACTIVE (status: {cluster_status})"

            )

            if cluster_status == "CREATING":

                raise HTTPException(

                    status_code=503,

                    detail=f"EKS cluster '{cluster_name}' is still being created. Please wait for it to become ACTIVE.",

                )

            elif cluster_status in ["FAILED", "DELETING"]:

                raise HTTPException(

                    status_code=503,

                    detail=f"EKS cluster '{cluster_name}' is in {cluster_status} state and cannot be accessed.",

                )

    except HTTPException:

        raise  # Re-raise HTTP exceptions

    except Exception as e:

        logger.error(f"Failed to describe EKS cluster {cluster_name}: {str(e)}")

        raise HTTPException(

            status_code=404,

            detail=f"EKS cluster '{cluster_name}' not found or not accessible: {str(e)}",

        )

    # Extract cluster information

    cluster_endpoint = cluster.get("endpoint")

    if not cluster_endpoint:

        raise HTTPException(

            status_code=500, detail=f"EKS cluster '{cluster_name}' endpoint not found"

        )

    # Check if cluster is private or if instance_id is provided (force SSM)

    resources_vpc_config = cluster.get("resourcesVpcConfig", {})

    endpoint_public_access = resources_vpc_config.get("endpointPublicAccess", True)

    endpoint_private_access = resources_vpc_config.get("endpointPrivateAccess", False)

    is_private_cluster = not endpoint_public_access and endpoint_private_access

    use_ssm = is_private_cluster or instance_id is not None

    # Save original endpoint for metadata and verification

    original_endpoint = cluster_endpoint

    if use_ssm:

        # For private clusters, instance_id is required

        if is_private_cluster and not instance_id:

            raise HTTPException(

                status_code=400,

                detail=f"Private cluster '{cluster_name}' requires 'instance_id' parameter. "

                f"Please provide an EC2 instance ID with SSM agent in the cluster's VPC for port forwarding.",

            )

        logger.info(

            f"Instance ID provided, using SSM port forwarding for cluster {cluster_name}"

        )

        logger.info(f"Using provided instance ID: {instance_id}")

        # Parse the cluster endpoint to get host and port

        original_endpoint = cluster_endpoint  # Save original endpoint before proxying

        parsed_url = urlparse(cluster_endpoint)

        remote_host = parsed_url.hostname

        remote_port = parsed_url.port or 443

        logger.info(

            f"Setting up SSM port forward to {remote_host}:{remote_port} via instance {instance_id}"

        )

        # Start SSM port forwarding

        try:

            forwarded_response = ssm.start_ssm_port_forward(

                env_ctx=env_ctx,

                region=region,

                role_arn=role_arn,

                instance_id=instance_id,

                remote_port=remote_port,

                remote_host=remote_host,

                local_port=None,  # Let it find an available port

                protocol="https",

            )

            # Extract the forwarded URL from the response object

            cluster_endpoint = forwarded_response.url

            logger.info(f"SSM port forward established: {cluster_endpoint}")

        except Exception as e:

            logger.error(f"Failed to set up SSM port forward: {str(e)}")

            raise HTTPException(

                status_code=503,

                detail=f"Failed to establish SSM connection to private cluster '{cluster_name}': {str(e)}",

            )

    # Extract certificate authority data

    cert_authority = cluster.get("certificateAuthority")

    if not cert_authority:

        raise HTTPException(

            status_code=500,

            detail=f"Certificate authority not found for cluster '{cluster_name}'",

        )

    cluster_ca_data = cert_authority.get("data")

    if not cluster_ca_data:

        raise HTTPException(

            status_code=500,

            detail=f"Certificate authority data not found for cluster '{cluster_name}'",

        )

    # Verify the proxy connection if using SSM

    if use_ssm:

        logger.info("Verifying proxy connection and certificate...")

        verification_result = verify_proxy_connection(

            proxy_url=cluster_endpoint,

            expected_hostname=urlparse(original_endpoint).hostname,

            ca_data=cluster_ca_data,

            timeout=10,

        )

        logger.info(

            f"Proxy verification completed: {verification_result.get('message')}"

        )

    # Generate token with specified TTL

    token = get_eks_token(cluster_name, ttl_minutes=ttl_minutes)

    # Calculate expiration time

    expiry_time = datetime.utcnow() + timedelta(minutes=ttl_minutes)

    # Access configuration (if provided by EKS)

    access_config = cluster.get("accessConfig", {}) if isinstance(cluster, dict) else {}

    # Create kubeconfig structure

    metadata = {

        "token-ttl-minutes": ttl_minutes,

        "token-expires-at": expiry_time.isoformat() + "Z",

        "cluster-arn": cluster.get("arn", ""),

        "cluster-version": cluster.get("version", ""),

        "cluster-status": cluster.get("status", ""),

        "authentication-mode": access_config.get("authenticationMode", "CONFIG_MAP"),

        "endpoint-public-access": endpoint_public_access,

        "endpoint-private-access": endpoint_private_access,

    }

    if use_ssm:

        if is_private_cluster:

            note = (

                "This is a PRIVATE cluster. SSM port forwarding has been established. "

                "TLS hostname verification is disabled (insecure-skip-tls-verify: true) because traffic is proxied through localhost. "

                "Ensure your AWS IAM user/role is mapped in the cluster's access entries (for API mode) "

                "or aws-auth ConfigMap (for CONFIG_MAP mode). "

                "The SSM session will remain active as long as the aws-auth-provider service is running."

            )

        else:

            note = (

                "SSM port forwarding has been forced via instance_id parameter. "

                "TLS hostname verification is disabled (insecure-skip-tls-verify: true) because traffic is proxied through localhost. "

                "Ensure your AWS IAM user/role is mapped in the cluster's access entries (for API mode) "

                "or aws-auth ConfigMap (for CONFIG_MAP mode). "

                "The SSM session will remain active as long as the aws-auth-provider service is running."

            )

        metadata["note"] = note

        metadata["ssm-forwarded"] = True

        metadata["original-endpoint"] = original_endpoint

        metadata["tls-hostname-verification"] = "disabled"

    else:

        metadata["note"] = (

            "If authentication fails, ensure your AWS IAM user/role is mapped in the cluster's "

            "access entries (for API mode) or aws-auth ConfigMap (for CONFIG_MAP mode)"

        )

        metadata["ssm-forwarded"] = False

    # Build cluster configuration

    cluster_config = {

        "server": cluster_endpoint,

    }

    # When using SSM proxy, skip TLS hostname verification since we're connecting via localhost/proxy

    if use_ssm:

        cluster_config["insecure-skip-tls-verify"] = True

    else:

        # Only include certificate-authority-data if not using SSM

        cluster_config["certificate-authority-data"] = cluster_ca_data

    kubeconfig = {

        "apiVersion": "v1",

        "kind": "Config",

        "clusters": [

            {

                "name": cluster_name,

                "cluster": cluster_config,

            }

        ],

        "contexts": [

            {

                "name": f"{cluster_name}-context",

                "context": {

                    "cluster": cluster_name,

                    "user": user_name,

                    "namespace": namespace,

                },

            }

        ],

        "current-context": f"{cluster_name}-context",

        "users": [{"name": user_name, "user": {"token": token}}],

        "preferences": {},

        "metadata": metadata,

    }

    # Convert to YAML and return

    return yaml.dump(kubeconfig, default_flow_style=False, sort_keys=False)