add an integration test for advanced audit feature

test/e2e/auth/BUILD

@@ -41,6 +41,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/apis/audit:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/audit/v1beta1:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/authentication/serviceaccount:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
@@ -49,6 +50,7 @@ go_library(
         "//staging/src/k8s.io/client-go/util/cert:go_default_library",
         "//test/e2e/common:go_default_library",
         "//test/e2e/framework:go_default_library",
+        "//test/utils:go_default_library",
         "//test/utils/image:go_default_library",
         "//vendor/github.com/evanphx/json-patch:go_default_library",
         "//vendor/github.com/onsi/ginkgo:go_default_library",

test/integration/master/BUILD

@@ -10,6 +10,7 @@ go_test(
     name = "go_default_test",
     size = "large",
     srcs = [
+        "audit_test.go",
         "crd_test.go",
         "kms_transformation_test.go",
         "kube_apiserver_test.go",
@@ -37,7 +38,11 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/apis/audit:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/apis/audit/v1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/apis/audit/v1beta1:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/authentication/group:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/authentication/request/bearertoken:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/authentication/user:go_default_library",
@@ -58,6 +63,8 @@ go_test(
         "//staging/src/k8s.io/kube-aggregator/pkg/apis/apiregistration:go_default_library",
         "//test/integration:go_default_library",
         "//test/integration/framework:go_default_library",
+        "//test/utils:go_default_library",
+        "//vendor/github.com/evanphx/json-patch:go_default_library",
         "//vendor/github.com/ghodss/yaml:go_default_library",
     ] + select({
         "@io_bazel_rules_go//go/platform:android": [

test/integration/master/audit_test.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package master
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"strings"
+	"testing"
+
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	auditinternal "k8s.io/apiserver/pkg/apis/audit"
+	auditv1 "k8s.io/apiserver/pkg/apis/audit/v1"
+	auditv1beta1 "k8s.io/apiserver/pkg/apis/audit/v1beta1"
+	"k8s.io/client-go/kubernetes"
+	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
+	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils"
+
+	"github.com/evanphx/json-patch"
+)
+
+var (
+	auditPolicyPattern = `
+apiVersion: {version}
+kind: Policy
+rules:
+  - level: RequestResponse
+    resources:
+      - group: "" # core
+        resources: ["configmaps"]
+
+`
+	namespace              = "default"
+	watchTestTimeout int64 = 1
+	watchOptions           = metav1.ListOptions{TimeoutSeconds: &watchTestTimeout}
+	patch, _               = json.Marshal(jsonpatch.Patch{})
+	auditTestUser          = "system:apiserver"
+	versions               = map[string]schema.GroupVersion{
+		"audit.k8s.io/v1":      auditv1.SchemeGroupVersion,
+		"audit.k8s.io/v1beta1": auditv1beta1.SchemeGroupVersion,
+	}
+)
+
+// TestAudit ensures that both v1beta1 and v1 version audit api could work.
+func TestAudit(t *testing.T) {
+	for version := range versions {
+		testAudit(t, version)
+	}
+}
+
+func testAudit(t *testing.T, version string) {
+	// prepare audit policy file
+	auditPolicy := []byte(strings.Replace(auditPolicyPattern, "{version}", version, 1))
+	policyFile, err := ioutil.TempFile("", "audit-policy.yaml")
+	if err != nil {
+		t.Fatalf("Failed to create audit policy file: %v", err)
+	}
+	defer os.Remove(policyFile.Name())
+	if _, err := policyFile.Write(auditPolicy); err != nil {
+		t.Fatalf("Failed to write audit policy file: %v", err)
+	}
+	if err := policyFile.Close(); err != nil {
+		t.Fatalf("Failed to close audit policy file: %v", err)
+	}
+
+	// prepare audit log file
+	logFile, err := ioutil.TempFile("", "audit.log")
+	if err != nil {
+		t.Fatalf("Failed to create audit log file: %v", err)
+	}
+	defer os.Remove(logFile.Name())
+
+	// start api server
+	result := kubeapiservertesting.StartTestServerOrDie(t, nil,
+		[]string{
+			"--audit-policy-file", policyFile.Name(),
+			"--audit-log-version", version,
+			"--audit-log-mode", "blocking",
+			"--audit-log-path", logFile.Name()},
+		framework.SharedEtcd())
+	defer result.TearDownFn()
+
+	kubeclient, err := kubernetes.NewForConfig(result.ClientConfig)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	func() {
+		// create, get, watch, update, patch, list and delete configmap.
+		configMap := &apiv1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "audit-configmap",
+			},
+			Data: map[string]string{
+				"map-key": "map-value",
+			},
+		}
+
+		_, err := kubeclient.CoreV1().ConfigMaps(namespace).Create(configMap)
+		expectNoError(t, err, "failed to create audit-configmap")
+
+		_, err = kubeclient.CoreV1().ConfigMaps(namespace).Get(configMap.Name, metav1.GetOptions{})
+		expectNoError(t, err, "failed to get audit-configmap")
+
+		configMapChan, err := kubeclient.CoreV1().ConfigMaps(namespace).Watch(watchOptions)
+		expectNoError(t, err, "failed to create watch for config maps")
+		for range configMapChan.ResultChan() {
+			// Block until watchOptions.TimeoutSeconds expires.
+			// If the test finishes before watchOptions.TimeoutSeconds expires, the watch audit
+			// event at stage ResponseComplete will not be generated.
+		}
+
+		_, err = kubeclient.CoreV1().ConfigMaps(namespace).Update(configMap)
+		expectNoError(t, err, "failed to update audit-configmap")
+
+		_, err = kubeclient.CoreV1().ConfigMaps(namespace).Patch(configMap.Name, types.JSONPatchType, patch)
+		expectNoError(t, err, "failed to patch configmap")
+
+		_, err = kubeclient.CoreV1().ConfigMaps(namespace).List(metav1.ListOptions{})
+		expectNoError(t, err, "failed to list config maps")
+
+		err = kubeclient.CoreV1().ConfigMaps(namespace).Delete(configMap.Name, &metav1.DeleteOptions{})
+		expectNoError(t, err, "failed to delete audit-configmap")
+	}()
+
+	expectedEvents := []utils.AuditEvent{
+		{
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps", namespace),
+			Verb:              "create",
+			Code:              201,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     true,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps/audit-configmap", namespace),
+			Verb:              "get",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     false,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps", namespace),
+			Verb:              "list",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     false,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseStarted,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps?timeoutSeconds=%d&watch=true", namespace, watchTestTimeout),
+			Verb:              "watch",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     false,
+			ResponseObject:    false,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps?timeoutSeconds=%d&watch=true", namespace, watchTestTimeout),
+			Verb:              "watch",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     false,
+			ResponseObject:    false,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps/audit-configmap", namespace),
+			Verb:              "update",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     true,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps/audit-configmap", namespace),
+			Verb:              "patch",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     true,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		}, {
+			Level:             auditinternal.LevelRequestResponse,
+			Stage:             auditinternal.StageResponseComplete,
+			RequestURI:        fmt.Sprintf("/api/v1/namespaces/%s/configmaps/audit-configmap", namespace),
+			Verb:              "delete",
+			Code:              200,
+			User:              auditTestUser,
+			Resource:          "configmaps",
+			Namespace:         namespace,
+			RequestObject:     true,
+			ResponseObject:    true,
+			AuthorizeDecision: "allow",
+		},
+	}
+
+	stream, err := os.Open(logFile.Name())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer stream.Close()
+	missing, err := utils.CheckAuditLines(stream, expectedEvents, versions[version])
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if len(missing) > 0 {
+		t.Errorf("Failed to match all expected events, events %#v not found!", missing)
+	}
+}
+
+func expectNoError(t *testing.T, err error, msg string) {
+	if err != nil {
+		t.Fatalf("%s: %v", msg, err)
+	}
+}

test/utils/BUILD

@@ -8,6 +8,7 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
+        "audit.go",
         "conditions.go",
         "create_resources.go",
         "delete_resources.go",
@@ -49,6 +50,8 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/apis/audit:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/audit:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/scale:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",

test/utils/audit.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	auditinternal "k8s.io/apiserver/pkg/apis/audit"
+	"k8s.io/apiserver/pkg/audit"
+)
+
+type AuditEvent struct {
+	Level             auditinternal.Level
+	Stage             auditinternal.Stage
+	RequestURI        string
+	Verb              string
+	Code              int32
+	User              string
+	Resource          string
+	Namespace         string
+	RequestObject     bool
+	ResponseObject    bool
+	AuthorizeDecision string
+}
+
+// Search the audit log for the expected audit lines.
+func CheckAuditLines(stream io.Reader, expected []AuditEvent, version schema.GroupVersion) (missing []AuditEvent, err error) {
+	expectations := map[AuditEvent]bool{}
+	for _, event := range expected {
+		expectations[event] = false
+	}
+
+	scanner := bufio.NewScanner(stream)
+	for scanner.Scan() {
+		line := scanner.Text()
+		event, err := parseAuditLine(line, version)
+		if err != nil {
+			return expected, err
+		}
+
+		// If the event was expected, mark it as found.
+		if _, found := expectations[event]; found {
+			expectations[event] = true
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return expected, err
+	}
+
+	missing = make([]AuditEvent, 0)
+	for event, found := range expectations {
+		if !found {
+			missing = append(missing, event)
+		}
+	}
+	return missing, nil
+}
+
+func parseAuditLine(line string, version schema.GroupVersion) (AuditEvent, error) {
+	e := &auditinternal.Event{}
+	decoder := audit.Codecs.UniversalDecoder(version)
+	if err := runtime.DecodeInto(decoder, []byte(line), e); err != nil {
+		return AuditEvent{}, fmt.Errorf("failed decoding buf: %s, apiVersion: %s", line, version)
+	}
+
+	event := AuditEvent{
+		Level:      e.Level,
+		Stage:      e.Stage,
+		RequestURI: e.RequestURI,
+		Verb:       e.Verb,
+		User:       e.User.Username,
+	}
+	if e.ObjectRef != nil {
+		event.Namespace = e.ObjectRef.Namespace
+		event.Resource = e.ObjectRef.Resource
+	}
+	if e.ResponseStatus != nil {
+		event.Code = e.ResponseStatus.Code
+	}
+	if e.ResponseObject != nil {
+		event.ResponseObject = true
+	}
+	if e.RequestObject != nil {
+		event.RequestObject = true
+	}
+	event.AuthorizeDecision = e.Annotations["authorization.k8s.io/decision"]
+	return event, nil
+}