Merge pull request #2498 from kmlebedev/s3_audit_log

docker/compose/fluent.json

@@ -0,0 +1,4 @@
+{
+    "fluent_port": 24224,
+    "fluent_host": "fluent"
+}

docker/compose/local-auditlog-compose.yml

@@ -0,0 +1,20 @@
+version: '2'
+
+services:
+  server:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8333:8333
+      - 9333:9333
+      - 19333:19333
+      - 8084:8080
+      - 18084:18080
+      - 8888:8888
+      - 18888:18888
+    command: "server -ip=server -filer -s3 -s3.auditLogConfig=/etc/seaweedfs/fluent.json -volume.max=0 -master.volumeSizeLimitMB=8 -volume.preStopSeconds=1"
+    volumes:
+      - ./fluent.json:/etc/seaweedfs/fluent.json
+  fluent:
+    image: fluent/fluentd:v1.14
+    ports:
+      - 24224:24224

go.mod

@@ -166,6 +166,7 @@ require (
 require (
 	cloud.google.com/go/kms v1.0.0 // indirect
 	github.com/d4l3k/messagediff v1.2.1 // indirect
+	github.com/fluent/fluent-logger-golang v1.8.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
@@ -173,7 +174,9 @@ require (
 	github.com/jcmturner/rpc/v2 v2.0.2 // indirect
 	github.com/mattn/go-runewidth v0.0.7 // indirect
 	github.com/mattn/go-sqlite3 v2.0.1+incompatible // indirect
+	github.com/philhofer/fwd v1.1.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/tinylib/msgp v1.1.6 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.0 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect

go.sum

@@ -234,6 +234,8 @@ github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4/go.mod h1:5tD+ne
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fclairamb/ftpserverlib v0.8.0 h1:ZsWUQ8Vg3Y8LIWRUAzVnFXY982Yztz2odDdK/UVJtik=
 github.com/fclairamb/ftpserverlib v0.8.0/go.mod h1:xF4cy07oCHA9ZorKehsFGqA/1UHYaonmqHK2g3P1X8U=
+github.com/fluent/fluent-logger-golang v1.8.0 h1:K/fUDqUAItNcdf/Rq7aA2d1apwqsNgNzzInlXZTwK28=
+github.com/fluent/fluent-logger-golang v1.8.0/go.mod h1:2/HCT/jTy78yGyeNGQLGQsjF3zzzAuy6Xlk6FCMV5eU=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
@@ -654,6 +656,8 @@ github.com/pelletier/go-toml v1.7.0/go.mod h1:vwGMzjaWMwyfHwgIBhI2YUM4fB6nL6lVAv
 github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9iaPbIdPPGyKcA8hKdoy6hAWba7Yac=
 github.com/peterh/liner v1.1.0 h1:f+aAedNJA6uk7+6rXsYBnhdo4Xux7ESLe+kcuVUF5os=
 github.com/peterh/liner v1.1.0/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
+github.com/philhofer/fwd v1.1.1 h1:GdGcTjf5RNAxwS4QLsiMzJYj5KEvPJD3Abr261yRQXQ=
+github.com/philhofer/fwd v1.1.1/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
@@ -791,6 +795,8 @@ github.com/tidwall/pretty v1.1.0 h1:K3hMW5epkdAVwibsQEfR/7Zj0Qgt4DxtNumTq/VloO8=
 github.com/tidwall/pretty v1.1.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tinylib/msgp v1.1.6 h1:i+SbKraHhnrf9M5MYmvQhFnbLhAXSDWF8WWsuyRdocw=
+github.com/tinylib/msgp v1.1.6/go.mod h1:75BAfg2hauQhs3qedfdDZmWAPcFMAvJE5b9rGOMufyw=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tsuna/gohbase v0.0.0-20201125011725-348991136365 h1:6iRwZdrFUzbcVYZwa8dXTIILGIxmmhjyUPJEcwzPGaU=
@@ -1170,6 +1176,7 @@ golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20200904185747-39188db58858/go.mod h1:Cj7w3i3Rnn0Xh82ur9kSqwfTHTeVxaDqrfMjpcNT6bE=
+golang.org/x/tools v0.0.0-20201022035929-9cf592e881e9/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201124115921-2c860bdd6e78/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=

weed/command/filer.go

@@ -84,6 +84,7 @@ func init() {
 	filerS3Options.tlsPrivateKey = cmdFiler.Flag.String("s3.key.file", "", "path to the TLS private key file")
 	filerS3Options.tlsCertificate = cmdFiler.Flag.String("s3.cert.file", "", "path to the TLS certificate file")
 	filerS3Options.config = cmdFiler.Flag.String("s3.config", "", "path to the config file")
+	filerS3Options.auditLogConfig = cmdFiler.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
 	filerS3Options.allowEmptyFolder = cmdFiler.Flag.Bool("s3.allowEmptyFolder", true, "allow empty folders")
 
 	// start webdav on filer

weed/command/s3.go

@@ -3,6 +3,7 @@ package command
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	"net/http"
 	"time"
 
@@ -31,6 +32,7 @@ type S3Options struct {
 	tlsCertificate   *string
 	metricsHttpPort  *int
 	allowEmptyFolder *bool
+	auditLogConfig   *string
 }
 
 func init() {
@@ -39,6 +41,7 @@ func init() {
 	s3StandaloneOptions.port = cmdS3.Flag.Int("port", 8333, "s3 server http listen port")
 	s3StandaloneOptions.domainName = cmdS3.Flag.String("domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
 	s3StandaloneOptions.config = cmdS3.Flag.String("config", "", "path to the config file")
+	s3StandaloneOptions.auditLogConfig = cmdS3.Flag.String("auditLogConfig", "", "path to the audit log config file")
 	s3StandaloneOptions.tlsPrivateKey = cmdS3.Flag.String("key.file", "", "path to the TLS private key file")
 	s3StandaloneOptions.tlsCertificate = cmdS3.Flag.String("cert.file", "", "path to the TLS certificate file")
 	s3StandaloneOptions.metricsHttpPort = cmdS3.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
@@ -192,6 +195,10 @@ func (s3opt *S3Options) startS3Server() bool {
 		glog.Fatalf("S3 API Server listener on %s error: %v", listenAddress, err)
 	}
 
+	if len(*s3opt.auditLogConfig) > 0 {
+		s3err.InitAuditLog(*s3opt.auditLogConfig)
+	}
+
 	if *s3opt.tlsPrivateKey != "" {
 		glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", util.Version(), *s3opt.port)
 		if err = httpS.ServeTLS(s3ApiListener, *s3opt.tlsCertificate, *s3opt.tlsPrivateKey); err != nil {

weed/command/server.go

@@ -131,6 +131,7 @@ func init() {
 	s3Options.tlsPrivateKey = cmdServer.Flag.String("s3.key.file", "", "path to the TLS private key file")
 	s3Options.tlsCertificate = cmdServer.Flag.String("s3.cert.file", "", "path to the TLS certificate file")
 	s3Options.config = cmdServer.Flag.String("s3.config", "", "path to the config file")
+	s3Options.auditLogConfig = cmdServer.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
 	s3Options.allowEmptyFolder = cmdServer.Flag.Bool("s3.allowEmptyFolder", true, "allow empty folders")
 
 	webdavOptions.port = cmdServer.Flag.Int("webdav.port", 7333, "webdav server http listen port")

weed/s3api/auth_credentials.go

@@ -236,7 +236,7 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
 
 	glog.V(3).Infof("user name: %v actions: %v, action: %v", identity.Name, identity.Actions, action)
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 
 	if !identity.canDo(action, bucket) {
 		return identity, s3err.ErrAccessDenied

weed/s3api/chunked_reader_v4.go

@@ -24,6 +24,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"errors"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	"hash"
 	"io"
@@ -90,7 +91,7 @@ func (iam *IdentityAccessManagement) calculateSeedSignature(r *http.Request) (cr
 		return nil, "", "", time.Time{}, s3err.ErrInvalidAccessKeyID
 	}
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	if !identity.canDo("Write", bucket) {
 		errCode = s3err.ErrAccessDenied
 		return

weed/s3api/http/header.go

@@ -16,6 +16,12 @@
 
 package http
 
+import (
+	"github.com/gorilla/mux"
+	"net/http"
+	"strings"
+)
+
 // Standard S3 HTTP request constants
 const (
 	// S3 storage class
@@ -34,3 +40,14 @@ const (
 	AmzIdentityId = "s3-identity-id"
 	AmzIsAdmin    = "s3-is-admin" // only set to http request header as a context
 )
+
+func GetBucketAndObject(r *http.Request) (bucket, object string) {
+	vars := mux.Vars(r)
+	bucket = vars["bucket"]
+	object = vars["object"]
+	if !strings.HasPrefix(object, "/") {
+		object = "/" + object
+	}
+
+	return
+}

weed/s3api/s3api_bucket_handlers.go

@@ -78,7 +78,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 
 func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("PutBucketHandler %s", bucket)
 
 	// avoid duplicated buckets
@@ -133,13 +133,12 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
-
 	writeSuccessResponseEmpty(w, r)
 }
 
 func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteBucketHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
@@ -174,7 +173,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 
 func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("HeadBucketHandler %s", bucket)
 
 	if entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket); entry == nil || err == filer_pb.ErrNotFound {
@@ -219,7 +218,7 @@ func (s3a *S3ApiServer) hasAccess(r *http.Request, entry *filer_pb.Entry) bool {
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketAcl.html
 func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Request) {
 	// collect parameters
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("GetBucketAclHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
@@ -259,7 +258,7 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketLifecycleConfiguration.html
 func (s3a *S3ApiServer) GetBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
 	// collect parameters
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("GetBucketAclHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {

weed/s3api/s3api_handlers.go

@@ -28,6 +28,7 @@ func (s3a *S3ApiServer) AdjustedUrl(location *filer_pb.Location) string {
 
 func writeSuccessResponseXML(w http.ResponseWriter, r *http.Request, response interface{}) {
 	s3err.WriteXMLResponse(w, r, http.StatusOK, response)
+	s3err.PostLog(r, http.StatusOK, s3err.ErrNone)
 }
 
 func writeSuccessResponseEmpty(w http.ResponseWriter, r *http.Request) {

weed/s3api/s3api_object_copy_handlers.go

@@ -3,6 +3,7 @@ package s3api
 import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	weed_server "github.com/chrislusf/seaweedfs/weed/server"
 	"net/http"
@@ -16,7 +17,7 @@ import (
 
 func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
 
-	dstBucket, dstObject := getBucketAndObject(r)
+	dstBucket, dstObject := xhttp.GetBucketAndObject(r)
 
 	// Copy source path.
 	cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))
@@ -116,7 +117,7 @@ type CopyPartResult struct {
 func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Request) {
 	// https://docs.aws.amazon.com/AmazonS3/latest/dev/CopyingObjctsUsingRESTMPUapi.html
 	// https://docs.aws.amazon.com/AmazonS3/latest/API/API_UploadPartCopy.html
-	dstBucket, _ := getBucketAndObject(r)
+	dstBucket, _ := xhttp.GetBucketAndObject(r)
 
 	// Copy source path.
 	cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))

weed/s3api/s3api_object_handlers.go

@@ -16,10 +16,9 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/pquerna/cachecontrol/cacheobject"
 
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 
-	"github.com/gorilla/mux"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	weed_server "github.com/chrislusf/seaweedfs/weed/server"
@@ -51,7 +50,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 	// http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("PutObjectHandler %s %s", bucket, object)
 
 	_, err := validateContentMd5(r.Header)
@@ -133,7 +132,7 @@ func urlPathEscape(object string) string {
 
 func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("GetObjectHandler %s %s", bucket, object)
 
 	if strings.HasSuffix(r.URL.Path, "/") {
@@ -145,34 +144,34 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
 
 	s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
-
 }
 
 func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object)
 
 	destUrl := fmt.Sprintf("http://%s%s/%s%s",
 		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
 
 	s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
-
 }
 
 func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteObjectHandler %s %s", bucket, object)
 
 	destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
 		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
 
-	s3a.proxyToFiler(w, r, destUrl, func(proxyResponse *http.Response, w http.ResponseWriter) {
+	s3a.proxyToFiler(w, r, destUrl, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int) {
+		statusCode = http.StatusNoContent
 		for k, v := range proxyResponse.Header {
 			w.Header()[k] = v
 		}
-		w.WriteHeader(http.StatusNoContent)
+		w.WriteHeader(statusCode)
+		return statusCode
 	})
 }
 
@@ -210,7 +209,7 @@ type DeleteObjectsResponse struct {
 // DeleteMultipleObjectsHandler - Delete multiple objects
 func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteMultipleObjectsHandler %s", bucket)
 
 	deleteXMLBytes, err := io.ReadAll(r.Body)
@@ -227,14 +226,17 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 
 	var deletedObjects []ObjectIdentifier
 	var deleteErrors []DeleteError
+	var auditLog *s3err.AccessLog
 
 	directoriesWithDeletion := make(map[string]int)
 
+	if s3err.Logger != nil {
+		auditLog = s3err.GetAccessLog(r, http.StatusNoContent, s3err.ErrNone)
+	}
 	s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		// delete file entries
 		for _, object := range deleteObjects.Objects {
-
 			lastSeparator := strings.LastIndex(object.ObjectName, "/")
 			parentDirectoryPath, entryName, isDeleteData, isRecursive := "", object.ObjectName, true, false
 			if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
@@ -257,6 +259,10 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 					Key:     object.ObjectName,
 				})
 			}
+			if auditLog != nil {
+				auditLog.Key = entryName
+				s3err.PostAccessLog(auditLog)
+			}
 		}
 
 		// purge empty folders, only checking folders with deletions
@@ -309,7 +315,7 @@ var passThroughHeaders = []string{
 	"response-expires",
 }
 
-func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter)) {
+func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int)) {
 
 	glog.V(3).Infof("s3 proxying %s to %s", r.Method, destUrl)
 
@@ -363,20 +369,23 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 		}
 	}
 
-	responseFn(resp, w)
-
+	responseStatusCode := responseFn(resp, w)
+	s3err.PostLog(r, responseStatusCode, s3err.ErrNone)
 }
 
-func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) {
+func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int) {
 	for k, v := range proxyResponse.Header {
 		w.Header()[k] = v
 	}
 	if proxyResponse.Header.Get("Content-Range") != "" && proxyResponse.StatusCode == 200 {
 		w.WriteHeader(http.StatusPartialContent)
+		statusCode = http.StatusPartialContent
 	} else {
-		w.WriteHeader(proxyResponse.StatusCode)
+		statusCode = proxyResponse.StatusCode
 	}
+	w.WriteHeader(statusCode)
 	io.Copy(w, proxyResponse.Body)
+	return statusCode
 }
 
 func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader io.Reader) (etag string, code s3err.ErrorCode) {
@@ -438,17 +447,6 @@ func setEtag(w http.ResponseWriter, etag string) {
 	}
 }
 
-func getBucketAndObject(r *http.Request) (bucket, object string) {
-	vars := mux.Vars(r)
-	bucket = vars["bucket"]
-	object = vars["object"]
-	if !strings.HasPrefix(object, "/") {
-		object = "/" + object
-	}
-
-	return
-}
-
 func filerErrorToS3Error(errString string) s3err.ErrorCode {
 	if strings.HasPrefix(errString, "existing ") && strings.HasSuffix(errString, "is a directory") {
 		return s3err.ErrExistingObjectIsDirectory

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -142,6 +142,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 			Location: w.Header().Get("Location"),
 		}
 		s3err.WriteXMLResponse(w, r, http.StatusCreated, resp)
+		s3err.PostLog(r, http.StatusCreated, s3err.ErrNone)
 	case "200":
 		s3err.WriteEmptyResponse(w, r, http.StatusOK)
 	default:

weed/s3api/s3api_object_multipart_handlers.go

@@ -3,6 +3,7 @@ package s3api
 import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	weed_server "github.com/chrislusf/seaweedfs/weed/server"
 	"net/http"
@@ -23,7 +24,7 @@ const (
 
 // NewMultipartUploadHandler - New multipart upload.
 func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 
 	createMultipartUploadInput := &s3.CreateMultipartUploadInput{
 		Bucket:   aws.String(bucket),
@@ -55,7 +56,7 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http
 
 // CompleteMultipartUploadHandler - Completes multipart upload.
 func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 
 	// Get upload id.
 	uploadID, _, _, _ := getObjectResources(r.URL.Query())
@@ -79,7 +80,7 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
 
 // AbortMultipartUploadHandler - Aborts multipart upload.
 func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 
 	// Get upload id.
 	uploadID, _, _, _ := getObjectResources(r.URL.Query())
@@ -103,7 +104,7 @@ func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *ht
 
 // ListMultipartUploadsHandler - Lists multipart uploads.
 func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 
 	prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType := getBucketMultipartResources(r.URL.Query())
 	if maxUploads < 0 {
@@ -142,7 +143,7 @@ func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *ht
 
 // ListObjectPartsHandler - Lists object parts in a multipart upload.
 func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 
 	uploadID, partNumberMarker, maxParts, _ := getObjectResources(r.URL.Query())
 	if partNumberMarker < 0 {
@@ -175,7 +176,7 @@ func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Re
 
 // PutObjectPartHandler - Put an object part in a multipart upload.
 func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 
 	uploadID := r.URL.Query().Get("uploadId")
 	exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true)

weed/s3api/s3api_object_tagging_handlers.go

@@ -3,6 +3,7 @@ package s3api
 import (
 	"encoding/xml"
 	"fmt"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"io"
 	"net/http"
 
@@ -16,7 +17,7 @@ import (
 // API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectTagging.html
 func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("GetObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
@@ -42,7 +43,7 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
 // API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectTagging.html
 func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("PutObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
@@ -91,14 +92,14 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	}
 
 	w.WriteHeader(http.StatusOK)
-
+	s3err.PostLog(r, http.StatusOK, s3err.ErrNone)
 }
 
 // DeleteObjectTaggingHandler Delete object tagging
 // API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjectTagging.html
 func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
-	bucket, object := getBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
@@ -117,4 +118,5 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt
 	}
 
 	w.WriteHeader(http.StatusNoContent)
+	s3err.PostLog(r, http.StatusNoContent, s3err.ErrNone)
 }

weed/s3api/s3api_objects_list_handlers.go

@@ -39,7 +39,7 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 	// https://docs.aws.amazon.com/AmazonS3/latest/API/v2-RESTBucketGET.html
 
 	// collect parameters
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("ListObjectsV2Handler %s", bucket)
 
 	originalPrefix, continuationToken, startAfter, delimiter, _, maxKeys := getListObjectsV2Args(r.URL.Query())
@@ -95,7 +95,7 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
 	// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTBucketGET.html
 
 	// collect parameters
-	bucket, _ := getBucketAndObject(r)
+	bucket, _ := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("ListObjectsV1Handler %s", bucket)
 
 	originalPrefix, marker, delimiter, maxKeys := getListObjectsV1Args(r.URL.Query())

weed/s3api/s3api_server.go

@@ -38,7 +38,6 @@ func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer
 	s3ApiServer.registerRouter(router)
 
 	go s3ApiServer.subscribeMetaEvents("s3", filer.IamConfigDirecotry+"/"+filer.IamIdentityFile, time.Now().UnixNano())
-
 	return s3ApiServer, nil
 }
 
@@ -132,7 +131,6 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 
 		// DeleteBucketLifecycleConfiguration
 		bucket.Methods("DELETE").HandlerFunc(s3a.iam.Auth(s3a.DeleteBucketLifecycleHandler, ACTION_WRITE)).Queries("lifecycle", "")
-
 		// ListObjectsV1 (Legacy)
 		bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV1Handler, ACTION_LIST), "LIST"))
 

weed/s3api/s3err/audit_fluent.go

@@ -0,0 +1,170 @@
+package s3err
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
+	"github.com/fluent/fluent-logger-golang/fluent"
+	"net/http"
+	"os"
+	"time"
+)
+
+type AccessLogExtend struct {
+	AccessLog
+	AccessLogHTTP
+}
+
+type AccessLog struct {
+	Bucket           string `msg:"bucket" json:"bucket"`                   // awsexamplebucket1
+	Time             int64  `msg:"time" json:"time"`                       // [06/Feb/2019:00:00:38 +0000]
+	RemoteIP         string `msg:"remote_ip" json:"remote_ip,omitempty"`   // 192.0.2.3
+	Requester        string `msg:"requester" json:"requester,omitempty"`   // IAM user id
+	RequestID        string `msg:"request_id" json:"request_id,omitempty"` // 3E57427F33A59F07
+	Operation        string `msg:"operation" json:"operation,omitempty"`   // REST.HTTP_method.resource_type REST.PUT.OBJECT
+	Key              string `msg:"key" json:"key,omitempty"`               // /photos/2019/08/puppy.jpg
+	ErrorCode        string `msg:"error_code" json:"error_code,omitempty"`
+	HostId           string `msg:"host_id" json:"host_id,omitempty"`
+	HostHeader       string `msg:"host_header" json:"host_header,omitempty"` // s3.us-west-2.amazonaws.com
+	UserAgent        string `msg:"user_agent" json:"user_agent,omitempty"`
+	HTTPStatus       int    `msg:"status" json:"status,omitempty"`
+	SignatureVersion string `msg:"signature_version" json:"signature_version,omitempty"`
+}
+
+type AccessLogHTTP struct {
+	RequestURI         string `json:"request_uri,omitempty"` // "GET /awsexamplebucket1/photos/2019/08/puppy.jpg?x-foo=bar HTTP/1.1"
+	BytesSent          string `json:"bytes_sent,omitempty"`
+	ObjectSize         string `json:"object_size,omitempty"`
+	TotalTime          int    `json:"total_time,omitempty"`
+	TurnAroundTime     int    `json:"turn_around_time,omitempty"`
+	Referer            string `json:"Referer,omitempty"`
+	VersionId          string `json:"version_id,omitempty"`
+	CipherSuite        string `json:"cipher_suite,omitempty"`
+	AuthenticationType string `json:"auth_type,omitempty"`
+	TLSVersion         string `json:"TLS_version,omitempty"`
+}
+
+const tag = "s3.access"
+
+var (
+	Logger   *fluent.Fluent
+	hostname = os.Getenv("HOSTNAME")
+)
+
+func InitAuditLog(config string) {
+	configContent, readErr := os.ReadFile(config)
+	if readErr != nil {
+		glog.Fatalf("fail to read fluent config %s : %v", config, readErr)
+	}
+	var fluentConfig fluent.Config
+	if err := json.Unmarshal(configContent, &fluentConfig); err != nil {
+		glog.Fatalf("fail to parse fluent config %s : %v", config, err)
+	}
+	var err error
+	Logger, err = fluent.New(fluentConfig)
+	if err != nil {
+		glog.Fatalf("fail to load fluent config: %v", err)
+	}
+}
+
+func getREST(httpMetod string, resourceType string) string {
+	return fmt.Sprintf("REST.%s.%s", httpMetod, resourceType)
+}
+
+func getResourceType(object string, query_key string, metod string) (string, bool) {
+	if object == "/" {
+		switch query_key {
+		case "delete":
+			return "BATCH.DELETE.OBJECT", true
+		case "tagging":
+			return getREST(metod, "OBJECTTAGGING"), true
+		case "lifecycle":
+			return getREST(metod, "LIFECYCLECONFIGURATION"), true
+		case "acl":
+			return getREST(metod, "ACCESSCONTROLPOLICY"), true
+		case "policy":
+			return getREST(metod, "BUCKETPOLICY"), true
+		default:
+			return getREST(metod, "BUCKET"), false
+		}
+	} else {
+		switch query_key {
+		case "tagging":
+			return getREST(metod, "OBJECTTAGGING"), true
+		default:
+			return getREST(metod, "OBJECT"), false
+		}
+	}
+}
+
+func getOperation(object string, r *http.Request) string {
+	queries := r.URL.Query()
+	var operation string
+	var queryFound bool
+	for key, _ := range queries {
+		operation, queryFound = getResourceType(object, key, r.Method)
+		if queryFound {
+			return operation
+		}
+	}
+	if len(queries) == 0 {
+		operation, _ = getResourceType(object, "", r.Method)
+	}
+	return operation
+}
+
+func GetAccessHttpLog(r *http.Request, statusCode int, s3errCode ErrorCode) AccessLogHTTP {
+	return AccessLogHTTP{
+		RequestURI: r.RequestURI,
+		Referer:    r.Header.Get("Referer"),
+	}
+}
+
+func GetAccessLog(r *http.Request, HTTPStatusCode int, s3errCode ErrorCode) *AccessLog {
+	bucket, key := xhttp.GetBucketAndObject(r)
+	var errorCode string
+	if s3errCode != ErrNone {
+		errorCode = GetAPIError(s3errCode).Code
+	}
+	remoteIP := r.Header.Get("X-Real-IP")
+	if len(remoteIP) == 0 {
+		remoteIP = r.RemoteAddr
+	}
+	hostHeader := r.Header.Get("Host")
+	if len(hostHeader) == 0 {
+		hostHeader = r.URL.Hostname()
+	}
+	return &AccessLog{
+		HostHeader: hostHeader,
+		RequestID:  r.Header.Get("X-Request-ID"),
+		RemoteIP:   remoteIP,
+		Requester:  r.Header.Get(xhttp.AmzIdentityId),
+		UserAgent:  r.Header.Get("UserAgent"),
+		HostId:     hostname,
+		Bucket:     bucket,
+		HTTPStatus: HTTPStatusCode,
+		Time:       time.Now().Unix(),
+		Key:        key,
+		Operation:  getOperation(key, r),
+		ErrorCode:  errorCode,
+	}
+}
+
+func PostLog(r *http.Request, HTTPStatusCode int, errorCode ErrorCode) {
+	if Logger == nil {
+		return
+	}
+	if err := Logger.Post(tag, *GetAccessLog(r, HTTPStatusCode, errorCode)); err != nil {
+		glog.Warning("Error while posting log: ", err)
+	}
+}
+
+func PostAccessLog(log *AccessLog) {
+	if Logger == nil || log == nil {
+		return
+	}
+	if err := Logger.Post(tag, *log); err != nil {
+		glog.Warning("Error while posting log: ", err)
+	}
+}

weed/s3api/s3err/error_handler.go

@@ -25,6 +25,7 @@ func WriteXMLResponse(w http.ResponseWriter, r *http.Request, statusCode int, re
 
 func WriteEmptyResponse(w http.ResponseWriter, r *http.Request, statusCode int) {
 	WriteResponse(w, r, statusCode, []byte{}, mimeNone)
+	PostLog(r, statusCode, ErrNone)
 }
 
 func WriteErrorResponse(w http.ResponseWriter, r *http.Request, errorCode ErrorCode) {
@@ -39,6 +40,7 @@ func WriteErrorResponse(w http.ResponseWriter, r *http.Request, errorCode ErrorC
 	errorResponse := getRESTErrorResponse(apiError, r.URL.Path, bucket, object)
 	encodedErrorResponse := EncodeXMLResponse(errorResponse)
 	WriteResponse(w, r, apiError.HTTPStatusCode, encodedErrorResponse, MimeXML)
+	PostLog(r, apiError.HTTPStatusCode, errorCode)
 }
 
 func getRESTErrorResponse(err APIError, resource string, bucket, object string) RESTErrorResponse {