cmd/tsrecorder: adds sending api level logging to tsrecorder

Signed-off-by: chaosinthecrd <tom@tmlabs.co.uk>

cmd/k8s-operator/sts.go

@@ -61,7 +61,7 @@ const (
 	AnnotationHostname           = "tailscale.com/hostname"
 	annotationTailnetTargetIPOld = "tailscale.com/ts-tailnet-target-ip"
 	AnnotationTailnetTargetIP    = "tailscale.com/tailnet-ip"
-	//MagicDNS name of tailnet node.
+	// MagicDNS name of tailnet node.
 	AnnotationTailnetTargetFQDN = "tailscale.com/tailnet-fqdn"
 
 	AnnotationProxyGroup = "tailscale.com/proxy-group"

go.mod

@@ -136,6 +136,7 @@ require (
 	github.com/alecthomas/go-check-sumtype v0.1.4 // indirect
 	github.com/alexkohler/nakedret/v2 v2.0.4 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
+	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/boltdb/bolt v1.3.1 // indirect
 	github.com/bombsimon/wsl/v4 v4.2.1 // indirect
 	github.com/butuzov/mirror v1.1.0 // indirect
@@ -186,6 +187,7 @@ require (
 	go.uber.org/automaxprocs v1.5.3 // indirect
 	golang.org/x/xerrors v0.0.0-20240716161551-93cc26a95ae9 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
+	k8s.io/component-base v0.32.0 // indirect
 )
 
 require (

go.sum

@@ -176,6 +176,8 @@ github.com/bkielbasa/cyclop v1.2.1 h1:AeF71HZDob1P2/pRm1so9cd1alZnrpyc4q2uP2l0gJ
 github.com/bkielbasa/cyclop v1.2.1/go.mod h1:K/dT/M0FPAiYjBgQGau7tz+3TMh4FWAEqlMhzFWCrgM=
 github.com/blakesmith/ar v0.0.0-20190502131153-809d4375e1fb h1:m935MPodAbYS46DG4pJSv7WO+VECIWUQ7OJYSoTrMh4=
 github.com/blakesmith/ar v0.0.0-20190502131153-809d4375e1fb/go.mod h1:PkYb9DJNAwrSvRx5DYA+gUcOIgTGVMNkfSCbZM8cWpI=
+github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
+github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/blizzy78/varnamelen v0.8.0 h1:oqSblyuQvFsW1hbBHh1zfwrKe3kcSj0rnXkKzsQ089M=
 github.com/blizzy78/varnamelen v0.8.0/go.mod h1:V9TzQZ4fLJ1DSrjVDfl89H7aMnTvKkApdHeyESmyR7k=
 github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
@@ -1546,6 +1548,8 @@ k8s.io/apiserver v0.32.0 h1:VJ89ZvQZ8p1sLeiWdRJpRD6oLozNZD2+qVSLi+ft5Qs=
 k8s.io/apiserver v0.32.0/go.mod h1:HFh+dM1/BE/Hm4bS4nTXHVfN6Z6tFIZPi649n83b4Ag=
 k8s.io/client-go v0.32.0 h1:DimtMcnN/JIKZcrSrstiwvvZvLjG0aSxy8PxN8IChp8=
 k8s.io/client-go v0.32.0/go.mod h1:boDWvdM1Drk4NJj/VddSLnx59X3OPgwrOo0vGbtq9+8=
+k8s.io/component-base v0.32.0 h1:d6cWHZkCiiep41ObYQS6IcgzOUQUNpywm39KVYaUqzU=
+k8s.io/component-base v0.32.0/go.mod h1:JLG2W5TUxUu5uDyKiH2R/7NnxJo1HlPoRIIbVLkK5eM=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f h1:GA7//TjRY9yWGy1poLzYYJJ4JRdzg3+O6e8I+e+8T5Y=

k8s-operator/api-proxy/proxy.go

@@ -6,10 +6,13 @@
 package apiproxy
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/http/httputil"
@@ -19,17 +22,20 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/transport"
 	"tailscale.com/client/local"
 	"tailscale.com/client/tailscale/apitype"
-	"tailscale.com/k8s-operator/sessionrecording"
 	ksr "tailscale.com/k8s-operator/sessionrecording"
 	"tailscale.com/kube/kubetypes"
+	"tailscale.com/sessionrecording"
 	"tailscale.com/tailcfg"
 	"tailscale.com/tsnet"
 	"tailscale.com/util/clientmetric"
 	"tailscale.com/util/ctxkey"
+	"tailscale.com/util/multierr"
 	"tailscale.com/util/set"
 )
 
@@ -192,7 +198,15 @@ func (ap *APIServerProxy) serveDefault(w http.ResponseWriter, r *http.Request) {
 		ap.authError(w, err)
 		return
 	}
+
+	err = ap.recordRequestAsEvent(r, who)
+	if err != nil {
+		ap.log.Errorf("error recording Kubernetes API request: %v", err)
+		return
+	}
+
 	counterNumRequestsProxied.Add(1)
+
 	ap.rp.ServeHTTP(w, r.WithContext(whoIsKey.WithValue(r.Context(), who)))
 }
 
@@ -220,7 +234,7 @@ func (ap *APIServerProxy) serveAttachWS(w http.ResponseWriter, r *http.Request)
 	ap.sessionForProto(w, r, ksr.AttachSessionType, ksr.WSProtocol)
 }
 
-func (ap *APIServerProxy) sessionForProto(w http.ResponseWriter, r *http.Request, sessionType sessionrecording.SessionType, proto ksr.Protocol) {
+func (ap *APIServerProxy) sessionForProto(w http.ResponseWriter, r *http.Request, sessionType ksr.SessionType, proto ksr.Protocol) {
 	const (
 		podNameKey       = "pod"
 		namespaceNameKey = "namespace"
@@ -232,6 +246,13 @@ func (ap *APIServerProxy) sessionForProto(w http.ResponseWriter, r *http.Request
 		ap.authError(w, err)
 		return
 	}
+
+	err = ap.recordRequestAsEvent(r, who)
+	if err != nil {
+		ap.log.Errorf("error recording Kubernetes API request: %v", err)
+		return
+	}
+
 	counterNumRequestsProxied.Add(1)
 	failOpen, addrs, err := determineRecorderConfig(who)
 	if err != nil {
@@ -283,6 +304,79 @@ func (ap *APIServerProxy) sessionForProto(w http.ResponseWriter, r *http.Request
 	ap.rp.ServeHTTP(h, r.WithContext(whoIsKey.WithValue(r.Context(), who)))
 }
 
+func (ap *APIServerProxy) recordRequestAsEvent(req *http.Request, who *apitype.WhoIsResponse) error {
+	failOpen, addrs, err := determineRecorderConfig(who)
+	if err != nil {
+		ap.log.Errorf("error trying to determine whether the kubernetes api request needs to be recorded: %v", err)
+		return err
+	}
+	if failOpen && len(addrs) == 0 { // will not send event
+		return err
+	}
+
+	if !failOpen && len(addrs) == 0 {
+		ap.log.Errorf("forbidden: kubernetes api request must be recorded, but no recorders are available.")
+		return err
+	}
+
+	factory := &request.RequestInfoFactory{
+		APIPrefixes:          sets.NewString("api", "apis"),
+		GrouplessAPIPrefixes: sets.NewString("api"),
+	}
+
+	reqInfo, err := factory.NewRequestInfo(req)
+	if err != nil {
+		ap.log.Errorf("Error parsing request %s %s: %v", req.Method, req.URL.Path, err)
+		return err
+	}
+
+	event := &sessionrecording.Event{
+		Timestamp:  time.Now(),
+		Kubernetes: *reqInfo,
+		Request: sessionrecording.Request{
+			Method: req.Method,
+			Path:   reqInfo.Path,
+		},
+		User: sessionrecording.User{
+			Email: who.UserProfile.LoginName,
+		},
+	}
+
+	if ct := req.Header.Get("Content-Type"); strings.Contains(ct, "application/json") {
+		bodyBytes, err := io.ReadAll(req.Body)
+		if err != nil {
+			ap.log.Errorf("Failed to read body: %v", err)
+			return err
+		}
+
+		req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
+
+		event.Request.Body = bodyBytes
+	}
+
+	var errs []error
+	for _, ad := range addrs {
+		eventJSON, err := json.Marshal(event)
+		if err != nil {
+			ap.log.Errorf("Error marshaling request event: %v", err)
+			return err
+		}
+
+		data := bytes.NewBuffer(eventJSON)
+
+		ap.log.Infof("event json looks like %q", string(eventJSON))
+
+		if err := sessionrecording.SendEvent(req.Context(), ad, data, ap.ts.Dial); err != nil {
+			ap.log.Errorf("Error sending event to recorder with address %q: %v", ad.String(), err)
+			errs = append(errs, err)
+		} else {
+			return nil
+		}
+	}
+
+	return multierr.New(errs...)
+}
+
 func (ap *APIServerProxy) addImpersonationHeadersAsRequired(r *http.Request) {
 	r.URL.Scheme = ap.upstreamURL.Scheme
 	r.URL.Host = ap.upstreamURL.Host

sessionrecording/connect.go

@@ -113,6 +113,52 @@ func supportsV2(ctx context.Context, hc *http.Client, ap netip.AddrPort) bool {
 	return resp.StatusCode == http.StatusOK && resp.ProtoMajor > 1
 }
 
+// supportsEvent checks whether a recorder instance supports the /v2/event
+// endpoint.
+func supportsEvent(ctx context.Context, hc *http.Client, ap netip.AddrPort) bool {
+	ctx, cancel := context.WithTimeout(ctx, http2ProbeTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, httpm.HEAD, fmt.Sprintf("http://%s/v2/event", ap), nil)
+	if err != nil {
+		return false
+	}
+	resp, err := hc.Do(req)
+	if err != nil {
+		return false
+	}
+	defer resp.Body.Close()
+	return resp.StatusCode == http.StatusOK
+}
+
+// SendEvent sends an event the tsrecorders /v2/event endpoint.
+func SendEvent(ctx context.Context, ap netip.AddrPort, event io.Reader, dial netx.DialFunc) error {
+	client := clientHTTP2(ctx, dial)
+
+	if !supportsEvent(ctx, client, ap) {
+		return fmt.Errorf("recorder at address %q does not support `/v2/event` endpoint", ap.String())
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://%s/v2/event", ap.String()), event)
+	if err != nil {
+		return fmt.Errorf("error creating request: %v", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("error sending request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("server returned non-OK status: %s", resp.Status)
+	}
+
+	resp.Body.Close()
+	return nil
+}
+
 // connectV1 connects to the legacy /record endpoint on the recorder. It is
 // used for backwards-compatibility with older tsrecorder instances.
 //