Provide a way to omit Event stages in audit policy

Updates kubernetes#48561
This provide a way to omit some stages for each audit policy rule.

For example:
  - level: Metadata
    resources:
       - group: "rbac.authorization.k8s.io"
         resources: ["roles"]
    omitStages:
      - "RequestReceived"

RequestReceived stage will not be emitted to audit backends with
previous config.

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

@@ -200,6 +200,11 @@ type PolicyRule struct {
 	//  "/healthz*" - Log all health checks
 	// +optional
 	NonResourceURLs []string
+
+	// OmitStages specify events generated in which stages will not be emitted to backend
+	// Allowed stages are RequestReceived,ResponseStarted,ResponseComplete,Panic
+	// empty list means every events will be emitted.
+	OmitStages []Stage
 }
 
 // GroupResources represents resource kinds in an API group.

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

@@ -201,6 +201,11 @@ type PolicyRule struct {
 	//  "/healthz*" - Log all health checks
 	// +optional
 	NonResourceURLs []string `json:"nonResourceURLs,omitempty" protobuf:"bytes,7,rep,name=nonResourceURLs"`
+
+	// OmitStages specify events generated in which stages will not be emitted to backend
+	// Allowed stages are RequestReceived,ResponseStarted,ResponseComplete,Panic
+	// empty list means every events will be emitted.
+	OmitStages []Stage `json:"omitStages,omitempty" protobuf:"bytes,8,rep,name=omitStages"`
 }
 
 // GroupResources represents resource kinds in an API group.

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

@@ -247,6 +247,7 @@ func autoConvert_v1alpha1_PolicyRule_To_audit_PolicyRule(in *PolicyRule, out *au
 	out.Resources = *(*[]audit.GroupResources)(unsafe.Pointer(&in.Resources))
 	out.Namespaces = *(*[]string)(unsafe.Pointer(&in.Namespaces))
 	out.NonResourceURLs = *(*[]string)(unsafe.Pointer(&in.NonResourceURLs))
+	out.OmitStages = *(*[]audit.Stage)(unsafe.Pointer(&in.OmitStages))
 	return nil
 }
 
@@ -263,6 +264,7 @@ func autoConvert_audit_PolicyRule_To_v1alpha1_PolicyRule(in *audit.PolicyRule, o
 	out.Resources = *(*[]GroupResources)(unsafe.Pointer(&in.Resources))
 	out.Namespaces = *(*[]string)(unsafe.Pointer(&in.Namespaces))
 	out.NonResourceURLs = *(*[]string)(unsafe.Pointer(&in.NonResourceURLs))
+	out.OmitStages = *(*[]Stage)(unsafe.Pointer(&in.OmitStages))
 	return nil
 }
 

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

@@ -321,6 +321,11 @@ func (in *PolicyRule) DeepCopyInto(out *PolicyRule) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.OmitStages != nil {
+		in, out := &in.OmitStages, &out.OmitStages
+		*out = make([]Stage, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 

staging/src/k8s.io/apiserver/pkg/apis/audit/zz_generated.deepcopy.go

@@ -318,6 +318,11 @@ func (in *PolicyRule) DeepCopyInto(out *PolicyRule) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.OmitStages != nil {
+		in, out := &in.OmitStages, &out.OmitStages
+		*out = make([]Stage, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 

staging/src/k8s.io/apiserver/pkg/audit/policy/checker.go

@@ -31,7 +31,7 @@ const (
 // Checker exposes methods for checking the policy rules.
 type Checker interface {
 	// Check the audit level for a request with the given authorizer attributes.
-	Level(authorizer.Attributes) audit.Level
+	Level(authorizer.Attributes) (audit.Level, []audit.Stage)
 }
 
 // NewChecker creates a new policy checker.
@@ -40,21 +40,21 @@ func NewChecker(policy *audit.Policy) Checker {
 }
 
 // FakeChecker creates a checker that returns a constant level for all requests (for testing).
-func FakeChecker(level audit.Level) Checker {
-	return &fakeChecker{level}
+func FakeChecker(level audit.Level, stage []audit.Stage) Checker {
+	return &fakeChecker{level, stage}
 }
 
 type policyChecker struct {
 	audit.Policy
 }
 
-func (p *policyChecker) Level(attrs authorizer.Attributes) audit.Level {
+func (p *policyChecker) Level(attrs authorizer.Attributes) (audit.Level, []audit.Stage) {
 	for _, rule := range p.Rules {
 		if ruleMatches(&rule, attrs) {
-			return rule.Level
+			return rule.Level, rule.OmitStages
 		}
 	}
-	return DefaultAuditLevel
+	return DefaultAuditLevel, nil
 }
 
 // Check whether the rule matches the request attrs.
@@ -170,8 +170,9 @@ func hasString(slice []string, value string) bool {
 
 type fakeChecker struct {
 	level audit.Level
+	stage []audit.Stage
 }
 
-func (f *fakeChecker) Level(_ authorizer.Attributes) audit.Level {
-	return f.level
+func (f *fakeChecker) Level(_ authorizer.Attributes) (audit.Level, []audit.Stage) {
+	return f.level, f.stage
 }

staging/src/k8s.io/apiserver/pkg/audit/policy/checker_test.go

@@ -120,7 +120,7 @@ func TestChecker(t *testing.T) {
 			policy.Rules = append(policy.Rules, rules[rule])
 		}
 		require.Contains(t, attrs, req)
-		actual := NewChecker(&policy).Level(attrs[req])
+		actual, _ := NewChecker(&policy).Level(attrs[req])
 		assert.Equal(t, expected, actual, "request:%s rules:%s", req, strings.Join(ruleNames, ","))
 	}
 

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

@@ -331,7 +331,7 @@ func handleInternal(storage map[string]rest.Storage, admissionControl admission.
 		}
 	}
 
-	handler := genericapifilters.WithAudit(mux, requestContextMapper, auditSink, auditpolicy.FakeChecker(auditinternal.LevelRequestResponse), func(r *http.Request, requestInfo *request.RequestInfo) bool {
+	handler := genericapifilters.WithAudit(mux, requestContextMapper, auditSink, auditpolicy.FakeChecker(auditinternal.LevelRequestResponse, nil), func(r *http.Request, requestInfo *request.RequestInfo) bool {
 		// simplified long-running check
 		return requestInfo.Verb == "watch" || requestInfo.Verb == "proxy"
 	})

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

@@ -55,7 +55,7 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 			return
 		}
 
-		level := policy.Level(attribs)
+		level, omitStages := policy.Level(attribs)
 		audit.ObservePolicyLevel(level)
 		if level == auditinternal.LevelNone {
 			// Don't audit.
@@ -78,7 +78,7 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 		}
 
 		ev.Stage = auditinternal.StageRequestReceived
-		processEvent(sink, ev)
+		processEvent(sink, ev, omitStages)
 
 		// intercept the status code
 		var longRunningSink audit.Sink
@@ -88,7 +88,7 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 				longRunningSink = sink
 			}
 		}
-		respWriter := decorateResponseWriter(w, ev, longRunningSink)
+		respWriter := decorateResponseWriter(w, ev, longRunningSink, omitStages)
 
 		// send audit event when we leave this func, either via a panic or cleanly. In the case of long
 		// running requests, this will be the second audit event.
@@ -102,7 +102,7 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 					Reason:  metav1.StatusReasonInternalError,
 					Message: fmt.Sprintf("APIServer panic'd: %v", r),
 				}
-				processEvent(sink, ev)
+				processEvent(sink, ev, omitStages)
 				return
 			}
 
@@ -115,29 +115,35 @@ func WithAudit(handler http.Handler, requestContextMapper request.RequestContext
 			if ev.ResponseStatus == nil && longRunningSink != nil {
 				ev.ResponseStatus = fakedSuccessStatus
 				ev.Stage = auditinternal.StageResponseStarted
-				processEvent(longRunningSink, ev)
+				processEvent(longRunningSink, ev, omitStages)
 			}
 
 			ev.Stage = auditinternal.StageResponseComplete
 			if ev.ResponseStatus == nil {
 				ev.ResponseStatus = fakedSuccessStatus
 			}
-			processEvent(sink, ev)
+			processEvent(sink, ev, omitStages)
 		}()
 		handler.ServeHTTP(respWriter, req)
 	})
 }
 
-func processEvent(sink audit.Sink, ev *auditinternal.Event) {
+func processEvent(sink audit.Sink, ev *auditinternal.Event, omitStages []auditinternal.Stage) {
+	for _, stage := range omitStages {
+		if ev.Stage == stage {
+			return
+		}
+	}
 	audit.ObserveEvent()
 	sink.ProcessEvents(ev)
 }
 
-func decorateResponseWriter(responseWriter http.ResponseWriter, ev *auditinternal.Event, sink audit.Sink) http.ResponseWriter {
+func decorateResponseWriter(responseWriter http.ResponseWriter, ev *auditinternal.Event, sink audit.Sink, omitStages []auditinternal.Stage) http.ResponseWriter {
 	delegate := &auditResponseWriter{
 		ResponseWriter: responseWriter,
 		event:          ev,
 		sink:           sink,
+		omitStages:     omitStages,
 	}
 
 	// check if the ResponseWriter we're wrapping is the fancy one we need
@@ -157,9 +163,10 @@ var _ http.ResponseWriter = &auditResponseWriter{}
 // create immediately an event (for long running requests).
 type auditResponseWriter struct {
 	http.ResponseWriter
-	event *auditinternal.Event
-	once  sync.Once
-	sink  audit.Sink
+	event      *auditinternal.Event
+	once       sync.Once
+	sink       audit.Sink
+	omitStages []auditinternal.Stage
 }
 
 func (a *auditResponseWriter) processCode(code int) {
@@ -171,7 +178,7 @@ func (a *auditResponseWriter) processCode(code int) {
 		a.event.Stage = auditinternal.StageResponseStarted
 
 		if a.sink != nil {
-			processEvent(a.sink, a.event)
+			processEvent(a.sink, a.event, a.omitStages)
 		}
 	})
 }

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

@@ -98,14 +98,14 @@ func (*fancyResponseWriter) Flush() {}
 func (*fancyResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { return nil, nil, nil }
 
 func TestConstructResponseWriter(t *testing.T) {
-	actual := decorateResponseWriter(&simpleResponseWriter{}, nil, nil)
+	actual := decorateResponseWriter(&simpleResponseWriter{}, nil, nil, nil)
 	switch v := actual.(type) {
 	case *auditResponseWriter:
 	default:
 		t.Errorf("Expected auditResponseWriter, got %v", reflect.TypeOf(v))
 	}
 
-	actual = decorateResponseWriter(&fancyResponseWriter{}, nil, nil)
+	actual = decorateResponseWriter(&fancyResponseWriter{}, nil, nil, nil)
 	switch v := actual.(type) {
 	case *fancyResponseWriterDelegator:
 	default:
@@ -115,7 +115,7 @@ func TestConstructResponseWriter(t *testing.T) {
 
 func TestDecorateResponseWriterWithoutChannel(t *testing.T) {
 	ev := &auditinternal.Event{}
-	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, nil)
+	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, nil, nil)
 
 	// write status. This will not block because firstEventSentCh is nil
 	actual.WriteHeader(42)
@@ -129,7 +129,7 @@ func TestDecorateResponseWriterWithoutChannel(t *testing.T) {
 
 func TestDecorateResponseWriterWithImplicitWrite(t *testing.T) {
 	ev := &auditinternal.Event{}
-	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, nil)
+	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, nil, nil)
 
 	// write status. This will not block because firstEventSentCh is nil
 	actual.Write([]byte("foo"))
@@ -144,7 +144,7 @@ func TestDecorateResponseWriterWithImplicitWrite(t *testing.T) {
 func TestDecorateResponseWriterChannel(t *testing.T) {
 	sink := &fakeAuditSink{}
 	ev := &auditinternal.Event{}
-	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, sink)
+	actual := decorateResponseWriter(&simpleResponseWriter{}, ev, sink, nil)
 
 	done := make(chan struct{})
 	go func() {
@@ -390,7 +390,7 @@ func TestAuditLegacy(t *testing.T) {
 	} {
 		var buf bytes.Buffer
 		backend := pluginlog.NewBackend(&buf, pluginlog.FormatLegacy)
-		policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse)
+		policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse, nil)
 		handler := WithAudit(http.HandlerFunc(test.handler), &fakeRequestContextMapper{
 			user: &user.DefaultInfo{Name: "admin"},
 		}, backend, policyChecker, func(r *http.Request, ri *request.RequestInfo) bool {
@@ -838,7 +838,7 @@ func TestAuditJson(t *testing.T) {
 	} {
 		var buf bytes.Buffer
 		backend := pluginlog.NewBackend(&buf, pluginlog.FormatJson)
-		policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse)
+		policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse, nil)
 		handler := WithAudit(http.HandlerFunc(test.handler), &fakeRequestContextMapper{
 			user: &user.DefaultInfo{Name: "admin"},
 		}, backend, policyChecker, func(r *http.Request, ri *request.RequestInfo) bool {
@@ -930,7 +930,7 @@ func (*fakeRequestContextMapper) Update(req *http.Request, context request.Conte
 }
 
 func TestAuditNoPanicOnNilUser(t *testing.T) {
-	policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse)
+	policyChecker := policy.FakeChecker(auditinternal.LevelRequestResponse, nil)
 	handler := WithAudit(&fakeHTTPHandler{}, &fakeRequestContextMapper{}, &fakeAuditSink{}, policyChecker, nil)
 	req, _ := http.NewRequest("GET", "/api/v1/namespaces/default/pods", nil)
 	req.RemoteAddr = "127.0.0.1"
@@ -943,7 +943,7 @@ func TestAuditLevelNone(t *testing.T) {
 	handler = http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(200)
 	})
-	policyChecker := policy.FakeChecker(auditinternal.LevelNone)
+	policyChecker := policy.FakeChecker(auditinternal.LevelNone, nil)
 	handler = WithAudit(handler, &fakeRequestContextMapper{
 		user: &user.DefaultInfo{Name: "admin"},
 	}, sink, policyChecker, nil)