Merge pull request #2672 from wanjunlei/auditing-log

make ws admin can be aware of anything happened in its workspace.

pkg/apiserver/apiserver.go

@@ -260,8 +260,7 @@ func (s *APIServer) buildHandlerChain(stopCh <-chan struct{}) {
 
 	if s.Config.AuditingOptions.Enable {
 		handler = filters.WithAuditing(handler,
-			audit.NewAuditing(s.InformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
+			audit.NewAuditing(s.InformerFactory, s.Config.AuditingOptions.WebhookUrl, stopCh))
-				s.Config.AuditingOptions.WebhookUrl, stopCh))
 	}
 
 	var authorizers authorizer.Authorizer

pkg/apiserver/auditing/types.go

@@ -12,9 +12,14 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/apis/audit"
 	"k8s.io/klog"
+	devopsv1alpha3 "kubesphere.io/kubesphere/pkg/apis/devops/v1alpha3"
 	auditv1alpha1 "kubesphere.io/kubesphere/pkg/apiserver/auditing/v1alpha1"
+	"kubesphere.io/kubesphere/pkg/apiserver/query"
 	"kubesphere.io/kubesphere/pkg/apiserver/request"
 	"kubesphere.io/kubesphere/pkg/client/listers/auditing/v1alpha1"
+	"kubesphere.io/kubesphere/pkg/informers"
+	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3"
+	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3/devops"
 	"kubesphere.io/kubesphere/pkg/utils/iputil"
 	"net"
 	"net/http"
@@ -37,15 +42,17 @@ type Auditing interface {
 }
 
 type auditing struct {
-	lister  v1alpha1.WebhookLister
+	webhookLister v1alpha1.WebhookLister
+	devopsGetter  v1alpha3.Interface
 	cache         chan *auditv1alpha1.EventList
 	backend       *Backend
 }
 
-func NewAuditing(lister v1alpha1.WebhookLister, url string, stopCh <-chan struct{}) Auditing {
+func NewAuditing(informers informers.InformerFactory, url string, stopCh <-chan struct{}) Auditing {
 
 	a := &auditing{
-		lister: lister,
+		webhookLister: informers.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
+		devopsGetter:  devops.New(informers.KubeSphereSharedInformerFactory()),
 		cache:         make(chan *auditv1alpha1.EventList, DefaultCacheCapacity),
 	}
 
@@ -54,7 +61,7 @@ func NewAuditing(lister v1alpha1.WebhookLister, url string, stopCh <-chan struct
 }
 
 func (a *auditing) getAuditLevel() audit.Level {
-	wh, err := a.lister.Get(DefaultWebhook)
+	wh, err := a.webhookLister.Get(DefaultWebhook)
 	if err != nil {
 		klog.V(8).Info(err)
 		return audit.LevelNone
@@ -73,7 +80,7 @@ func (a *auditing) Enabled() bool {
 }
 
 func (a *auditing) K8sAuditingEnabled() bool {
-	wh, err := a.lister.Get(DefaultWebhook)
+	wh, err := a.webhookLister.Get(DefaultWebhook)
 	if err != nil {
 		klog.V(8).Info(err)
 		return false
@@ -105,6 +112,7 @@ func (a *auditing) LogRequestObject(req *http.Request, info *request.RequestInfo
 	}
 
 	e := &auditv1alpha1.Event{
+		Devops:    info.DevOps,
 		Workspace: info.Workspace,
 		Cluster:   info.Cluster,
 		Event: audit.Event{
@@ -115,7 +123,7 @@ func (a *auditing) LogRequestObject(req *http.Request, info *request.RequestInfo
 			Stage:                    audit.StageResponseComplete,
 			ImpersonatedUser:         nil,
 			UserAgent:                req.UserAgent(),
-			RequestReceivedTimestamp: v1.NewMicroTime(time.Now()),
+			RequestReceivedTimestamp: v1.NowMicro(),
 			Annotations:              nil,
 			ObjectRef: &audit.ObjectReference{
 				Resource:        info.Resource,
@@ -130,16 +138,22 @@ func (a *auditing) LogRequestObject(req *http.Request, info *request.RequestInfo
 		},
 	}
 
-	// Handle the devops request which request url matched /devops/{devops}/kind.
+	// Get the workspace which the devops project be in.
-	if len(info.Parts) >= 3 && info.Parts[0] == "devops" {
+	if len(e.Devops) > 0 && len(e.Workspace) == 0 {
-		e.ObjectRef.Subresource = ""
+		res, err := a.devopsGetter.List("", query.New())
-		e.Devops = info.Parts[1]
+		if err != nil {
-		// set resource as kind
+			klog.Error(err)
-		e.ObjectRef.Resource = info.Parts[2]
+		}
 
-		// If the request url matched /devops/{devops}/kind/{kind}, set resource name as {kind}
+		for _, obj := range res.Items {
-		if len(info.Parts) >= 4 {
+			d := obj.(*devopsv1alpha3.DevOpsProject)
-			e.ObjectRef.Name = info.Parts[3]
+
+			if d.Name == e.Devops {
+				e.Workspace = d.Labels["kubesphere.io/workspace"]
+			} else if d.Status.AdminNamespace == e.Devops {
+				e.Workspace = d.Labels["kubesphere.io/workspace"]
+				e.Devops = d.Name
+			}
 		}
 	}
 
@@ -185,7 +199,7 @@ func (a *auditing) LogRequestObject(req *http.Request, info *request.RequestInfo
 
 func (a *auditing) LogResponseObject(e *auditv1alpha1.Event, resp *ResponseCapture) {
 
-	e.StageTimestamp = v1.NewMicroTime(time.Now())
+	e.StageTimestamp = v1.NowMicro()
 	e.ResponseStatus = &v1.Status{Code: int32(resp.StatusCode())}
 	if e.Level.GreaterOrEqual(audit.LevelRequestResponse) {
 		e.ResponseObject = &runtime.Unknown{Raw: resp.Bytes()}

pkg/apiserver/auditing/types_test.go

@@ -9,11 +9,12 @@ import (
 	"k8s.io/apiserver/pkg/apis/audit"
 	"k8s.io/apiserver/pkg/authentication/user"
 	k8srequest "k8s.io/apiserver/pkg/endpoints/request"
+	fakek8s "k8s.io/client-go/kubernetes/fake"
 	auditingv1alpha1 "kubesphere.io/kubesphere/pkg/apis/auditing/v1alpha1"
 	v1alpha12 "kubesphere.io/kubesphere/pkg/apiserver/auditing/v1alpha1"
 	"kubesphere.io/kubesphere/pkg/apiserver/request"
 	"kubesphere.io/kubesphere/pkg/client/clientset/versioned/fake"
-	ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
+	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/utils/iputil"
 	"net/http"
 	"net/http/httptest"
@@ -37,13 +38,15 @@ func TestGetAuditLevel(t *testing.T) {
 		},
 	}
 
-	informer := ksinformers.NewSharedInformerFactory(fake.NewSimpleClientset(), noResyncPeriodFunc())
+	ksClient := fake.NewSimpleClientset()
+	k8sClient := fakek8s.NewSimpleClientset()
+	fakeInformerFactory := informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
 
 	a := auditing{
-		lister: informer.Auditing().V1alpha1().Webhooks().Lister(),
+		webhookLister: fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 	}
 
-	err := informer.Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
+	err := fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
 	if err != nil {
 		panic(err)
 	}
@@ -64,13 +67,15 @@ func TestAuditing_Enabled(t *testing.T) {
 		},
 	}
 
-	informer := ksinformers.NewSharedInformerFactory(fake.NewSimpleClientset(), noResyncPeriodFunc())
+	ksClient := fake.NewSimpleClientset()
+	k8sClient := fakek8s.NewSimpleClientset()
+	fakeInformerFactory := informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
 
 	a := auditing{
-		lister: informer.Auditing().V1alpha1().Webhooks().Lister(),
+		webhookLister: fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 	}
 
-	err := informer.Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
+	err := fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
 	if err != nil {
 		panic(err)
 	}
@@ -92,13 +97,15 @@ func TestAuditing_K8sAuditingEnabled(t *testing.T) {
 		},
 	}
 
-	informer := ksinformers.NewSharedInformerFactory(fake.NewSimpleClientset(), noResyncPeriodFunc())
+	ksClient := fake.NewSimpleClientset()
+	k8sClient := fakek8s.NewSimpleClientset()
+	fakeInformerFactory := informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
 
 	a := auditing{
-		lister: informer.Auditing().V1alpha1().Webhooks().Lister(),
+		webhookLister: fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 	}
 
-	err := informer.Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
+	err := fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
 	if err != nil {
 		panic(err)
 	}
@@ -120,13 +127,15 @@ func TestAuditing_LogRequestObject(t *testing.T) {
 		},
 	}
 
-	informer := ksinformers.NewSharedInformerFactory(fake.NewSimpleClientset(), noResyncPeriodFunc())
+	ksClient := fake.NewSimpleClientset()
+	k8sClient := fakek8s.NewSimpleClientset()
+	fakeInformerFactory := informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
 
 	a := auditing{
-		lister: informer.Auditing().V1alpha1().Webhooks().Lister(),
+		webhookLister: fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 	}
 
-	err := informer.Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
+	err := fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
 	if err != nil {
 		panic(err)
 	}
@@ -208,13 +217,15 @@ func TestAuditing_LogResponseObject(t *testing.T) {
 		},
 	}
 
-	informer := ksinformers.NewSharedInformerFactory(fake.NewSimpleClientset(), noResyncPeriodFunc())
+	ksClient := fake.NewSimpleClientset()
+	k8sClient := fakek8s.NewSimpleClientset()
+	fakeInformerFactory := informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
 
 	a := auditing{
-		lister: informer.Auditing().V1alpha1().Webhooks().Lister(),
+		webhookLister: fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Lister(),
 	}
 
-	err := informer.Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
+	err := fakeInformerFactory.KubeSphereSharedInformerFactory().Auditing().V1alpha1().Webhooks().Informer().GetIndexer().Add(webhook)
 	if err != nil {
 		panic(err)
 	}

pkg/models/auditing/events.go

@@ -38,6 +38,10 @@ func NewEventsOperator(client auditing.Client) Interface {
 func (eo *eventsOperator) Events(queryParam *v1alpha1.Query,
 	MutateFilterFunc func(*auditing.Filter)) (*v1alpha1.APIResponse, error) {
 	filter := &auditing.Filter{
+		ObjectRefNamespaces:     stringutils.Split(queryParam.ObjectRefNamespaceFilter, ","),
+		ObjectRefNamespaceFuzzy: stringutils.Split(queryParam.ObjectRefNamespaceSearch, ","),
+		Workspaces:              stringutils.Split(queryParam.WorkspaceFilter, ","),
+		WorkspaceFuzzy:          stringutils.Split(queryParam.WorkspaceSearch, ","),
 		ObjectRefNames:          stringutils.Split(queryParam.ObjectRefNameFilter, ","),
 		ObjectRefNameFuzzy:      stringutils.Split(queryParam.ObjectRefNameSearch, ","),
 		Levels:                  stringutils.Split(queryParam.LevelFilter, ","),
@@ -70,19 +74,19 @@ func (eo *eventsOperator) Events(queryParam *v1alpha1.Query,
 	var err error
 	switch queryParam.Operation {
 	case "histogram":
-		if len(filter.ObjectRefNamespaceMap) == 0 {
+		if len(filter.ObjectRefNamespaceMap) == 0 && len(filter.WorkspaceMap) == 0 {
 			ar.Histogram = &auditing.Histogram{}
 		} else {
 			ar.Histogram, err = eo.client.CountOverTime(filter, queryParam.Interval)
 		}
 	case "statistics":
-		if len(filter.ObjectRefNamespaceMap) == 0 {
+		if len(filter.ObjectRefNamespaceMap) == 0 && len(filter.WorkspaceMap) == 0 {
 			ar.Statistics = &auditing.Statistics{}
 		} else {
 			ar.Statistics, err = eo.client.StatisticsOnResources(filter)
 		}
 	default:
-		if len(filter.ObjectRefNamespaceMap) == 0 {
+		if len(filter.ObjectRefNamespaceMap) == 0 && len(filter.WorkspaceMap) == 0 {
 			ar.Events = &auditing.Events{}
 		} else {
 			ar.Events, err = eo.client.SearchAuditingEvent(filter, queryParam.From, queryParam.Size, queryParam.Sort)

pkg/models/tenant/tenant.go

@@ -609,6 +609,40 @@ func (t *tenantOperator) listIntersectedNamespaces(workspaces, workspaceSubstrs,
 	return iNamespaces, nil
 }
 
+// listIntersectedWorkspaces returns a list of workspaces that MUST meet ALL the following filters:
+// 1. If `workspaces` is not empty, the workspace SHOULD be one of the specified workpsaces.
+// 2. Else if `workspaceSubstrs` is not empty, the workspace SHOULD be contains one of the specified substrings.
+// 3. Else, return all workspace in the cluster.
+func (t *tenantOperator) listIntersectedWorkspaces(workspaces, workspaceSubstrs []string) ([]*tenantv1alpha1.Workspace, error) {
+	var (
+		workspaceSet = stringSet(workspaces)
+		iWorkspaces  []*tenantv1alpha1.Workspace
+	)
+
+	result, err := t.resourceGetter.List("workspaces", "", query.New())
+	if err != nil {
+		return nil, err
+	}
+	for _, obj := range result.Items {
+		ws, ok := obj.(*tenantv1alpha1.Workspace)
+		if !ok {
+			continue
+		}
+
+		if len(workspaceSet) > 0 {
+			if _, ok := workspaceSet[ws.Name]; !ok {
+				continue
+			}
+		}
+		if len(workspaceSubstrs) > 0 && !stringContains(ws.Name, workspaceSubstrs) {
+			continue
+		}
+
+		iWorkspaces = append(iWorkspaces, ws)
+	}
+	return iWorkspaces, nil
+}
+
 func (t *tenantOperator) Events(user user.Info, queryParam *eventsv1alpha1.Query) (*eventsv1alpha1.APIResponse, error) {
 	iNamespaces, err := t.listIntersectedNamespaces(
 		stringutils.Split(queryParam.WorkspaceFilter, ","),
@@ -870,7 +904,16 @@ func (t *tenantOperator) Auditing(user user.Info, queryParam *auditingv1alpha1.Q
 		return nil, err
 	}
 
+	iWorkspaces, err := t.listIntersectedWorkspaces(
+		stringutils.Split(queryParam.WorkspaceFilter, ","),
+		stringutils.Split(queryParam.WorkspaceSearch, ","))
+	if err != nil {
+		klog.Error(err)
+		return nil, err
+	}
+
 	namespaceCreateTimeMap := make(map[string]time.Time)
+	workspaceCreateTimeMap := make(map[string]time.Time)
 
 	// Now auditing and event have the same authorization mechanism, so we can determine whether the user
 	// has permission to view the auditing log in ns by judging whether the user has the permission to view the event in ns.
@@ -894,6 +937,30 @@ func (t *tenantOperator) Auditing(user user.Info, queryParam *auditingv1alpha1.Q
 			namespaceCreateTimeMap[ns.Name] = ns.CreationTimestamp.Time
 		}
 	}
+
+	// Now auditing and event have the same authorization mechanism, so we can determine whether the user
+	// has permission to view the auditing log in ws by judging whether the user has the permission to view the event in ws.
+	for _, ws := range iWorkspaces {
+		listEvts := authorizer.AttributesRecord{
+			User:            user,
+			Verb:            "list",
+			APIGroup:        "",
+			APIVersion:      "v1",
+			Workspace:       ws.Name,
+			Resource:        "events",
+			ResourceRequest: true,
+			ResourceScope:   request.WorkspaceScope,
+		}
+		decision, _, err := t.authorizer.Authorize(listEvts)
+		if err != nil {
+			klog.Error(err)
+			return nil, err
+		}
+		if decision == authorizer.DecisionAllow {
+			workspaceCreateTimeMap[ws.Name] = ws.CreationTimestamp.Time
+		}
+	}
+
 	// If there are no ns and ws query conditions,
 	// those events with empty `objectRef.namespace` will also be listed when user can list all events
 	if len(queryParam.WorkspaceFilter) == 0 && len(queryParam.ObjectRefNamespaceFilter) == 0 &&
@@ -914,11 +981,13 @@ func (t *tenantOperator) Auditing(user user.Info, queryParam *auditingv1alpha1.Q
 		}
 		if decision == authorizer.DecisionAllow {
 			namespaceCreateTimeMap[""] = time.Time{}
+			workspaceCreateTimeMap[""] = time.Time{}
 		}
 	}
 
 	return t.auditing.Events(queryParam, func(filter *auditingclient.Filter) {
 		filter.ObjectRefNamespaceMap = namespaceCreateTimeMap
+		filter.WorkspaceMap = workspaceCreateTimeMap
 	})
 }
 

pkg/simple/client/auditing/elasticsearch/elasticsearch.go

@@ -305,7 +305,7 @@ func parseToQueryPart(f *auditing.Filter) interface{} {
 		"bool": &b,
 	}
 
-	if len(f.ObjectRefNamespaceMap) > 0 {
+	if len(f.ObjectRefNamespaceMap) > 0 || len(f.WorkspaceMap) > 0 {
 		bi := BoolBody{MinimumShouldMatch: &mini}
 		for k, v := range f.ObjectRefNamespaceMap {
 			bi.Should = append(bi.Should, map[string]interface{}{
@@ -322,6 +322,23 @@ func parseToQueryPart(f *auditing.Filter) interface{} {
 				},
 			})
 		}
+
+		for k, v := range f.WorkspaceMap {
+			bi.Should = append(bi.Should, map[string]interface{}{
+				"bool": &BoolBody{
+					Filter: []map[string]interface{}{{
+						"match_phrase": map[string]string{"Workspace.keyword": k},
+					}, {
+						"range": map[string]interface{}{
+							"RequestReceivedTimestamp": map[string]interface{}{
+								"gte": v,
+							},
+						},
+					}},
+				},
+			})
+		}
+
 		if len(bi.Should) > 0 {
 			b.Filter = append(b.Filter, map[string]interface{}{"bool": &bi})
 		}
@@ -343,6 +360,36 @@ func parseToQueryPart(f *auditing.Filter) interface{} {
 		return &bi
 	}
 
+	if len(f.ObjectRefNamespaces) > 0 {
+		if bi := shouldBoolbody("match_phrase", "ObjectRef.Namespace.keyword",
+			f.ObjectRefNamespaces, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.ObjectRefNamespaceFuzzy) > 0 {
+		if bi := shouldBoolbody("wildcard", "ObjectRef.Namespace",
+			f.ObjectRefNamespaceFuzzy, func(s string) string {
+				return fmt.Sprintf("*" + s + "*")
+			}); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+
+	if len(f.Workspaces) > 0 {
+		if bi := shouldBoolbody("match_phrase", "Workspace.keyword",
+			f.Workspaces, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.WorkspaceFuzzy) > 0 {
+		if bi := shouldBoolbody("wildcard", "Workspace",
+			f.WorkspaceFuzzy, func(s string) string {
+				return fmt.Sprintf("*" + s + "*")
+			}); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+
 	if len(f.ObjectRefNames) > 0 {
 		if bi := shouldBoolbody("match_phrase_prefix", "ObjectRef.Name.keyword",
 			f.ObjectRefNames, nil); bi != nil {

pkg/simple/client/auditing/interface.go

@@ -28,6 +28,11 @@ type Client interface {
 
 type Filter struct {
 	ObjectRefNamespaceMap   map[string]time.Time
+	WorkspaceMap            map[string]time.Time
+	ObjectRefNamespaces     []string
+	ObjectRefNamespaceFuzzy []string
+	Workspaces              []string
+	WorkspaceFuzzy          []string
 	ObjectRefNames          []string
 	ObjectRefNameFuzzy      []string
 	Levels                  []string