Entkube/src/EntKube.Web/Services/KubernetesS3Proxy.cs

using System.Text;
using Amazon.Runtime;
using Amazon.S3;
using k8s;
using k8s.Models;

namespace EntKube.Web.Services;

/// <summary>
/// Routes Amazon S3 SDK requests through the Kubernetes API server pod-proxy so that
/// cluster-internal MinIO pods are reachable from EntKube.Web.
///
/// The core challenge: AWS Signature V4 includes the Host header. The K8s pod proxy
/// rewrites Host to podIP:port when forwarding to the pod, causing a signature mismatch
/// when the SDK signs for the API server hostname.
///
/// Solution: sign with Host = podIP:port (what MinIO actually receives), then intercept
/// the request in KubernetesProxyHandler and rewrite the URL to the K8s pod-proxy path.
/// The signature remains valid; K8s routes based on the URL path, not the Host header.
///
///   SDK signs request for: http://podIP:9000/bucket?cors
///   Handler rewrites to:   https://k8s-api/api/v1/namespaces/{ns}/pods/{pod}:9000/proxy/bucket?cors
///   K8s forwards to pod:   http://podIP:9000/bucket?cors  (Host = podIP:9000)
///   MinIO validates:       Host matches signed value ✓
/// </summary>
public static class KubernetesS3Proxy
{
    // ── URL Parsing ──

    /// <summary>
    /// Parses a cluster-internal service URL of the form
    /// http(s)://svcName.namespace.svc.cluster.local[:port].
    /// Returns null if the URL doesn't match that pattern.
    /// </summary>
    public static (string Service, string Namespace, int Port)? ParseInternalServiceUrl(string url)
    {
        if (!Uri.TryCreate(url, UriKind.Absolute, out Uri? uri))
            return null;

        string[] parts = uri.Host.Split('.');
        if (parts.Length < 3 || !parts[2].Equals("svc", StringComparison.OrdinalIgnoreCase))
            return null;

        return (parts[0], parts[1], uri.Port > 0 ? uri.Port : (uri.Scheme == "https" ? 443 : 80));
    }

    // ── Open ──

    /// <summary>
    /// Resolves a ready MinIO pod, builds a correctly-signed S3 client that routes
    /// through the Kubernetes API server pod proxy, and returns both as a single
    /// disposable. Returns null if the endpoint URL cannot be parsed.
    ///
    /// Usage:
    ///   using var proxied = await KubernetesS3Proxy.OpenAsync(kubeconfig, endpoint, key, secret);
    ///   var result = await proxied.S3.ListBucketsAsync();
    /// </summary>
    public static async Task<ProxiedS3Client?> OpenAsync(
        string kubeconfig,
        string internalEndpoint,
        string accessKey,
        string secretKey,
        string region = "us-east-1",
        CancellationToken ct = default)
    {
        var parsed = ParseInternalServiceUrl(internalEndpoint);
        if (parsed is null)
            return null;

        using MemoryStream stream = new(Encoding.UTF8.GetBytes(kubeconfig));
        KubernetesClientConfiguration k8sCfg = KubernetesClientConfiguration.BuildConfigFromConfigFile(stream);
        Kubernetes k8s = new(k8sCfg);

        (string podName, string podIP) = await FindReadyPodAsync(
            k8s, parsed.Value.Namespace, parsed.Value.Service, ct);

        // Use the K8s SERVICE proxy instead of the pod proxy.
        // Pod proxy requires the API server to make a direct TCP connection to the pod IP,
        // which fails with 503 in clusters where the control plane cannot reach pod IPs
        // (common in managed/cloud K8s environments).
        // Service proxy routes through kube-proxy/iptables — the same path as normal
        // in-cluster traffic — and works regardless of control-plane ↔ pod connectivity.
        //
        // Look up the real service port (e.g. 443 for TLS MinIO, 9000 for plain HTTP).
        // The service port in the URL must match a port the service actually exposes.
        bool endpointTls = internalEndpoint.StartsWith("https://", StringComparison.OrdinalIgnoreCase);
        (int svcPort, bool svcTls) = await GetServicePortAsync(
            k8s, parsed.Value.Namespace, parsed.Value.Service, parsed.Value.Port, endpointTls, ct);

        string svcSchemePrefix = svcTls ? "https:" : "";

        // K8s service proxy URL: the scheme prefix tells K8s whether to use TLS when
        // connecting to the backend pod (InsecureSkipVerify=true for self-signed certs).
        string proxyBase = k8s.BaseUri.ToString().TrimEnd('/')
            + $"/api/v1/namespaces/{parsed.Value.Namespace}/services/{svcSchemePrefix}{parsed.Value.Service}:{svcPort}/proxy";

        // AWS Signature V4 is signed for the host MinIO actually receives.
        // K8s service proxy creates the backend request to podIP:podPort (from the Endpoints
        // resource), so MinIO sees Host = podIP:podPort — same as pod proxy.
        string signingScheme = svcTls ? "https" : "http";
        string signingBaseUrl = $"{signingScheme}://{podIP}:{parsed.Value.Port}";

        // Custom handler intercepts SDK requests (addressed to podIP:port) and reroutes to K8s.
        HttpClient proxyHttpClient = new(new KubernetesProxyHandler(k8s.HttpClient, proxyBase));

        AmazonS3Config config = new()
        {
            ServiceURL           = signingBaseUrl,
            ForcePathStyle       = true,
            AuthenticationRegion = region,
            UseHttp              = !svcTls,  // must match signingBaseUrl scheme
            Timeout              = TimeSpan.FromSeconds(30),
            MaxErrorRetry        = 0,
            HttpClientFactory    = new SingletonHttpClientFactory(proxyHttpClient)
        };

        AmazonS3Client s3 = new(new BasicAWSCredentials(accessKey, secretKey), config);
        return new ProxiedS3Client(k8s, s3, proxyHttpClient);
    }

    // ── Service Port Detection ──

    /// <summary>
    /// Looks up the K8s service to find the port that maps to the given pod port.
    /// Returns (servicePort=443, tls=true) for TLS MinIO (operator default),
    /// or (servicePort=podPort, tls=false) if no 443 mapping is found.
    /// </summary>
    /// <param name="endpointTls">Whether the stored endpoint uses https:// — determines the
    /// service proxy scheme prefix (https: tells K8s to use TLS for the backend connection).
    /// Service port 443 does NOT imply TLS; the endpoint scheme is the source of truth.</param>
    private static async Task<(int ServicePort, bool Tls)> GetServicePortAsync(
        Kubernetes k8s, string ns, string svc, int podPort, bool endpointTls, CancellationToken ct)
    {
        try
        {
            V1Service service = await k8s.CoreV1.ReadNamespacedServiceAsync(svc, ns, cancellationToken: ct);

            // Find the service port that targets our pod port.
            V1ServicePort? match = service.Spec?.Ports?.FirstOrDefault(p =>
                p.TargetPort?.Value == podPort.ToString()
                || (p.TargetPort?.Value == null && p.Port == podPort));

            if (match is not null)
                return (match.Port, endpointTls); // TLS from endpoint scheme, not service port

            // Fallback: use the first exposed port.
            int first = service.Spec?.Ports?.FirstOrDefault()?.Port ?? podPort;
            return (first, endpointTls);
        }
        catch
        {
            return (podPort, endpointTls);
        }
    }

    // ── Pod Discovery ──

    private static async Task<(string PodName, string PodIP)> FindReadyPodAsync(
        Kubernetes k8s, string ns, string tenantName, CancellationToken ct)
    {
        // Primary: MinIO operator labels pods with v1.min.io/tenant=<tenantName>.
        V1PodList pods = await k8s.CoreV1.ListNamespacedPodAsync(
            ns, labelSelector: $"v1.min.io/tenant={tenantName}", cancellationToken: ct);

        var found = PickReadyPodWithIP(pods);
        if (found is not null)
            return found.Value;

        // Fallback: standalone MinIO or different label convention.
        V1PodList all = await k8s.CoreV1.ListNamespacedPodAsync(ns, cancellationToken: ct);
        found = PickReadyPodWithIP(all);

        return found ?? throw new InvalidOperationException(
            $"No running MinIO pods with a pod IP found in namespace '{ns}'. " +
            "Ensure MinIO is healthy before accessing bucket management.");
    }

    private static (string PodName, string PodIP)? PickReadyPodWithIP(V1PodList podList)
    {
        foreach (V1Pod pod in podList.Items)
        {
            if (pod.Status?.Phase != "Running") continue;
            if (string.IsNullOrEmpty(pod.Status.PodIP)) continue;
            if (pod.Status.ContainerStatuses?.All(cs => cs.Ready) != true) continue;
            return (pod.Metadata.Name, pod.Status.PodIP);
        }
        return null;
    }

    // ── Proxy Handler ──

    /// <summary>
    /// Intercepts Amazon SDK requests addressed to http://podIP:port/path and
    /// rewrites the URL to the K8s pod-proxy endpoint before forwarding via the
    /// K8s-authenticated HttpClient. All headers (including Authorization) are
    /// preserved so AWS Signature V4 reaches MinIO intact.
    /// </summary>
    private sealed class KubernetesProxyHandler(HttpClient k8sClient, string proxyBase) : HttpMessageHandler
    {
        private readonly string _proxyBase = proxyBase.TrimEnd('/');

        protected override async Task<HttpResponseMessage> SendAsync(
            HttpRequestMessage request, CancellationToken cancellationToken)
        {
            string pathAndQuery = request.RequestUri?.PathAndQuery ?? "/";
            string targetUrl = _proxyBase + pathAndQuery;

            HttpRequestMessage proxied = new(request.Method, targetUrl);

            // Copy all headers except Host.
            // - Authorization (AWS Signature V4), X-Amz-Date, etc. must reach MinIO unchanged.
            // - Host must NOT be copied: the AWS SDK sets Host=podIP:port for signing, but
            //   the K8s API server validates Host against its own hostname and closes the
            //   connection if it doesn't match ("unexpected EOF" / SSL error).
            //   K8s will set Host=podIP:port itself when forwarding to the pod, which is
            //   exactly what MinIO needs to validate the signature.
            foreach (KeyValuePair<string, IEnumerable<string>> h in request.Headers)
            {
                if (string.Equals(h.Key, "Host", StringComparison.OrdinalIgnoreCase))
                    continue;
                proxied.Headers.TryAddWithoutValidation(h.Key, h.Value);
            }

            if (request.Content is not null)
                proxied.Content = request.Content;

            HttpResponseMessage response = await k8sClient.SendAsync(proxied, cancellationToken);

            // For 5xx errors from K8s, read the body so we can surface the real reason
            // (e.g. "error trying to reach service: ...", certificate errors, RBAC, etc.)
            // rather than a generic "ServiceUnavailable" from the Amazon SDK.
            if ((int)response.StatusCode >= 500)
            {
                string body = await response.Content.ReadAsStringAsync(cancellationToken);
                string snippet = body.Length > 600 ? body[..600] : body;
                throw new InvalidOperationException(
                    $"K8s proxy returned {(int)response.StatusCode} for {targetUrl} — {snippet}");
            }

            return response;
        }
    }

    // ── Amazon SDK HttpClientFactory injection ──

    private sealed class SingletonHttpClientFactory(HttpClient client) : Amazon.Runtime.HttpClientFactory
    {
        public override HttpClient CreateHttpClient(IClientConfig clientConfig) => client;
        public override bool UseSDKHttpClientCaching(IClientConfig clientConfig) => false;
        public override bool DisposeHttpClientsAfterUse(IClientConfig clientConfig) => false;
    }
}

/// <summary>
/// A paired K8s client + proxied S3 client. Dispose to release all three resources
/// in the correct order (S3 → proxyHttpClient → K8s).
/// </summary>
public sealed class ProxiedS3Client(Kubernetes k8s, AmazonS3Client s3, HttpClient proxyHttpClient) : IDisposable
{
    public AmazonS3Client S3 { get; } = s3;

    public void Dispose()
    {
        S3.Dispose();
        proxyHttpClient.Dispose();
        k8s.Dispose();
    }
}