versitygw/s3api/server.go

// Copyright 2023 Versity Software
// This file is 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 s3api

import (
	"errors"
	"fmt"
	"net"
	"net/http"
	"strings"
	"time"

	"github.com/gofiber/fiber/v2"
	"github.com/gofiber/fiber/v2/middleware/logger"
	"github.com/gofiber/fiber/v2/middleware/recover"
	"github.com/versity/versitygw/auth"
	"github.com/versity/versitygw/backend"
	"github.com/versity/versitygw/debuglogger"
	"github.com/versity/versitygw/metrics"
	"github.com/versity/versitygw/s3api/controllers"
	"github.com/versity/versitygw/s3api/middlewares"
	"github.com/versity/versitygw/s3api/utils"
	"github.com/versity/versitygw/s3err"
	"github.com/versity/versitygw/s3event"
	"github.com/versity/versitygw/s3log"
)

const (
	shutDownDuration = time.Second * 10
)

type S3ApiServer struct {
	Router          *S3ApiRouter
	app             *fiber.App
	backend         backend.Backend
	CertStorage     *utils.CertStorage
	quiet           bool
	readonly        bool
	keepAlive       bool
	health          string
	virtualDomain   string
	corsAllowOrigin string
	maxConnections  int
	maxRequests     int
}

func New(
	be backend.Backend,
	root middlewares.RootUserConfig,
	region string,
	iam auth.IAMService,
	l s3log.AuditLogger,
	adminLogger s3log.AuditLogger,
	evs s3event.S3EventSender,
	mm metrics.Manager,
	opts ...Option,
) (*S3ApiServer, error) {
	server := &S3ApiServer{
		backend: be,
		Router:  new(S3ApiRouter),
	}

	for _, opt := range opts {
		opt(server)
	}

	app := fiber.New(fiber.Config{
		AppName:               "versitygw",
		ServerHeader:          "VERSITYGW",
		StreamRequestBody:     true,
		DisableKeepalive:      !server.keepAlive,
		Network:               fiber.NetworkTCP,
		DisableStartupMessage: true,
		ErrorHandler:          globalErrorHandler,
		Concurrency:           server.maxConnections,
	})

	server.app = app

	// initialize the panic recovery middleware
	app.Use(recover.New(
		recover.Config{
			EnableStackTrace:  true,
			StackTraceHandler: stackTraceHandler,
		}))

	// Logging middlewares
	if !server.quiet {
		app.Use(logger.New(logger.Config{
			Format: "${time} | vgw | ${status} | ${latency} | ${ip} | ${method} | ${path} | ${error} | ${queryParams}\n",
		}))
	}
	// Set up health endpoint if specified
	if server.health != "" {
		app.Get(server.health, func(ctx *fiber.Ctx) error {
			return ctx.SendStatus(http.StatusOK)
		})
	}

	// initialize total requests cap limiter middleware
	app.Use(middlewares.RateLimiter(server.maxRequests, mm, l))

	// initilaze the default value setter middleware
	app.Use(middlewares.SetDefaultValues(root, region))

	// initialize the 'DecodeURL' middleware which
	// path unescapes the url
	app.Use(controllers.WrapMiddleware(middlewares.DecodeURL, l, mm))

	// initialize host-style parser in virtual domain is specified
	if server.virtualDomain != "" {
		app.Use(middlewares.HostStyleParser(server.virtualDomain))
	}

	// initialize the debug logger in debug mode
	if debuglogger.IsDebugEnabled() {
		app.Use(middlewares.DebugLogger())
	}

	server.Router.Init(app, be, iam, l, adminLogger, evs, mm, server.readonly, region, server.virtualDomain, root, server.corsAllowOrigin)

	return server, nil
}

// Option sets various options for New()
type Option func(*S3ApiServer)

// WithTLS sets TLS Credentials
func WithTLS(cs *utils.CertStorage) Option {
	return func(s *S3ApiServer) { s.CertStorage = cs }
}

// WithAdminServer runs admin endpoints with the gateway in the same network
func WithAdminServer() Option {
	return func(s *S3ApiServer) { s.Router.WithAdmSrv = true }
}

// WithQuiet silences default logging output
func WithQuiet() Option {
	return func(s *S3ApiServer) { s.quiet = true }
}

// WithHealth sets up a GET health endpoint
func WithHealth(health string) Option {
	return func(s *S3ApiServer) { s.health = health }
}

func WithReadOnly() Option {
	return func(s *S3ApiServer) { s.readonly = true }
}

// WithHostStyle enabled host-style bucket addressing on the server
func WithHostStyle(virtualDomain string) Option {
	return func(s *S3ApiServer) { s.virtualDomain = virtualDomain }
}

// WithKeepAlive enables the server keep alive
func WithKeepAlive() Option {
	return func(s *S3ApiServer) { s.keepAlive = true }
}

// WithCORSAllowOrigin sets the default CORS Access-Control-Allow-Origin value.
// This is applied when no bucket CORS configuration exists, and for admin APIs.
func WithCORSAllowOrigin(origin string) Option {
	return func(s *S3ApiServer) { s.corsAllowOrigin = origin }
}

// WithConcurrencyLimiter sets the server's maximum connection limit
// and the hard limit for in-flight requests.
func WithConcurrencyLimiter(maxConnections, maxRequests int) Option {
	return func(s *S3ApiServer) {
		s.maxConnections = maxConnections
		s.maxRequests = maxRequests
	}
}

// ServeMultiPort creates listeners for multiple port specifications and serves
// on all of them simultaneously. This supports listening on multiple ports and/or
// addresses (e.g., [":7070", "localhost:8080", "0.0.0.0:9090"]).
func (sa *S3ApiServer) ServeMultiPort(ports []string) error {
	if len(ports) == 0 {
		return fmt.Errorf("no ports specified")
	}

	// Multiple ports - create listeners for each
	var listeners []net.Listener

	for _, portSpec := range ports {
		var ln net.Listener
		var err error

		if sa.CertStorage != nil {
			ln, err = utils.NewMultiAddrTLSListener(sa.app.Config().Network, portSpec, sa.CertStorage.GetCertificate)
		} else {
			ln, err = utils.NewMultiAddrListener(sa.app.Config().Network, portSpec)
		}
		if err != nil {
			return fmt.Errorf("failed to bind s3 listener %s: %w", portSpec, err)
		}

		listeners = append(listeners, ln)
	}

	if len(listeners) == 0 {
		return fmt.Errorf("failed to create any s3 listeners")
	}

	// Combine all listeners
	finalListener := utils.NewMultiListener(listeners...)

	return sa.app.Listener(finalListener)
}

// ShutDown gracefully shuts down the server with a context timeout
func (sa *S3ApiServer) ShutDown() error {
	return sa.app.ShutdownWithTimeout(shutDownDuration)
}

// stackTraceHandler stores the system panics
// in the context locals
func stackTraceHandler(ctx *fiber.Ctx, e any) {
	utils.ContextKeyStack.Set(ctx, e)
}

// globalErrorHandler catches the errors before reaching to
// the handlers and any system panics
func globalErrorHandler(ctx *fiber.Ctx, er error) error {
	// set content type to application/xml
	ctx.Response().Header.SetContentType(fiber.MIMEApplicationXML)

	if utils.ContextKeyStack.IsSet(ctx) {
		// if stack is set, it means the stack trace
		// has caught a panic
		// log it as a panic log
		debuglogger.Panic(er)
	} else {
		// handle the fiber specific errors
		var fiberErr *fiber.Error
		if errors.As(er, &fiberErr) {
			if strings.Contains(fiberErr.Message, "cannot parse Content-Length") {
				ctx.Status(http.StatusBadRequest)
				return nil
			}
			if strings.Contains(fiberErr.Message, "error when reading request headers") {
				// This error means fiber failed to parse the incoming request
				// which is a malfoedmed one. Return a BadRequest in this case
				err := s3err.GetAPIError(s3err.ErrCannotParseHTTPRequest)
				ctx.Status(err.HTTPStatusCode)
				return ctx.Send(s3err.GetAPIErrorResponse(err, "", "", ""))
			}
		}

		// additionally log the internal error
		debuglogger.InternalError(er)
	}

	ctx.Status(http.StatusInternalServerError)

	return ctx.Send(s3err.GetAPIErrorResponse(
		s3err.GetAPIError(s3err.ErrInternalError), "", "", ""))
}