240 lines
7.1 KiB
Go
240 lines
7.1 KiB
Go
package outbound
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"encoding/pem"
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
appsv1 "k8s.io/api/apps/v1"
|
|
corev1 "k8s.io/api/core/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/util/httpstream"
|
|
streamspdy "k8s.io/apimachinery/pkg/util/httpstream/spdy"
|
|
"k8s.io/client-go/rest"
|
|
"k8s.io/client-go/tools/portforward"
|
|
)
|
|
|
|
type persistentKubernetesFixture struct {
|
|
config *rest.Config
|
|
upgrades atomic.Int32
|
|
pod atomic.Value
|
|
connections chan httpstream.Connection
|
|
paths chan string
|
|
closed chan struct{}
|
|
}
|
|
|
|
func newPersistentKubernetesFixture(t *testing.T) *persistentKubernetesFixture {
|
|
t.Helper()
|
|
f := &persistentKubernetesFixture{connections: make(chan httpstream.Connection, 16), paths: make(chan string, 16), closed: make(chan struct{}, 16)}
|
|
f.pod.Store("db-0")
|
|
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.Header.Get("Authorization") != "Bearer persistent-kube-secret" {
|
|
http.Error(w, "denied", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
pod := f.pod.Load().(string)
|
|
switch r.URL.Path {
|
|
case "/apis/apps/v1/namespaces/testing/statefulsets/mongo":
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(&appsv1.StatefulSet{Spec: appsv1.StatefulSetSpec{Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "mongo"}}}})
|
|
case "/api/v1/namespaces/testing/pods":
|
|
if r.URL.Query().Get("labelSelector") != "app=mongo" {
|
|
t.Error("incorrect Pod selector")
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(&corev1.PodList{Items: []corev1.Pod{{ObjectMeta: metav1.ObjectMeta{Name: pod}, Status: corev1.PodStatus{Phase: corev1.PodRunning}}}})
|
|
case "/api/v1/namespaces/testing/pods/" + pod + "/portforward":
|
|
if _, err := httpstream.Handshake(r, w, []string{portforward.PortForwardProtocolV1Name}); err != nil {
|
|
return
|
|
}
|
|
conn := streamspdy.NewResponseUpgrader().UpgradeResponse(w, r, func(stream httpstream.Stream, reply <-chan struct{}) error {
|
|
go func() {
|
|
<-reply
|
|
defer stream.Close()
|
|
if stream.Headers().Get(corev1.PortHeader) != "3306" {
|
|
t.Error("incorrect remote port")
|
|
}
|
|
if stream.Headers().Get(corev1.StreamType) == corev1.StreamTypeData {
|
|
io.Copy(stream, stream)
|
|
}
|
|
}()
|
|
return nil
|
|
})
|
|
if conn != nil {
|
|
f.upgrades.Add(1)
|
|
f.paths <- r.URL.Path
|
|
f.connections <- conn
|
|
<-conn.CloseChan()
|
|
f.closed <- struct{}{}
|
|
}
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}))
|
|
t.Cleanup(server.Close)
|
|
f.config = &rest.Config{Host: server.URL, BearerToken: "persistent-kube-secret", TLSClientConfig: rest.TLSClientConfig{CAData: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})}}
|
|
return f
|
|
}
|
|
|
|
func (f *persistentKubernetesFixture) open(t *testing.T) *kubernetesTunnel {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
tunnel, err := startKubernetesConfig(ctx, f.config, "testing", "statefulset/mongo", 3306)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { tunnel.Close() })
|
|
return tunnel
|
|
}
|
|
|
|
func TestKubernetesFixedLocalPort(t *testing.T) {
|
|
f := newPersistentKubernetesFixture(t)
|
|
available := f.open(t)
|
|
_, rawPort, _ := net.SplitHostPort(available.address)
|
|
port, _ := strconv.Atoi(rawPort)
|
|
available.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
tunnel, err := startKubernetesConfigWithLocalPort(ctx, f.config, "testing", "statefulset/mongo", 3306, port)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tunnel.Close()
|
|
if tunnel.address != net.JoinHostPort("127.0.0.1", rawPort) {
|
|
t.Fatalf("address = %s", tunnel.address)
|
|
}
|
|
c, err := tunnel.Dial(ctx, "tcp", "target:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := persistentEcho(c, "fixed-port"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c.Close()
|
|
other, err := startKubernetesConfigWithLocalPort(ctx, f.config, "testing", "statefulset/mongo", 3306, port)
|
|
if other != nil {
|
|
other.Close()
|
|
t.Fatal("accepted occupied fixed port")
|
|
}
|
|
if !errors.Is(err, errKubernetesLocalPort) {
|
|
t.Fatalf("collision error: %v", err)
|
|
}
|
|
tunnel.Close()
|
|
listener, err := net.Listen("tcp", tunnel.address)
|
|
if err != nil {
|
|
t.Fatalf("port not released: %v", err)
|
|
}
|
|
listener.Close()
|
|
}
|
|
|
|
func TestKubernetesPersistentMultiplexLifecycle(t *testing.T) {
|
|
f := newPersistentKubernetesFixture(t)
|
|
tunnel := f.open(t)
|
|
<-f.connections
|
|
_, rawPort, _ := net.SplitHostPort(tunnel.address)
|
|
port, _ := strconv.Atoi(rawPort)
|
|
if port < 49152 || port > 65535 {
|
|
t.Fatalf("not a high port: %d", port)
|
|
}
|
|
exercisePersistentTunnel(t, tunnel, f.upgrades.Load)
|
|
waitPersistentDone(t, f.closed)
|
|
}
|
|
|
|
func TestKubernetesPersistentHalfClose(t *testing.T) {
|
|
f := newPersistentKubernetesFixture(t)
|
|
tunnel := f.open(t)
|
|
<-f.connections
|
|
c, err := tunnel.Dial(context.Background(), "tcp", "target:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
c.SetDeadline(time.Now().Add(3 * time.Second))
|
|
if _, err := c.Write([]byte("last-request")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := c.(interface{ CloseWrite() error }).CloseWrite(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := io.ReadAll(c)
|
|
if err != nil || string(data) != "last-request" {
|
|
t.Fatalf("half-close: %q %v", data, err)
|
|
}
|
|
next, err := tunnel.Dial(context.Background(), "tcp", "target:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer next.Close()
|
|
if err := persistentEcho(next, "after-half-close"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestKubernetesPersistentBrokenTunnelResolvesNewPod(t *testing.T) {
|
|
f := newPersistentKubernetesFixture(t)
|
|
tunnel := f.open(t)
|
|
conn := <-f.connections
|
|
if path := <-f.paths; !strings.Contains(path, "/db-0/") {
|
|
t.Fatal(path)
|
|
}
|
|
conn.Close()
|
|
waitPersistentDone(t, tunnel.Done())
|
|
f.pod.Store("db-1")
|
|
next := f.open(t)
|
|
<-f.connections
|
|
if path := <-f.paths; !strings.Contains(path, "/db-1/") {
|
|
t.Fatal("rebuild retained old Pod: " + path)
|
|
}
|
|
if n := f.upgrades.Load(); n != 2 {
|
|
t.Fatalf("upgrade count = %d", n)
|
|
}
|
|
c, err := next.Dial(context.Background(), "tcp", "target:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
if err := persistentEcho(c, "new-pod"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if c, err := next.Dial(context.Background(), "tcp", "target:5432"); err == nil {
|
|
c.Close()
|
|
t.Fatal("accepted different target port")
|
|
}
|
|
}
|
|
|
|
func TestKubernetesPersistentRejectsCredentialsAndUntrustedTLS(t *testing.T) {
|
|
f := newPersistentKubernetesFixture(t)
|
|
for _, mode := range []string{"token", "ca"} {
|
|
config := rest.CopyConfig(f.config)
|
|
if mode == "token" {
|
|
config.BearerToken = "wrong-private-token"
|
|
} else {
|
|
config.CAData = nil
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
tunnel, err := startKubernetesConfig(ctx, config, "testing", "statefulset/mongo", 3306)
|
|
cancel()
|
|
if err == nil {
|
|
tunnel.Close()
|
|
t.Fatalf("accepted %s", mode)
|
|
}
|
|
if strings.Contains(err.Error(), config.BearerToken) {
|
|
t.Fatal("token echoed")
|
|
}
|
|
}
|
|
if n := f.upgrades.Load(); n != 0 {
|
|
t.Fatalf("unauthorized upgrades=%d", n)
|
|
}
|
|
}
|