Merge pull request #46335 from sttts/sttts-audit-stages

Automatic merge from submit-queue (batch tested with PRs 45699, 46200, 46335, 46599)

audit: implement stages

Based on https://github.com/kubernetes/kubernetes/pull/46065. Implementing https://github.com/kubernetes/kubernetes/pull/46065#discussion_r117438531.

staging/src/k8s.io/apiserver/pkg/apis/audit/types.go

@@ -61,6 +61,8 @@ const (
 	// The stage for events generated once the response body has been completed, and no more bytes
 	// will be sent.
 	StageResponseComplete = "ResponseComplete"
+	// The stage for events generated when a panic occured.
+	StagePanic = "Panic"
 )
 
 // Event captures all the information that can be included in an API audit log.

staging/src/k8s.io/apiserver/pkg/apis/audit/v1alpha1/types.go

@@ -62,6 +62,8 @@ const (
 	// The stage for events generated once the response body has been completed, and no more bytes
 	// will be sent.
 	StageResponseComplete = "ResponseComplete"
+	// The stage for events generated when a panic occured.
+	StagePanic = "Panic"
 )
 
 // Event captures all the information that can be included in an API audit log.

staging/src/k8s.io/apiserver/pkg/audit/request.go

@@ -94,10 +94,11 @@ func NewEventFromRequest(req *http.Request, level auditinternal.Level, attribs a
 
 	if attribs.IsResourceRequest() {
 		ev.ObjectRef = &auditinternal.ObjectReference{
-			Namespace:  attribs.GetNamespace(),
-			Name:       attribs.GetName(),
-			Resource:   attribs.GetResource(),
-			APIVersion: attribs.GetAPIGroup() + "/" + attribs.GetAPIVersion(),
+			Namespace:   attribs.GetNamespace(),
+			Name:        attribs.GetName(),
+			Resource:    attribs.GetResource(),
+			Subresource: attribs.GetSubresource(),
+			APIVersion:  attribs.GetAPIGroup() + "/" + attribs.GetAPIVersion(),
 		}
 	}
 

staging/src/k8s.io/apiserver/pkg/endpoints/audit_test.go

@@ -127,7 +127,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			200,
-			1,
+			2,
 			[]eventCheck{
 				noRequestBody(0),
 				responseBodyMatches(0, `{.*"name":"c".*}`),
@@ -144,7 +144,7 @@ func TestAudit(t *testing.T) {
 				namespace:   "other",
 			},
 			200,
-			1,
+			2,
 			[]eventCheck{
 				noRequestBody(0),
 				responseBodyMatches(0, `{.*"name":"a".*"name":"b".*}`),
@@ -157,7 +157,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			201,
-			1,
+			2,
 			[]eventCheck{
 				requestBodyIs(0, string(simpleFooJSON)),
 				responseBodyMatches(0, `{.*"foo".*}`),
@@ -170,7 +170,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			405,
-			1,
+			2,
 			[]eventCheck{
 				noRequestBody(0),  // the 405 is thrown long before the create handler would be executed
 				noResponseBody(0), // the 405 is thrown long before the create handler would be executed
@@ -183,7 +183,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			200,
-			1,
+			2,
 			[]eventCheck{
 				noRequestBody(0),
 				responseBodyMatches(0, `{.*"kind":"Status".*"status":"Success".*}`),
@@ -196,7 +196,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			200,
-			1,
+			2,
 			[]eventCheck{
 				requestBodyMatches(0, "DeleteOptions"),
 				responseBodyMatches(0, `{.*"kind":"Status".*"status":"Success".*}`),
@@ -209,7 +209,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			200,
-			1,
+			2,
 			[]eventCheck{
 				requestBodyIs(0, string(simpleCPrimeJSON)),
 				responseBodyMatches(0, `{.*"bla".*}`),
@@ -222,7 +222,7 @@ func TestAudit(t *testing.T) {
 			},
 			selfLinker,
 			400,
-			1,
+			2,
 			[]eventCheck{
 				requestBodyIs(0, string(simpleCPrimeJSON)),
 				responseBodyMatches(0, `"Status".*"status":"Failure".*"code":400}`),
@@ -242,7 +242,7 @@ func TestAudit(t *testing.T) {
 				namespace:   "other",
 			},
 			200,
-			1,
+			2,
 			[]eventCheck{
 				requestBodyIs(0, `{"labels":{"foo":"bar"}}`),
 				responseBodyMatches(0, `"name":"c".*"labels":{"foo":"bar"}`),
@@ -259,7 +259,7 @@ func TestAudit(t *testing.T) {
 				namespace:   "other",
 			},
 			200,
-			2,
+			3,
 			[]eventCheck{
 				noRequestBody(0),
 				noResponseBody(0),

staging/src/k8s.io/apiserver/pkg/endpoints/filters/audit.go

@@ -88,12 +88,14 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 			return
 		}
 
+		ev.Stage = auditinternal.StageRequestReceived
+		sink.ProcessEvents(ev)
+
 		// intercept the status code
-		longRunning := false
 		var longRunningSink audit.Sink
 		if longRunningCheck != nil {
 			ri, _ := request.RequestInfoFrom(ctx)
-			if longRunning = longRunningCheck(req, ri); longRunning {
+			if longRunningCheck(req, ri) {
 				longRunningSink = sink
 			}
 		}
@@ -103,19 +105,34 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 		// running requests, this will be the second audit event.
 		defer func() {
 			if r := recover(); r != nil {
+				defer panic(r)
+				ev.Stage = auditinternal.StagePanic
 				ev.ResponseStatus = &metav1.Status{
-					Code: http.StatusInternalServerError,
+					Code:    http.StatusInternalServerError,
+					Status:  metav1.StatusFailure,
+					Reason:  metav1.StatusReasonInternalError,
+					Message: fmt.Sprintf("APIServer panic'd: %v", r),
 				}
 				sink.ProcessEvents(ev)
-				panic(r)
+				return
 			}
 
+			// if no StageResponseStarted event was sent b/c neither a status code nor a body was sent, fake it here
+			fakedSuccessStatus := &metav1.Status{
+				Code:    http.StatusOK,
+				Status:  metav1.StatusSuccess,
+				Message: "Connection closed early",
+			}
+			if ev.ResponseStatus == nil && longRunningSink != nil {
+				ev.ResponseStatus = fakedSuccessStatus
+				ev.Stage = auditinternal.StageResponseStarted
+				longRunningSink.ProcessEvents(ev)
+			}
+
+			ev.Stage = auditinternal.StageResponseComplete
 			if ev.ResponseStatus == nil {
-				ev.ResponseStatus = &metav1.Status{
-					Code: 200,
-				}
+				ev.ResponseStatus = fakedSuccessStatus
 			}
-
 			sink.ProcessEvents(ev)
 		}()
 		handler.ServeHTTP(respWriter, req)
@@ -153,17 +170,15 @@ type auditResponseWriter struct {
 
 func (a *auditResponseWriter) processCode(code int) {
 	a.once.Do(func() {
-		if a.sink != nil {
-			a.sink.ProcessEvents(a.event)
-		}
-
-		// for now we use the ResponseStatus as marker that it's the first or second event
-		// of a long running request. As soon as we have such a field in the event, we can
-		// change this.
 		if a.event.ResponseStatus == nil {
 			a.event.ResponseStatus = &metav1.Status{}
 		}
 		a.event.ResponseStatus.Code = int32(code)
+		a.event.Stage = auditinternal.StageResponseStarted
+
+		if a.sink != nil {
+			a.sink.ProcessEvents(a.event)
+		}
 	})
 }
 

staging/src/k8s.io/apiserver/pkg/endpoints/filters/audit_test.go

@@ -19,6 +19,7 @@ package filters
 import (
 	"bufio"
 	"bytes"
+	"fmt"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -177,10 +178,17 @@ func (*fakeHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 }
 
 func TestAudit(t *testing.T) {
-	shortRunningPrefix := `[\d\:\-\.\+TZ]+ AUDIT: id="[\w-]+" ip="127.0.0.1" method="list" user="admin" groups="<none>" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods"`
-	longRunningPrefix := `[\d\:\-\.\+TZ]+ AUDIT: id="[\w-]+" ip="127.0.0.1" method="watch" user="admin" groups="<none>" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods\?watch=true"`
+	writingShortRunningPrefix := func(stage string) string {
+		return fmt.Sprintf(`[\d\:\-\.\+TZ]+ AUDIT: id="[\w-]+" stage="%s" ip="127.0.0.1" method="update" user="admin" groups="<none>" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods/foo"`, stage)
+	}
+	readOnlyShortRunningPrefix := func(stage string) string {
+		return fmt.Sprintf(`[\d\:\-\.\+TZ]+ AUDIT: id="[\w-]+" stage="%s" ip="127.0.0.1" method="get" user="admin" groups="<none>" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods/foo"`, stage)
+	}
+	longRunningPrefix := func(stage string) string {
+		return fmt.Sprintf(`[\d\:\-\.\+TZ]+ AUDIT: id="[\w-]+" stage="%s" ip="127.0.0.1" method="watch" user="admin" groups="<none>" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods\?watch=true"`, stage)
+	}
 
-	shortRunningPath := "/api/v1/namespaces/default/pods"
+	shortRunningPath := "/api/v1/namespaces/default/pods/foo"
 	longRunningPath := "/api/v1/namespaces/default/pods?watch=true"
 
 	delay := 500 * time.Millisecond
@@ -188,58 +196,93 @@ func TestAudit(t *testing.T) {
 	for _, test := range []struct {
 		desc     string
 		path     string
+		verb     string
 		handler  func(http.ResponseWriter, *http.Request)
 		expected []string
 	}{
-		// short running requests
+		// short running requests with read-only verb
 		{
-			"empty",
+			"read-only empty",
 			shortRunningPath,
+			"GET",
 			func(http.ResponseWriter, *http.Request) {},
 			[]string{
-				shortRunningPrefix + ` response="200"`,
+				readOnlyShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				readOnlyShortRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
 			},
 		},
 		{
-			"sleep",
+			"read-only panic",
 			shortRunningPath,
+			"GET",
+			func(w http.ResponseWriter, req *http.Request) {
+				panic("kaboom")
+			},
+			[]string{
+				readOnlyShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				readOnlyShortRunningPrefix(auditinternal.StagePanic) + ` response="500"`,
+			},
+		},
+
+		// short running request with non-read-only verb
+		{
+			"writing empty",
+			shortRunningPath,
+			"PUT",
+			func(http.ResponseWriter, *http.Request) {},
+			[]string{
+				writingShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				writingShortRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
+			},
+		},
+		{
+			"writing sleep",
+			shortRunningPath,
+			"PUT",
 			func(http.ResponseWriter, *http.Request) {
 				time.Sleep(delay)
 			},
 			[]string{
-				shortRunningPrefix + ` response="200"`,
+				writingShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				writingShortRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
 			},
 		},
 		{
-			"403+write",
+			"writing 403+write",
 			shortRunningPath,
+			"PUT",
 			func(w http.ResponseWriter, req *http.Request) {
 				w.WriteHeader(403)
 				w.Write([]byte("foo"))
 			},
 			[]string{
-				shortRunningPrefix + ` response="403"`,
+				writingShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				writingShortRunningPrefix(auditinternal.StageResponseComplete) + ` response="403"`,
 			},
 		},
 		{
-			"panic",
+			"writing panic",
 			shortRunningPath,
+			"PUT",
 			func(w http.ResponseWriter, req *http.Request) {
 				panic("kaboom")
 			},
 			[]string{
-				shortRunningPrefix + ` response="500"`,
+				writingShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				writingShortRunningPrefix(auditinternal.StagePanic) + ` response="500"`,
 			},
 		},
 		{
-			"write+panic",
+			"writing write+panic",
 			shortRunningPath,
+			"PUT",
 			func(w http.ResponseWriter, req *http.Request) {
 				w.Write([]byte("foo"))
 				panic("kaboom")
 			},
 			[]string{
-				shortRunningPrefix + ` response="500"`,
+				writingShortRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				writingShortRunningPrefix(auditinternal.StagePanic) + ` response="500"`,
 			},
 		},
 
@@ -247,76 +290,92 @@ func TestAudit(t *testing.T) {
 		{
 			"empty longrunning",
 			longRunningPath,
+			"GET",
 			func(http.ResponseWriter, *http.Request) {},
 			[]string{
-				longRunningPrefix + ` response="200"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="200"`,
+				longRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
 			},
 		},
 		{
 			"sleep longrunning",
 			longRunningPath,
+			"GET",
 			func(http.ResponseWriter, *http.Request) {
 				time.Sleep(delay)
 			},
 			[]string{
-				longRunningPrefix + ` response="200"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="200"`,
+				longRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
 			},
 		},
 		{
 			"sleep+403 longrunning",
 			longRunningPath,
+			"GET",
 			func(w http.ResponseWriter, req *http.Request) {
 				time.Sleep(delay)
 				w.WriteHeader(403)
 			},
 			[]string{
-				longRunningPrefix + ` response="<deferred>"`,
-				longRunningPrefix + ` response="403"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="403"`,
+				longRunningPrefix(auditinternal.StageResponseComplete) + ` response="403"`,
 			},
 		},
 		{
 			"write longrunning",
 			longRunningPath,
+			"GET",
 			func(w http.ResponseWriter, req *http.Request) {
 				w.Write([]byte("foo"))
 			},
 			[]string{
-				longRunningPrefix + ` response="<deferred>"`,
-				longRunningPrefix + ` response="200"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="200"`,
+				longRunningPrefix(auditinternal.StageResponseComplete) + ` response="200"`,
 			},
 		},
 		{
 			"403+write longrunning",
 			longRunningPath,
+			"GET",
 			func(w http.ResponseWriter, req *http.Request) {
 				w.WriteHeader(403)
 				w.Write([]byte("foo"))
 			},
 			[]string{
-				longRunningPrefix + ` response="<deferred>"`,
-				longRunningPrefix + ` response="403"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="403"`,
+				longRunningPrefix(auditinternal.StageResponseComplete) + ` response="403"`,
 			},
 		},
 		{
 			"panic longrunning",
 			longRunningPath,
+			"GET",
 			func(w http.ResponseWriter, req *http.Request) {
 				panic("kaboom")
 			},
 			[]string{
-				longRunningPrefix + ` response="500"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StagePanic) + ` response="500"`,
 			},
 		},
 		{
 			"write+panic longrunning",
 			longRunningPath,
+			"GET",
 			func(w http.ResponseWriter, req *http.Request) {
 				w.Write([]byte("foo"))
 				panic("kaboom")
 			},
 			[]string{
-				longRunningPrefix + ` response="<deferred>"`,
-				longRunningPrefix + ` response="500"`,
+				longRunningPrefix(auditinternal.StageRequestReceived) + ` response="<deferred>"`,
+				longRunningPrefix(auditinternal.StageResponseStarted) + ` response="200"`,
+				longRunningPrefix(auditinternal.StagePanic) + ` response="500"`,
 			},
 		},
 	} {
@@ -330,7 +389,7 @@ func TestAudit(t *testing.T) {
 			return ri.Verb == "watch"
 		})
 
-		req, _ := http.NewRequest("GET", test.path, nil)
+		req, _ := http.NewRequest(test.verb, test.path, nil)
 		req.RemoteAddr = "127.0.0.1"
 
 		func() {

staging/src/k8s.io/apiserver/plugin/pkg/audit/log/backend.go

@@ -82,8 +82,8 @@ func (b *backend) logEvent(ev *auditinternal.Event) {
 		ip = ev.SourceIPs[0]
 	}
 
-	line := fmt.Sprintf("%s AUDIT: id=%q ip=%q method=%q user=%q groups=%q as=%q asgroups=%q namespace=%q uri=%q response=\"%s\"\n",
-		ev.Timestamp.Format(time.RFC3339Nano), ev.AuditID, ip, ev.Verb, username, groups, asuser, asgroups, namespace, ev.RequestURI, response)
+	line := fmt.Sprintf("%s AUDIT: id=%q stage=%q ip=%q method=%q user=%q groups=%q as=%q asgroups=%q namespace=%q uri=%q response=\"%s\"\n",
+		ev.Timestamp.Format(time.RFC3339Nano), ev.AuditID, ev.Stage, ip, ev.Verb, username, groups, asuser, asgroups, namespace, ev.RequestURI, response)
 	if _, err := fmt.Fprint(b.out, line); err != nil {
 		glog.Errorf("Unable to write audit log: %s, the error is: %v", line, err)
 	}