Merge pull request #16128 from soltysh/advanced_audit

Automatic merge from submit-queue (batch tested with PRs 16480, 16486, 16270, 16128, 16489)

Advanced audit as tech preview in origin

@sttts this enables the advance auditing features in origin, ptal

@openshift/api-review for config changes

Author: openshift-merge-robot

contrib/completions/bash/openshift

@@ -32067,6 +32067,8 @@ _openshift_start_kubernetes_apiserver()
     local_nonpersistent_flags+=("--anonymous-auth")
     flags+=("--apiserver-count=")
     local_nonpersistent_flags+=("--apiserver-count=")
+    flags+=("--audit-log-format=")
+    local_nonpersistent_flags+=("--audit-log-format=")
     flags+=("--audit-log-maxage=")
     local_nonpersistent_flags+=("--audit-log-maxage=")
     flags+=("--audit-log-maxbackup=")
@@ -33079,6 +33081,8 @@ _openshift_start_template-service-broker()
     flags_with_completion=()
     flags_completion=()
 
+    flags+=("--audit-log-format=")
+    local_nonpersistent_flags+=("--audit-log-format=")
     flags+=("--audit-log-maxage=")
     local_nonpersistent_flags+=("--audit-log-maxage=")
     flags+=("--audit-log-maxbackup=")

contrib/completions/zsh/openshift

@@ -32216,6 +32216,8 @@ _openshift_start_kubernetes_apiserver()
     local_nonpersistent_flags+=("--anonymous-auth")
     flags+=("--apiserver-count=")
     local_nonpersistent_flags+=("--apiserver-count=")
+    flags+=("--audit-log-format=")
+    local_nonpersistent_flags+=("--audit-log-format=")
     flags+=("--audit-log-maxage=")
     local_nonpersistent_flags+=("--audit-log-maxage=")
     flags+=("--audit-log-maxbackup=")
@@ -33228,6 +33230,8 @@ _openshift_start_template-service-broker()
     flags_with_completion=()
     flags_completion=()
 
+    flags+=("--audit-log-format=")
+    local_nonpersistent_flags+=("--audit-log-format=")
     flags+=("--audit-log-maxage=")
     local_nonpersistent_flags+=("--audit-log-maxage=")
     flags+=("--audit-log-maxbackup=")

pkg/cmd/server/api/helpers.go

@@ -271,6 +271,7 @@ func GetMasterFileReferences(config *MasterConfig) []*string {
 	}
 
 	refs = append(refs, &config.AuditConfig.AuditFilePath)
+	refs = append(refs, &config.AuditConfig.PolicyFile)
 
 	return refs
 }

pkg/cmd/server/api/install/install.go

@@ -3,10 +3,10 @@ package install
 import (
 	"fmt"
 
-	"github.com/golang/glog"
-
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apiserver/pkg/apis/audit"
+	auditv1alpha1 "k8s.io/apiserver/pkg/apis/audit/v1alpha1"
 
 	configapi "github.com/openshift/origin/pkg/cmd/server/api"
 	configapiv1 "github.com/openshift/origin/pkg/cmd/server/api/v1"
@@ -27,34 +27,12 @@ var accessor = meta.NewAccessor()
 var availableVersions = []schema.GroupVersion{configapiv1.SchemeGroupVersion}
 
 func init() {
-	if err := enableVersions(availableVersions); err != nil {
-		panic(err)
-	}
-}
-
-// TODO: enableVersions should be centralized rather than spread in each API
-// group.
-// We can combine registered.RegisterVersions, registered.EnableVersions and
-// registered.RegisterGroup once we have moved enableVersions there.
-func enableVersions(externalVersions []schema.GroupVersion) error {
-	addVersionsToScheme(externalVersions...)
-	return nil
-}
-
-func addVersionsToScheme(externalVersions ...schema.GroupVersion) {
-	// add the internal version to Scheme
 	configapi.AddToScheme(configapi.Scheme)
-	// add the enabled external versions to Scheme
-	for _, v := range externalVersions {
-		switch v {
-		case configapiv1.SchemeGroupVersion:
-			configapiv1.AddToScheme(configapi.Scheme)
-
-		default:
-			glog.Errorf("Version %s is not known, so it will not be added to the Scheme.", v)
-			continue
-		}
-	}
+	configapiv1.AddToScheme(configapi.Scheme)
+	// we additionally need to enable audit versions, since we embed the audit
+	// policy file inside master-config.yaml
+	audit.AddToScheme(configapi.Scheme)
+	auditv1alpha1.AddToScheme(configapi.Scheme)
 }
 
 func interfacesFor(version schema.GroupVersion) (*meta.VersionInterfaces, error) {

pkg/cmd/server/api/types.go

@@ -461,10 +461,29 @@ type AggregatorConfig struct {
 	ProxyClientInfo CertInfo
 }
 
+type LogFormatType string
+
+type WebHookModeType string
+
+const (
+	// LogFormatLegacy saves event in 1-line text format.
+	LogFormatLegacy LogFormatType = "legacy"
+	// LogFormatJson saves event in structured json format.
+	LogFormatJson LogFormatType = "json"
+
+	// WebHookModeBatch indicates that the webhook should buffer audit events
+	// internally, sending batch updates either once a certain number of
+	// events have been received or a certain amount of time has passed.
+	WebHookModeBatch WebHookModeType = "batch"
+	// WebHookModeBlocking causes the webhook to block on every attempt to process
+	// a set of events. This causes requests to the API server to wait for a
+	// round trip to the external audit service before sending a response.
+	WebHookModeBlocking WebHookModeType = "blocking"
+)
+
 // AuditConfig holds configuration for the audit capabilities
 type AuditConfig struct {
 	// If this flag is set, audit log will be printed in the logs.
-	// The logs contains, method, user and a requested URL.
 	Enabled bool
 	// All requests coming to the apiserver will be logged to this file.
 	AuditFilePath string
@@ -474,6 +493,21 @@ type AuditConfig struct {
 	MaximumRetainedFiles int
 	// Maximum size in megabytes of the log file before it gets rotated. Defaults to 100MB.
 	MaximumFileSizeMegabytes int
+
+	// PolicyFile is a path to the file that defines the audit policy configuration.
+	PolicyFile string
+	// PolicyConfiguration is an embedded policy configuration object to be used
+	// as the audit policy configuration. If present, it will be used instead of
+	// the path to the policy file.
+	PolicyConfiguration runtime.Object
+
+	// Format of saved audits (legacy or json).
+	LogFormat LogFormatType
+
+	// Path to a .kubeconfig formatted file that defines the audit webhook configuration.
+	WebHookKubeConfig string
+	// Strategy for sending audit events (block or batch).
+	WebHookMode WebHookModeType
 }
 
 // JenkinsPipelineConfig holds configuration for the Jenkins pipeline strategy

pkg/cmd/server/api/v1/conversions.go

@@ -405,6 +405,7 @@ func (c *MasterConfig) DecodeNestedObjects(d runtime.Decoder) error {
 			apihelpers.DecodeNestedRawExtensionOrUnknown(d, &c.OAuthConfig.IdentityProviders[i].Provider)
 		}
 	}
+	apihelpers.DecodeNestedRawExtensionOrUnknown(d, &c.AuditConfig.PolicyConfiguration)
 	return nil
 }
 
@@ -434,5 +435,8 @@ func (c *MasterConfig) EncodeNestedObjects(e runtime.Encoder) error {
 			}
 		}
 	}
+	if err := apihelpers.EncodeNestedRawExtension(e, &c.AuditConfig.PolicyConfiguration); err != nil {
+		return err
+	}
 	return nil
 }

pkg/cmd/server/api/v1/swagger_doc.go

@@ -90,6 +90,11 @@ var map_AuditConfig = map[string]string{
 	"maximumFileRetentionDays": "Maximum number of days to retain old log files based on the timestamp encoded in their filename.",
 	"maximumRetainedFiles":     "Maximum number of old log files to retain.",
 	"maximumFileSizeMegabytes": "Maximum size in megabytes of the log file before it gets rotated. Defaults to 100MB.",
+	"policyFile":               "PolicyFile is a path to the file that defines the audit policy configuration.",
+	"policyConfiguration":      "PolicyConfiguration is an embedded policy configuration object to be used as the audit policy configuration. If present, it will be used instead of the path to the policy file.",
+	"logFormat":                "Format of saved audits (legacy or json).",
+	"webHookKubeConfig":        "Path to a .kubeconfig formatted file that defines the audit webhook configuration.",
+	"webHookMode":              "Strategy for sending audit events (block or batch).",
 }
 
 func (AuditConfig) SwaggerDoc() map[string]string {

pkg/cmd/server/api/v1/types.go

@@ -318,6 +318,26 @@ type AggregatorConfig struct {
 	ProxyClientInfo CertInfo `json:"proxyClientInfo"`
 }
 
+type LogFormatType string
+
+type WebHookModeType string
+
+const (
+	// LogFormatLegacy saves event in 1-line text format.
+	LogFormatLegacy LogFormatType = "legacy"
+	// LogFormatJson saves event in structured json format.
+	LogFormatJson LogFormatType = "json"
+
+	// WebHookModeBatch indicates that the webhook should buffer audit events
+	// internally, sending batch updates either once a certain number of
+	// events have been received or a certain amount of time has passed.
+	WebHookModeBatch WebHookModeType = "batch"
+	// WebHookModeBlocking causes the webhook to block on every attempt to process
+	// a set of events. This causes requests to the API server to wait for a
+	// round trip to the external audit service before sending a response.
+	WebHookModeBlocking WebHookModeType = "blocking"
+)
+
 // AuditConfig holds configuration for the audit capabilities
 type AuditConfig struct {
 	// If this flag is set, audit log will be printed in the logs.
@@ -331,6 +351,21 @@ type AuditConfig struct {
 	MaximumRetainedFiles int `json:"maximumRetainedFiles"`
 	// Maximum size in megabytes of the log file before it gets rotated. Defaults to 100MB.
 	MaximumFileSizeMegabytes int `json:"maximumFileSizeMegabytes"`
+
+	// PolicyFile is a path to the file that defines the audit policy configuration.
+	PolicyFile string `json:"policyFile"`
+	// PolicyConfiguration is an embedded policy configuration object to be used
+	// as the audit policy configuration. If present, it will be used instead of
+	// the path to the policy file.
+	PolicyConfiguration runtime.RawExtension `json:"policyConfiguration"`
+
+	// Format of saved audits (legacy or json).
+	LogFormat LogFormatType `json:"logFormat"`
+
+	// Path to a .kubeconfig formatted file that defines the audit webhook configuration.
+	WebHookKubeConfig string `json:"webHookKubeConfig"`
+	// Strategy for sending audit events (block or batch).
+	WebHookMode WebHookModeType `json:"webHookMode"`
 }
 
 // JenkinsPipelineConfig holds configuration for the Jenkins pipeline strategy

pkg/cmd/server/api/v1/types_test.go

@@ -117,9 +117,14 @@ assetConfig:
 auditConfig:
   auditFilePath: ""
   enabled: false
+  logFormat: ""
   maximumFileRetentionDays: 0
   maximumFileSizeMegabytes: 0
   maximumRetainedFiles: 0
+  policyConfiguration: null
+  policyFile: ""
+  webHookKubeConfig: ""
+  webHookMode: ""
 authConfig:
   requestHeader: null
 controllerConfig:

pkg/cmd/server/api/validation/master.go

@@ -14,6 +14,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	kuval "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+	auditinternal "k8s.io/apiserver/pkg/apis/audit"
+	auditvalidation "k8s.io/apiserver/pkg/apis/audit/validation"
+	auditpolicy "k8s.io/apiserver/pkg/audit/policy"
+	"k8s.io/client-go/tools/clientcmd"
 	apiserveroptions "k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	kcmoptions "k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
 	kvalidation "k8s.io/kubernetes/pkg/api/validation"
@@ -238,6 +242,9 @@ func ValidateAggregatorConfig(config api.AggregatorConfig, fldPath *field.Path)
 
 func ValidateAuditConfig(config api.AuditConfig, fldPath *field.Path) ValidationResults {
 	validationResults := ValidationResults{}
+	if !config.Enabled {
+		return validationResults
+	}
 
 	if len(config.AuditFilePath) == 0 {
 		// for backwards compatibility reasons we can't error this out
@@ -253,6 +260,61 @@ func ValidateAuditConfig(config api.AuditConfig, fldPath *field.Path) Validation
 		validationResults.AddErrors(field.Invalid(fldPath.Child("maximumFileSizeMegabytes"), config.MaximumFileSizeMegabytes, "must be greater than or equal to 0"))
 	}
 
+	// setting policy file will turn the advanced auditing on
+	if config.PolicyConfiguration != nil && len(config.PolicyFile) > 0 {
+		validationResults.AddErrors(field.Forbidden(fldPath.Child("policyFile"), "both policyFile and policyConfiguration cannot be specified"))
+	}
+	if config.PolicyConfiguration != nil || len(config.PolicyFile) > 0 {
+		if config.PolicyConfiguration == nil {
+			policy, err := auditpolicy.LoadPolicyFromFile(config.PolicyFile)
+			if err != nil {
+				validationResults.AddErrors(field.Invalid(fldPath.Child("policyFile"), config.PolicyFile, err.Error()))
+			}
+			if policy == nil || len(policy.Rules) == 0 {
+				validationResults.AddErrors(field.Invalid(fldPath.Child("policyFile"), config.PolicyFile, "a policy file with 0 policies is not valid"))
+			}
+		} else {
+			policyConfiguration, ok := config.PolicyConfiguration.(*auditinternal.Policy)
+			if !ok {
+				validationResults.AddErrors(field.Invalid(fldPath.Child("policyConfiguration"), config.PolicyConfiguration, "must be of type audit/v1alpha1.Policy"))
+			} else {
+				if err := auditvalidation.ValidatePolicy(policyConfiguration); err != nil {
+					validationResults.AddErrors(field.Invalid(fldPath.Child("policyConfiguration"), config.PolicyConfiguration, err.ToAggregate().Error()))
+				}
+				if len(policyConfiguration.Rules) == 0 {
+					validationResults.AddErrors(field.Invalid(fldPath.Child("policyConfiguration"), config.PolicyFile, "a policy configuration with 0 policies is not valid"))
+				}
+			}
+		}
+
+		if len(config.AuditFilePath) == 0 {
+			validationResults.AddErrors(field.Required(fldPath.Child("auditFilePath"), "advanced audit requires a separate log file"))
+		}
+		switch config.LogFormat {
+		case api.LogFormatLegacy, api.LogFormatJson:
+			// ok
+		default:
+			validationResults.AddErrors(field.NotSupported(fldPath.Child("logFormat"), config.LogFormat, []string{string(api.LogFormatLegacy), string(api.LogFormatJson)}))
+		}
+
+		if len(config.WebHookKubeConfig) > 0 {
+			switch config.WebHookMode {
+			case api.WebHookModeBatch, api.WebHookModeBlocking:
+				// ok
+			default:
+				validationResults.AddErrors(field.NotSupported(fldPath.Child("webHookMode"), config.WebHookMode, []string{string(api.WebHookModeBatch), string(api.WebHookModeBlocking)}))
+			}
+			loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
+			loadingRules.ExplicitPath = config.WebHookKubeConfig
+			loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, &clientcmd.ConfigOverrides{})
+			if _, err := loader.ClientConfig(); err != nil {
+				validationResults.AddErrors(field.Invalid(fldPath.Child("webHookKubeConfig"), config.WebHookKubeConfig, err.Error()))
+			}
+		} else if len(config.WebHookMode) > 0 {
+			validationResults.AddErrors(field.Required(fldPath.Child("webHookKubeConfig"), "must be specified when webHookMode is set"))
+		}
+	}
+
 	return validationResults
 }
 

pkg/cmd/server/kubernetes/master/master_config.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/admission"
 	auditinternal "k8s.io/apiserver/pkg/apis/audit"
+	"k8s.io/apiserver/pkg/audit"
 	auditpolicy "k8s.io/apiserver/pkg/audit/policy"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
@@ -45,6 +46,7 @@ import (
 	storagefactory "k8s.io/apiserver/pkg/storage/storagebackend/factory"
 	utilflag "k8s.io/apiserver/pkg/util/flag"
 	auditlog "k8s.io/apiserver/plugin/pkg/audit/log"
+	auditwebhook "k8s.io/apiserver/plugin/pkg/audit/webhook"
 	kapiserveroptions "k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	cmapp "k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
 	kapi "k8s.io/kubernetes/pkg/api"
@@ -177,12 +179,11 @@ func BuildKubeAPIserverOptions(masterConfig configapi.MasterConfig) (*kapiserver
 			args["feature-gates"] = []string{existing[0] + ",AdvancedAuditing=true"}
 		} else {
 			args["feature-gates"] = []string{"AdvancedAuditing=true"}
-
 		}
 	}
 	// TODO: this should be done in config validation (along with the above) so we can provide
 	// proper errors
-	if err := cmdflags.Resolve(masterConfig.KubernetesMasterConfig.APIServerArguments, server.AddFlags); len(err) > 0 {
+	if err := cmdflags.Resolve(args, server.AddFlags); len(err) > 0 {
 		return nil, kerrors.NewAggregate(err)
 	}
 
@@ -533,12 +534,38 @@ func buildKubeApiserverConfig(
 			// backwards compatible writer to regular log
 			writer = cmdutil.NewGLogWriterV(0)
 		}
-		genericConfig.AuditBackend = auditlog.NewBackend(writer)
+		genericConfig.AuditBackend = auditlog.NewBackend(writer, auditlog.FormatLegacy)
 		genericConfig.AuditPolicyChecker = auditpolicy.NewChecker(&auditinternal.Policy{
 			// This is for backwards compatibility maintaining the old visibility, ie. just
 			// raw overview of the requests comming in.
 			Rules: []auditinternal.PolicyRule{{Level: auditinternal.LevelMetadata}},
 		})
+
+		// when a policy file is defined we enable the advanced auditing
+		if masterConfig.AuditConfig.PolicyConfiguration != nil || len(masterConfig.AuditConfig.PolicyFile) > 0 {
+			// policy configuration
+			if masterConfig.AuditConfig.PolicyConfiguration == nil {
+				p, _ := auditpolicy.LoadPolicyFromFile(masterConfig.AuditConfig.PolicyFile)
+				genericConfig.AuditPolicyChecker = auditpolicy.NewChecker(p)
+			} else if len(masterConfig.AuditConfig.PolicyFile) > 0 {
+				p := masterConfig.AuditConfig.PolicyConfiguration.(*auditinternal.Policy)
+				genericConfig.AuditPolicyChecker = auditpolicy.NewChecker(p)
+			}
+
+			// log configuration, only when file path was provided
+			if len(masterConfig.AuditConfig.AuditFilePath) > 0 {
+				genericConfig.AuditBackend = auditlog.NewBackend(writer, string(masterConfig.AuditConfig.LogFormat))
+			}
+
+			// webhook configuration, only when config file was provided
+			if len(masterConfig.AuditConfig.WebHookKubeConfig) > 0 {
+				webhook, err := auditwebhook.NewBackend(masterConfig.AuditConfig.WebHookKubeConfig, string(masterConfig.AuditConfig.WebHookMode))
+				if err != nil {
+					glog.Fatalf("Audit webhook initialization failed: %v", err)
+				}
+				genericConfig.AuditBackend = audit.Union(genericConfig.AuditBackend, webhook)
+			}
+		}
 	}
 
 	kubeApiserverConfig := &master.Config{