cubefs/objectnode/audit.go

// Copyright 2023 The CubeFS 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 objectnode

import (
	"encoding/json"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/cubefs/cubefs/blobstore/common/rpc/auditlog"
	"github.com/cubefs/cubefs/util/log"
)

const (
	auditVersion = "1.0"
)

type AuditLogger interface {
	Name() string
	Send(data []byte) error
	Close() error
}

type AuditEntry struct {
	Version   string    `json:"Version"`
	Time      time.Time `json:"Time"`
	RequestID string    `json:"RequestID"`
	Requester string    `json:"Requester"`
	AccessKey string    `json:"AccessKey"`
	Owner     string    `json:"Owner"`

	Request struct {
		API      string            `json:"API,omitempty"`
		Bucket   string            `json:"Bucket,omitempty"`
		Object   string            `json:"Object,omitempty"`
		Method   string            `json:"Method,omitempty"`
		Proto    string            `json:"Proto,omitempty"`
		Path     string            `json:"Path,omitempty"`
		Query    map[string]string `json:"Query,omitempty"`
		Header   map[string]string `json:"Header,omitempty"`
		Host     string            `json:"Host,omitempty"`
		RemoteIP string            `json:"RemoteIP,omitempty"`
	} `json:"Request"`

	Response struct {
		Status     string            `json:"Status,omitempty"`
		StatusCode int               `json:"StatusCode,omitempty"`
		Header     map[string]string `json:"Header,omitempty"`
		Error      string            `json:"Error,omitempty"`
	} `json:"Response"`

	BytesRequest  int64  `json:"BytesRequest,omitempty"`
	BytesResponse int64  `json:"BytesResponse,omitempty"`
	Duration      string `json:"Duration,omitempty"`
	DurationNS    string `json:"DurationNS,omitempty"`
}

type AuditLogConfig struct {
	Local *auditlog.Config `json:"local,omitempty"`
	// The key of map is a unique identifier of audit
	Kafka   map[string]KafkaAuditConfig   `json:"kafka,omitempty"`
	Webhook map[string]WebhookAuditConfig `json:"webhook,omitempty"`
}

type ExternalAudit struct {
	loggers []AuditLogger

	sync.RWMutex
}

func NewExternalAudit() *ExternalAudit {
	return &ExternalAudit{}
}

func (a *ExternalAudit) AddLoggers(al ...AuditLogger) {
	a.Lock()
	defer a.Unlock()

	a.loggers = append(a.loggers, al...)
}

func (a *ExternalAudit) Close() {
	a.Lock()
	loggers := a.loggers
	a.loggers = nil
	a.Unlock()

	for _, logger := range loggers {
		if logger != nil {
			logger.Close()
		}
	}
}

func (a *ExternalAudit) getLoggers() []AuditLogger {
	a.RLock()
	defer a.RUnlock()

	return a.loggers
}

func (a *ExternalAudit) Logger(w http.ResponseWriter, r *http.Request) {
	loggers := a.getLoggers()
	if len(loggers) <= 0 || w == nil || r == nil {
		return
	}

	var (
		statusCode int
		respBytes  int64
		timeCost   time.Duration
	)
	startTime := time.Now().UTC()
	if rs, ok := w.(*ResponseStater); ok {
		statusCode = rs.StatusCode
		respBytes = rs.Written
		startTime = rs.StartTime
		timeCost = time.Now().UTC().Sub(startTime)
	}
	duration := strconv.FormatInt(timeCost.Nanoseconds(), 10)
	param := ParseRequestParam(r)
	entry := AuditEntry{
		Version:       auditVersion,
		Time:          startTime,
		RequestID:     param.RequestID(),
		Requester:     param.Requester(),
		AccessKey:     param.AccessKey(),
		Owner:         param.Owner(),
		BytesRequest:  r.ContentLength,
		BytesResponse: respBytes,
		DurationNS:    duration,
		Duration:      duration + "ns",
	}

	entry.Request.API = param.API()
	entry.Request.Bucket = param.Bucket()
	entry.Request.Object = param.Object()
	entry.Request.Method = r.Method
	entry.Request.Proto = r.Proto
	entry.Request.Path = r.URL.Path
	entry.Request.Host = r.Host
	entry.Request.RemoteIP = getRequestIP(r)
	query := r.URL.Query()
	reqQuery := make(map[string]string, len(query))
	for k, v := range query {
		reqQuery[k] = strings.Join(v, ",")
	}
	entry.Request.Query = reqQuery
	header := make(map[string]string, len(r.Header))
	for k, v := range r.Header {
		header[k] = strings.Join(v, ",")
	}
	entry.Request.Header = header

	rh := w.Header()
	header = make(map[string]string, len(rh))
	for k, v := range rh {
		header[k] = strings.Join(v, ",")
	}
	entry.Response.Header = header
	entry.Response.StatusCode = statusCode
	entry.Response.Status = http.StatusText(statusCode)
	entry.Response.Error = getResponseErrorMessage(r)

	data, err := json.Marshal(entry)
	if err != nil {
		log.LogErrorf("audit entry json marshal failed: %v", err)
		return
	}

	for _, logger := range loggers {
		if logger != nil {
			go func(logger AuditLogger) {
				if err = logger.Send(data); err != nil {
					log.LogErrorf("send to external '%s' failed: %v", logger.Name(), err)
				}
			}(logger)
		}
	}
}