[Bug]: kubectl ate proxy actor returns an empty response when the client half-closes (e.g. nc -N)

// Repro: when a client half-closes through `kubectl ate proxy actor`,
// forwardedConn.CloseWrite sends a TLS close_notify on the router's
// connect_terminate_tls listener and Envoy drops the actor's reply.
//
// Save as cmd/atenet/internal/router/repro_halfclose_test.go and run:
//
//	go test ./cmd/atenet/internal/router -run TestRepro_EnvoyConnectHalfCloseDropsActorResponse -v -count=1
//
// Needs Docker. Uses envoyproxy/envoy:v1.39.5 (the base of
// cmd/dataplane/envoy/Dockerfile) unless ENVOY_IMAGE is set.
package router

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/pem"
	"fmt"
	"io"
	"math/big"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/structpb"

	"go.ate.dev/substrate/cmd/atenet/internal/router/ingress"
	"go.ate.dev/substrate/internal/ateclient"
	"go.ate.dev/substrate/internal/atenet"
	"go.ate.dev/substrate/internal/resources"
)

type fixedUpstreamExtProc struct {
	extprocv3.UnimplementedExternalProcessorServer
	targetAddr string
}

func (f *fixedUpstreamExtProc) Process(stream extprocv3.ExternalProcessor_ProcessServer) error {
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		if req.GetRequestHeaders() != nil {
			dm, err := structpb.NewStruct(map[string]any{
				ingress.OriginalDstMetadataKey: map[string]any{
					ingress.OriginalDstAddressKey: f.targetAddr,
					ingress.OriginalDstPortKey:    "8000",
					ingress.WorkerSPIFFEIDKey:     "spiffe://cluster.local/ns/workers/sa/worker",
				},
			})
			if err != nil {
				return err
			}
			if err := stream.Send(&extprocv3.ProcessingResponse{
				DynamicMetadata: dm,
				Response: &extprocv3.ProcessingResponse_RequestHeaders{
					RequestHeaders: &extprocv3.HeadersResponse{},
				},
			}); err != nil {
				return err
			}
		}
	}
}

func freeTCPPort(t *testing.T, start int) int {
	t.Helper()
	for p := start; p < start+500; p++ {
		l4, err4 := net.Listen("tcp4", fmt.Sprintf("0.0.0.0:%d", p))
		if err4 != nil {
			continue
		}
		l6, err6 := net.Listen("tcp6", fmt.Sprintf("[::]:%d", p))
		_ = l4.Close()
		if err6 != nil {
			continue
		}
		_ = l6.Close()
		return p
	}
	t.Fatalf("no free dual-stack TCP port found starting at %d", start)
	return 0
}

func TestRepro_EnvoyConnectHalfCloseDropsActorResponse(t *testing.T) {
	// Same Envoy as cmd/dataplane/envoy/Dockerfile, which only adds an egress
	// module on top of it. ENVOY_IMAGE overrides it.
	envoyImage := os.Getenv("ENVOY_IMAGE")
	if envoyImage == "" {
		envoyImage = "envoyproxy/envoy:v1.39.5@sha256:8fc594a5d85eb76dca24155cfb9208882406873932b20cf33649a0a611194c43"
	}
	if out, err := exec.Command("docker", "image", "inspect", envoyImage).CombinedOutput(); err != nil {
		t.Skipf("envoy image %s not cached locally: %v (%s)", envoyImage, err, out)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
	defer cancel()

	// 1. Upstream HTTP server representing the actor (takes 80ms to generate response).
	actorLis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("actor listen: %v", err)
	}
	defer actorLis.Close()
	actorSrv := &http.Server{
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			time.Sleep(80 * time.Millisecond)
			w.Header().Set("Content-Type", "text/plain")
			_, _ = w.Write([]byte("hello from actor\n"))
		}),
	}
	go func() { _ = actorSrv.Serve(actorLis) }()
	defer actorSrv.Close()

	// 2. Minimal ext_proc server routing to actorLis.
	extLis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("extproc listen: %v", err)
	}
	defer extLis.Close()
	extGrpc := grpc.NewServer()
	extprocv3.RegisterExternalProcessorServer(extGrpc, &fixedUpstreamExtProc{targetAddr: actorLis.Addr().String()})
	go func() { _ = extGrpc.Serve(extLis) }()
	defer extGrpc.Stop()

	// 3. Generate CA, server bundle, and SPIFFE client cert in a world-readable temp dir for UID 65532.
	dir, err := os.MkdirTemp("/tmp", "envoy-repro-*")
	if err != nil {
		t.Fatalf("MkdirTemp: %v", err)
	}
	t.Cleanup(func() { _ = os.RemoveAll(dir) })
	_ = os.Chmod(dir, 0o755)

	caKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	caTmpl := &x509.Certificate{
		SerialNumber:          big.NewInt(1),
		Subject:               pkix.Name{CommonName: "test-ca"},
		NotBefore:             time.Now().Add(-time.Hour),
		NotAfter:              time.Now().Add(time.Hour),
		IsCA:                  true,
		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
		BasicConstraintsValid: true,
	}
	caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey)
	if err != nil {
		t.Fatalf("create CA: %v", err)
	}
	caCert, _ := x509.ParseCertificate(caDER)
	caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER})
	caPath := filepath.Join(dir, "ca.pem")
	if err := os.WriteFile(caPath, caPEM, 0o644); err != nil {
		t.Fatalf("write CA: %v", err)
	}
	// The container runs as another user, so make it readable whatever the umask.
	if err := os.Chmod(caPath, 0o644); err != nil {
		t.Fatalf("chmod caPath: %v", err)
	}

	srvKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	srvTmpl := &x509.Certificate{
		SerialNumber: big.NewInt(2),
		Subject:      pkix.Name{CommonName: "atenet-router.ate-system.svc"},
		DNSNames:     []string{"atenet-router.ate-system.svc"},
		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
		NotBefore:    time.Now().Add(-time.Hour),
		NotAfter:     time.Now().Add(time.Hour),
		KeyUsage:     x509.KeyUsageDigitalSignature,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
	}
	srvDER, err := x509.CreateCertificate(rand.Reader, srvTmpl, caCert, &srvKey.PublicKey, caKey)
	if err != nil {
		t.Fatalf("create server cert: %v", err)
	}
	srvKeyDER, _ := x509.MarshalECPrivateKey(srvKey)
	srvBundle := append(
		pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srvDER}),
		pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: srvKeyDER})...,
	)
	srvBundlePath := filepath.Join(dir, "server-bundle.pem")
	if err := os.WriteFile(srvBundlePath, srvBundle, 0o644); err != nil {
		t.Fatalf("write server bundle: %v", err)
	}
	// The container runs as another user, so make it readable whatever the umask.
	if err := os.Chmod(srvBundlePath, 0o644); err != nil {
		t.Fatalf("chmod srvBundlePath: %v", err)
	}

	const clientSPIFFE = "spiffe://cluster.local/ns/ate-system/sa/test-client"
	spiffeURL, _ := url.Parse(clientSPIFFE)
	cliKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	cliTmpl := &x509.Certificate{
		SerialNumber: big.NewInt(3),
		NotBefore:    time.Now().Add(-time.Hour),
		NotAfter:     time.Now().Add(time.Hour),
		URIs:         []*url.URL{spiffeURL},
		KeyUsage:     x509.KeyUsageDigitalSignature,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
	}
	cliDER, err := x509.CreateCertificate(rand.Reader, cliTmpl, caCert, &cliKey.PublicKey, caKey)
	if err != nil {
		t.Fatalf("create client cert: %v", err)
	}
	cliKeyDER, _ := x509.MarshalECPrivateKey(cliKey)
	cliTLSCert, err := tls.X509KeyPair(
		pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cliDER}),
		pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: cliKeyDER}),
	)
	if err != nil {
		t.Fatalf("X509KeyPair: %v", err)
	}

	// 4. Start production XdsServer serving ADS/SDS for connect_terminate_tls + main_internal.
	xdsLis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("xds listen: %v", err)
	}
	defer xdsLis.Close()
	xdsPort := xdsLis.Addr().(*net.TCPAddr).Port
	connectPort := freeTCPPort(t, 19100)
	adminPort := freeTCPPort(t, connectPort+1)

	xds := NewXdsServer(xdsPort)
	xds.SetConfig(extLis.Addr().(*net.TCPAddr).Port, "127.0.0.1")
	xds.SetTlsConfig(0, srvBundlePath)
	xds.SetConnectTLSPort(connectPort)
	xds.SetDownstreamClientAuth(caPath, []string{clientSPIFFE})
	if err := xds.UpdateSnapshot(); err != nil {
		t.Fatalf("UpdateSnapshot: %v", err)
	}
	xdsCtx, xdsCancel := context.WithCancel(ctx)
	defer xdsCancel()
	go func() { _ = xds.Serve(xdsCtx, xdsLis) }()

	// 5. Write bootstrap matching manifests/ate-install/atenet-router.yaml and launch Envoy.
	envoyYAML := fmt.Sprintf(`admin:
  address:
    socket_address:
      address: 127.0.0.1
      port_value: %d
node:
  id: %s
  cluster: substrate-router-cluster
bootstrap_extensions:
- name: envoy.bootstrap.internal_listener
  typed_config:
    "@type": type.googleapis.com/envoy.extensions.bootstrap.internal_listener.v3.InternalListener
dynamic_resources:
  lds_config:
    resource_api_version: V3
    ads: {}
  cds_config:
    resource_api_version: V3
    ads: {}
  ads_config:
    api_type: GRPC
    transport_api_version: V3
    grpc_services:
    - envoy_grpc:
        cluster_name: xds_cluster
static_resources:
  clusters:
  - name: xds_cluster
    connect_timeout: 0.25s
    type: STATIC
    typed_extension_protocol_options:
      envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
        "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
        explicit_http_config:
          http2_protocol_options: {}
    load_assignment:
      cluster_name: xds_cluster
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: %d
`, adminPort, NodeID, xdsPort)

	envoyCfgPath := filepath.Join(dir, "envoy.yaml")
	if err := os.WriteFile(envoyCfgPath, []byte(envoyYAML), 0o644); err != nil {
		t.Fatalf("write envoy.yaml: %v", err)
	}
	// The container runs as another user, so make it readable whatever the umask.
	if err := os.Chmod(envoyCfgPath, 0o644); err != nil {
		t.Fatalf("chmod envoyCfgPath: %v", err)
	}

	var envoyLogs bytes.Buffer
	containerName := fmt.Sprintf("envoy-halfclose-repro-%d", time.Now().UnixNano())
	cmd := exec.CommandContext(ctx, "docker", "run", "--rm", "--name", containerName, "--network=host",
		"--user", "65532:65532",
		"-v", dir+":"+dir+":ro",
		envoyImage,
		"/usr/local/bin/envoy", "-c", envoyCfgPath, "--concurrency", "1", "--log-level", "info",
	)
	cmd.Stdout = &envoyLogs
	cmd.Stderr = &envoyLogs
	if err := cmd.Start(); err != nil {
		t.Fatalf("start envoy container: %v", err)
	}
	defer func() {
		// Killing the docker client does not stop the container.
		_ = exec.Command("docker", "rm", "-f", containerName).Run()
		_ = cmd.Process.Kill()
		_ = cmd.Wait()
	}()

	// Wait for Envoy /ready and connect_terminate_tls port to accept TCP connections.
	readyDeadline := time.Now().Add(10 * time.Second)
	for {
		if time.Now().After(readyDeadline) {
			t.Fatalf("Envoy did not become ready in time; logs:\n%s", envoyLogs.String())
		}
		resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/ready", adminPort))
		if err == nil {
			body, _ := io.ReadAll(resp.Body)
			_ = resp.Body.Close()
			if strings.TrimSpace(string(body)) == "LIVE" {
				if c, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", connectPort), 100*time.Millisecond); err == nil {
					_ = c.Close()
					break
				}
			}
		}
		time.Sleep(50 * time.Millisecond)
	}

	pool := x509.NewCertPool()
	pool.AppendCertsFromPEM(caPEM)
	tlsCfg := &tls.Config{
		RootCAs:      pool,
		ServerName:   "atenet-router.ate-system.svc",
		Certificates: []tls.Certificate{cliTLSCert},
		MinVersion:   tls.VersionTLS13,
	}
	actorRef := resources.ActorRef{Atespace: "ns", Name: "actor1"}
	headers := http.Header{atenet.TargetActorHeader: []string{actorRef.String()}}
	const httpReq = "GET / HTTP/1.1\r\nHost: actor1:8000\r\nConnection: close\r\n\r\n"

	// Control check: WITHOUT CloseWrite, the CONNECT tunnel delivers the full response.
	ctrlConn, err := ateclient.ConnectActorWithHeaders(ctx, fmt.Sprintf("127.0.0.1:%d", connectPort), tlsCfg, actorRef, 8000, headers, nil)
	if err != nil {
		t.Fatalf("control ConnectActorWithHeaders: %v\nenvoy logs:\n%s", err, envoyLogs.String())
	}
	if _, err := io.WriteString(ctrlConn, httpReq); err != nil {
		t.Fatalf("control write: %v", err)
	}
	ctrlResp, err := io.ReadAll(ctrlConn)
	_ = ctrlConn.Close()
	if err != nil || !strings.Contains(string(ctrlResp), "hello from actor") {
		t.Fatalf("control tunnel without CloseWrite failed: err=%v resp=%q\nenvoy logs:\n%s", err, string(ctrlResp), envoyLogs.String())
	}

	// Repro check: WITH CloseWrite (what `kubectl ate proxy actor`'s copyHalf calls on client EOF),
	// forwardedConn.CloseWrite() sends TLS close_notify on :8444 and Envoy tears down the stream!
	halfConn, err := ateclient.ConnectActorWithHeaders(ctx, fmt.Sprintf("127.0.0.1:%d", connectPort), tlsCfg, actorRef, 8000, headers, nil)
	if err != nil {
		t.Fatalf("half-close ConnectActorWithHeaders: %v", err)
	}
	defer halfConn.Close()
	if _, err := io.WriteString(halfConn, httpReq); err != nil {
		t.Fatalf("half-close write: %v", err)
	}
	if cw, ok := halfConn.(interface{ CloseWrite() error }); !ok {
		t.Fatal("ConnectActorWithHeaders connection does not implement CloseWrite")
	} else if err := cw.CloseWrite(); err != nil {
		t.Fatalf("CloseWrite: %v", err)
	}
	gotResp, err := io.ReadAll(halfConn)
	if err != nil || !strings.Contains(string(gotResp), "hello from actor") {
		t.Fatalf("CloseWrite on CONNECT tunnel aborted actor response: err=%v got=%q (len=%d)\nenvoy logs:\n%s",
			err, string(gotResp), len(gotResp), envoyLogs.String())
	}
}