feat(enterprise): audit logs for alpha and zero (#7295)

* starting work on audit logs

* making audit logs to work with alpha

* making audit logs to work with zero too

* adding endpoint to grpc audit

* adding skip method for grpc audits

* making zero and alpha logs from the start itself.

* fixing zero init audit process

* adding dgraph audit tool and logs encryption

* adding logwriter to handle encryption and everything

* adding test cases to check requests are getting logged into the audit logs

* adding interceptor ee version

* basic refactoring and log message truncate functionality

* making log writer performant using buffered writer

* gracefully closing all the go routines

* fixing oss build

* fixing failed test case

Author: aman-bansal

dgraph/cmd/alpha/run.go

@@ -36,6 +36,8 @@ import (
 	"time"
 
 	badgerpb "github.com/dgraph-io/badger/v3/pb"
+	"github.com/dgraph-io/dgraph/ee/audit"
+
 	"github.com/dgraph-io/dgo/v200/protos/api"
 	"github.com/dgraph-io/dgraph/edgraph"
 	"github.com/dgraph-io/dgraph/ee/enc"
@@ -191,6 +193,14 @@ they form a Raft group and provide synchronous replication.
 		`Cache percentages summing up to 100 for various caches (FORMAT:
 		PostingListCache,PstoreBlockCache,PstoreIndexCache,WAL).`)
 
+	flag.String("audit", "",
+		`Various audit options.
+	dir=/path/to/audits to define the path where to store the audit logs.
+	compress=true/false to enabled the compression of old audit logs (default behaviour is false).
+	encrypt_file=enc/key/file enables the audit log encryption with the key path provided with the
+	flag.
+	Sample flag could look like --audit dir=aa;encrypt_file=/filepath;compress=true`)
+
 	// TLS configurations
 	x.RegisterServerTLSFlags(flag)
 }
@@ -379,6 +389,7 @@ func serveGRPC(l net.Listener, tlsCfg *tls.Config, closer *z.Closer) {
 		grpc.MaxSendMsgSize(x.GrpcMaxSize),
 		grpc.MaxConcurrentStreams(1000),
 		grpc.StatsHandler(&ocgrpc.ServerHandler{}),
+		grpc.UnaryInterceptor(audit.AuditRequestGRPC),
 	}
 	if tlsCfg != nil {
 		opt = append(opt, grpc.Creds(credentials.NewTLS(tlsCfg)))
@@ -417,15 +428,18 @@ func setupServer(closer *z.Closer) {
 		log.Fatal(err)
 	}
 
-	http.HandleFunc("/query", queryHandler)
-	http.HandleFunc("/query/", queryHandler)
-	http.HandleFunc("/mutate", mutationHandler)
-	http.HandleFunc("/mutate/", mutationHandler)
-	http.HandleFunc("/commit", commitHandler)
-	http.HandleFunc("/alter", alterHandler)
-	http.HandleFunc("/health", healthCheck)
-	http.HandleFunc("/state", stateHandler)
-	http.HandleFunc("/jemalloc", x.JemallocHandler)
+	baseMux := http.NewServeMux()
+	http.Handle("/", audit.AuditRequestHttp(baseMux))
+
+	baseMux.HandleFunc("/query", queryHandler)
+	baseMux.HandleFunc("/query/", queryHandler)
+	baseMux.HandleFunc("/mutate", mutationHandler)
+	baseMux.HandleFunc("/mutate/", mutationHandler)
+	baseMux.HandleFunc("/commit", commitHandler)
+	baseMux.HandleFunc("/alter", alterHandler)
+	baseMux.HandleFunc("/health", healthCheck)
+	baseMux.HandleFunc("/state", stateHandler)
+	baseMux.HandleFunc("/jemalloc", x.JemallocHandler)
 
 	// TODO: Figure out what this is for?
 	http.HandleFunc("/debug/store", storeStatsHandler)
@@ -451,8 +465,9 @@ func setupServer(closer *z.Closer) {
 	var gqlHealthStore *admin.GraphQLHealthStore
 	// Do not use := notation here because adminServer is a global variable.
 	mainServer, adminServer, gqlHealthStore = admin.NewServers(introspection, &globalEpoch, closer)
-	http.Handle("/graphql", mainServer.HTTPHandler())
-	http.HandleFunc("/probe/graphql", func(w http.ResponseWriter, r *http.Request) {
+	baseMux.Handle("/graphql", mainServer.HTTPHandler())
+	baseMux.HandleFunc("/probe/graphql", func(w http.ResponseWriter,
+		r *http.Request) {
 		healthStatus := gqlHealthStore.GetHealth()
 		httpStatusCode := http.StatusOK
 		if !healthStatus.Healthy {
@@ -463,18 +478,19 @@ func setupServer(closer *z.Closer) {
 		x.Check2(w.Write([]byte(fmt.Sprintf(`{"status":"%s","schemaUpdateCounter":%d}`,
 			healthStatus.StatusMsg, atomic.LoadUint64(&globalEpoch)))))
 	})
-	http.Handle("/admin", allowedMethodsHandler(allowedMethods{
+	baseMux.Handle("/admin", allowedMethodsHandler(allowedMethods{
 		http.MethodGet:     true,
 		http.MethodPost:    true,
 		http.MethodOptions: true,
 	}, adminAuthHandler(adminServer.HTTPHandler())))
 
-	http.Handle("/admin/schema", adminAuthHandler(http.HandlerFunc(func(w http.ResponseWriter,
+	baseMux.Handle("/admin/schema", adminAuthHandler(http.HandlerFunc(func(
+		w http.ResponseWriter,
 		r *http.Request) {
 		adminSchemaHandler(w, r, adminServer)
 	})))
 
-	http.Handle("/admin/schema/validate", http.HandlerFunc(func(w http.ResponseWriter,
+	baseMux.HandleFunc("/admin/schema/validate", func(w http.ResponseWriter,
 		r *http.Request) {
 		schema := readRequest(w, r)
 		w.Header().Set("Content-Type", "application/json")
@@ -489,26 +505,28 @@ func setupServer(closer *z.Closer) {
 		w.WriteHeader(http.StatusBadRequest)
 		errs := strings.Split(strings.TrimSpace(err.Error()), "\n")
 		x.SetStatusWithErrors(w, x.ErrorInvalidRequest, errs)
-	}))
+	})
 
-	http.Handle("/admin/shutdown", allowedMethodsHandler(allowedMethods{http.MethodGet: true},
+	baseMux.Handle("/admin/shutdown", allowedMethodsHandler(allowedMethods{http.
+		MethodGet: true},
 		adminAuthHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			shutDownHandler(w, r, adminServer)
 		}))))
 
-	http.Handle("/admin/draining", allowedMethodsHandler(allowedMethods{
+	baseMux.Handle("/admin/draining", allowedMethodsHandler(allowedMethods{
 		http.MethodPut:  true,
 		http.MethodPost: true,
 	}, adminAuthHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		drainingHandler(w, r, adminServer)
 	}))))
 
-	http.Handle("/admin/export", allowedMethodsHandler(allowedMethods{http.MethodGet: true},
+	baseMux.Handle("/admin/export", allowedMethodsHandler(
+		allowedMethods{http.MethodGet: true},
 		adminAuthHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			exportHandler(w, r, adminServer)
 		}))))
 
-	http.Handle("/admin/config/cache_mb", allowedMethodsHandler(allowedMethods{
+	baseMux.Handle("/admin/config/cache_mb", allowedMethodsHandler(allowedMethods{
 		http.MethodGet: true,
 		http.MethodPut: true,
 	}, adminAuthHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -520,10 +538,10 @@ func setupServer(closer *z.Closer) {
 	glog.Infof("Bringing up GraphQL HTTP admin API at %s/admin", addr)
 
 	// Add OpenCensus z-pages.
-	zpages.Handle(http.DefaultServeMux, "/z")
+	zpages.Handle(baseMux, "/z")
 
-	http.HandleFunc("/", homeHandler)
-	http.HandleFunc("/ui/keywords", keywordHandler)
+	baseMux.Handle("/", http.HandlerFunc(homeHandler))
+	baseMux.Handle("/ui/keywords", http.HandlerFunc(keywordHandler))
 
 	// Initialize the servers.
 	admin.ServerCloser.AddRunning(3)
@@ -585,6 +603,8 @@ func run() {
 	walCache := (cachePercent[3] * (totalCache << 20)) / 100
 
 	ctype, clevel := x.ParseCompression(Alpha.Conf.GetString("badger.compression"))
+
+	conf := audit.GetAuditConf(Alpha.Conf.GetString("audit"))
 	opts := worker.Options{
 		PostingDir:                 Alpha.Conf.GetString("postings"),
 		WALDir:                     Alpha.Conf.GetString("wal"),
@@ -597,6 +617,7 @@ func run() {
 
 		MutationsMode: worker.AllowMutations,
 		AuthToken:     Alpha.Conf.GetString("auth_token"),
+		Audit:         conf,
 	}
 
 	secretFile := Alpha.Conf.GetString("acl_secret_file")
@@ -658,6 +679,8 @@ func run() {
 		LudicrousConcurrency: Alpha.Conf.GetInt("ludicrous_concurrency"),
 		TLSClientConfig:      tlsClientConf,
 		TLSServerConfig:      tlsServerConf,
+		HmacSecret:           opts.HmacSecret,
+		Audit:                opts.Audit != nil,
 	}
 	x.WorkerConfig.Parse(Alpha.Conf)
 
@@ -699,6 +722,9 @@ func run() {
 
 	worker.InitServerState()
 
+	// Audit is enterprise feature.
+	x.Check(audit.InitAuditorIfNecessary(opts.Audit, worker.EnterpriseEnabled))
+
 	if Alpha.Conf.GetBool("expose_trace") {
 		// TODO: Remove this once we get rid of event logs.
 		trace.AuthRequest = func(req *http.Request) (any, sensitive bool) {
@@ -792,6 +818,8 @@ func run() {
 	adminCloser.SignalAndWait()
 	glog.Infoln("adminCloser closed.")
 
+	audit.Close()
+
 	worker.State.Dispose()
 	x.RemoveCidFile()
 	glog.Info("worker.State disposed.")

dgraph/cmd/bulk/count_index.go

@@ -156,7 +156,7 @@ func (c *countIndexer) writeIndex(buf *z.Buffer) {
 		encoder = codec.Encoder{BlockSize: 256, Alloc: alloc}
 		pl.Reset()
 
-		// Flush out the buffer.
+		// flush out the buffer.
 		if outBuf.LenNoPadding() > 4<<20 {
 			x.Check(c.writer.Write(outBuf))
 			outBuf.Reset()

dgraph/cmd/bulk/reduce.go

@@ -292,7 +292,7 @@ func (r *reducer) writeTmpSplits(ci *countIndexer, wg *sync.WaitGroup) {
 		}
 
 		for i := 0; i < len(kvs.Kv); i += maxSplitBatchLen {
-			// Flush the write batch when the max batch length is reached to prevent the
+			// flush the write batch when the max batch length is reached to prevent the
 			// value log from growing over the allowed limit.
 			if splitBatchLen >= maxSplitBatchLen {
 				x.Check(ci.splitWriter.Flush())

dgraph/cmd/live/batch.go

@@ -34,7 +34,6 @@ import (
 	"github.com/dgraph-io/badger/v3"
 	"github.com/dgraph-io/dgo/v200"
 	"github.com/dgraph-io/dgo/v200/protos/api"
-	"github.com/dgraph-io/dgraph/dgraph/cmd/zero"
 	"github.com/dgraph-io/dgraph/gql"
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/tok"
@@ -132,7 +131,7 @@ func handleError(err error, isRetry bool) {
 		dur := time.Duration(1+rand.Intn(10)) * time.Minute
 		fmt.Printf("Server is overloaded. Will retry after %s.\n", dur.Round(time.Minute))
 		time.Sleep(dur)
-	case err != zero.ErrConflict && err != dgo.ErrAborted:
+	case err != x.ErrConflict && err != dgo.ErrAborted:
 		fmt.Printf("Error while mutating: %v s.Code %v\n", s.Message(), s.Code())
 	}
 }

dgraph/cmd/root_ee.go

@@ -14,6 +14,7 @@ package cmd
 
 import (
 	acl "github.com/dgraph-io/dgraph/ee/acl"
+	"github.com/dgraph-io/dgraph/ee/audit"
 	"github.com/dgraph-io/dgraph/ee/backup"
 )
 
@@ -24,5 +25,6 @@ func init() {
 		&backup.LsBackup,
 		&backup.ExportBackup,
 		&acl.CmdAcl,
+		&audit.CmdAudit,
 	)
 }

dgraph/cmd/zero/license_ee.go

@@ -20,6 +20,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/dgraph-io/dgraph/ee/audit"
+
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/x"
 	"github.com/dgraph-io/ristretto/z"
@@ -91,6 +93,7 @@ func (n *node) updateEnterpriseState(closer *z.Closer) {
 			active := time.Now().UTC().Before(expiry)
 			if !active {
 				n.server.expireLicense()
+				audit.Close()
 				glog.Warningf("Your enterprise license has expired and enterprise features are " +
 					"disabled. To continue using enterprise features, apply a valid license. To receive " +
 					"a new license, contact us at https://dgraph.io/contact.")

dgraph/cmd/zero/oracle.go

@@ -134,7 +134,7 @@ func (o *Oracle) commit(src *api.TxnContext) error {
 	defer o.Unlock()
 
 	if o.hasConflict(src) {
-		return ErrConflict
+		return x.ErrConflict
 	}
 	// We store src.Keys as string to ensure compatibility with all the various language clients we
 	// have. But, really they are just uint64s encoded as strings. We use base 36 during creation of
@@ -310,9 +310,6 @@ func (o *Oracle) MaxPending() uint64 {
 	return o.maxAssigned
 }
 
-// ErrConflict is returned when commit couldn't succeed due to conflicts.
-var ErrConflict = errors.New("Transaction conflict")
-
 // proposeTxn proposes a txn update, and then updates src to reflect the state
 // of the commit after proposal is run.
 func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error {

dgraph/cmd/zero/raft.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/dgraph/conn"
+	"github.com/dgraph-io/dgraph/ee/audit"
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/x"
 	"github.com/dgraph-io/ristretto/z"
@@ -404,6 +405,11 @@ func (n *node) applyProposal(e raftpb.Entry) (uint64, error) {
 		// Check expiry and set enabled accordingly.
 		expiry := time.Unix(state.License.ExpiryTs, 0).UTC()
 		state.License.Enabled = time.Now().UTC().Before(expiry)
+		if state.License.Enabled && opts.audit != nil {
+			if err := audit.InitAuditor(opts.audit); err != nil {
+				glog.Errorf("error while initializing audit logs %+v", err)
+			}
+		}
 	}
 	if p.Snapshot != nil {
 		if err := n.applySnapshot(p.Snapshot); err != nil {