Merge branch 'sharding' of https://github.com/allegro/akubra into con…

…tent-length-to-synclog

Conflicts:
	httphandler/httphandler.go
	httphandler/log.go
	httphandler/response_merger.go

README.md

@@ -102,6 +102,8 @@ AdditionalRequestHeaders:
 ConnectionTimeout: "3s"
 # Dial timeout on outgoing connections
 ConnectionDialTimeout: "1s"
+# Maximum accepted body size
+BodyMaxSize: "100M"
 # Backend in maintenance mode. Akubra will skip this endpoint
 
 # MaintainedBackends:

VERSION

@@ -1 +1 @@
-0.1.0-dev
+0.1.1b-dev

config/config.go

@@ -27,6 +27,8 @@ type YamlConfig struct {
 	ConnectionTimeout string `yaml:"ConnectionTimeout,omitempty"`
 	// Dial timeout on outgoing connections
 	ConnectionDialTimeout string `yaml:"ConnectionDialTimeout,omitempty"`
+	// Maximum accepted body size
+	BodyMaxSize string `yaml:"BodyMaxSize,omitempty"`
 	// Backend in maintenance mode. Akubra will not send data there
 	MaintainedBackends []YAMLURL `yaml:"MaintainedBackends,omitempty"`
 	// List request methods to be logged in synclog in case of backend failure

examples/akubra.config.dist

@@ -17,6 +17,8 @@ AdditionalRequestHeaders:
 ConnectionTimeout: "3s"
 # Dial timeout on outgoing connections
 ConnectionDialTimeout: "1s"
+# Maximum accepted body size
+BodyMaxSize: "100M"
 # Backend in maintenance mode. Akubra will skip this endpoint
 
 # MaintainedBackends:

glide.yaml

@@ -11,6 +11,7 @@ import:
   version: ~2.2.x
 - package: github.com/sirupsen/logrus
   version: ~0.11.x
+- package: github.com/docker/go-units
 testImport:
 - package: github.com/stretchr/testify
   version: ~1.1.x

httphandler/httphandler.go

@@ -4,12 +4,15 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"errors"
 	"io"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/allegro/akubra/config"
 	"github.com/allegro/akubra/log"
+	"github.com/docker/go-units"
 )
 
 // Handler implements http.Handler interface
@@ -18,6 +21,7 @@ type Handler struct {
 	roundTripper http.RoundTripper
 	mainLog      log.Logger
 	accessLog    log.Logger
+	bodyMaxSize  int64
 }
 
 func (h *Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
@@ -28,6 +32,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		randomID = []byte("notrandomid")
 	}
 
+	validationCode := h.validateIncomingRequest(req)
+	if validationCode > 0 {
+		log.Printf("Rejected invalid incoming request from %s, code %d", req.RemoteAddr, validationCode)
+		w.WriteHeader(validationCode)
+		return
+	}
+
 	randomIDStr := hex.EncodeToString(randomID)
 	randomIDContext := context.WithValue(req.Context(), log.ContextreqIDKey, randomIDStr)
 	log.Debugf("Request id %s", randomIDStr)
@@ -62,6 +73,22 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	}()
 }
 
+func (h *Handler) validateIncomingRequest(req *http.Request) int {
+	var contentLength int64
+	contentLengthHeader := req.Header.Get("Content-Length")
+	if contentLengthHeader != "" {
+		var err error
+		contentLength, err = strconv.ParseInt(contentLengthHeader, 10, 64)
+		if err != nil {
+			return http.StatusBadRequest
+		}
+	}
+	if contentLength > h.bodyMaxSize || req.ContentLength > h.bodyMaxSize {
+		return http.StatusRequestEntityTooLarge
+	}
+	return 0
+}
+
 // ConfigureHTTPTransport returns http.Transport with customized dialer,
 // MaxIdleConnsPerHost and DisableKeepAlives
 func ConfigureHTTPTransport(conf config.Config) (*http.Transport, error) {
@@ -93,10 +120,15 @@ func DecorateRoundTripper(conf config.Config, rt http.RoundTripper) http.RoundTr
 
 // NewHandlerWithRoundTripper returns Handler, but will not construct transport.MultiTransport by itself
 func NewHandlerWithRoundTripper(conf config.Config, roundTripper http.RoundTripper) (http.Handler, error) {
+	bodyMaxSize, err := units.FromHumanSize(conf.BodyMaxSize)
+	if err != nil {
+		return nil, errors.New("Unable to parse BodyMaxSize: " + err.Error())
+	}
 	return &Handler{
 		config:       conf,
 		mainLog:      conf.Mainlog,
 		accessLog:    conf.Accesslog,
 		roundTripper: roundTripper,
+		bodyMaxSize:  bodyMaxSize,
 	}, nil
 }

httphandler/httphandler_test.go

@@ -0,0 +1,27 @@
+package httphandler
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestShouldReturnEntityTooLargeCode(t *testing.T) {
+	request := httptest.NewRequest("POST", "http://somepath", nil)
+	request.Header.Set("Content-Length", "4096")
+	handler := &Handler{bodyMaxSize: 1024}
+	writer := httptest.NewRecorder()
+	handler.ServeHTTP(writer, request)
+	assert.Equal(t, http.StatusRequestEntityTooLarge, writer.Code)
+}
+
+func TestShouldReturnBadRequestOnUnparsableContentLengthHeader(t *testing.T) {
+	request := httptest.NewRequest("POST", "http://somepath", nil)
+	request.Header.Set("Content-Length", "strange-content-header")
+	handler := &Handler{bodyMaxSize: 1024}
+	writer := httptest.NewRecorder()
+	handler.ServeHTTP(writer, request)
+	assert.Equal(t, http.StatusBadRequest, writer.Code)
+}

httphandler/log.go

@@ -62,16 +62,17 @@ type SyncLogMessageData struct {
 	Path        string `json:"path"`
 	SuccessHost string `json:"successhost"`
 	UserAgent   string `json:"useragent"`
-	ContentLength string `json:"content-length"`
-	ErrorMsg    string `json:"error"`
-	ReqID       string `json:"reqID"`
-	Time        string `json:"ts"`
+	// ContentLength if negative means no content length header provided
+	ContentLength int64  `json:"content-length"`
+	ErrorMsg      string `json:"error"`
+	ReqID         string `json:"reqID"`
+	Time          string `json:"ts"`
 }
 
 // String produces data in csv format with fields in following order:
 // Method, Host, Path, UserAgent, StatusCode, Duration, RespErr)
 func (slmd SyncLogMessageData) String() string {
-	return fmt.Sprintf("%q, %q, %q, %q, %q, %q",
+	return fmt.Sprintf("%q, %q, %q, %q, %q, %d, %q",
 		slmd.Method,
 		slmd.FailedHost,
 		slmd.Path,
@@ -83,7 +84,7 @@ func (slmd SyncLogMessageData) String() string {
 
 // NewSyncLogMessageData creates new SyncLogMessageData
 func NewSyncLogMessageData(method, failedHost, path, successHost, userAgent,
-	contentLength, reqID, errorMsg string) *SyncLogMessageData {
+	reqID, errorMsg string, contentLength int64) *SyncLogMessageData {
 	ts := time.Now().Format(time.RFC3339Nano)
 	return &SyncLogMessageData{
 		method, failedHost, path, successHost, userAgent,

httphandler/response_merger.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"io"
 	"io/ioutil"
+	"strconv"
 
 	"github.com/allegro/akubra/config"
 	"github.com/allegro/akubra/log"
@@ -36,16 +37,23 @@ func (rd *responseMerger) synclog(r, successfulTup *transport.ReqResErrTuple) {
 	if r.Err != nil {
 		errorMsg = r.Err.Error()
 	}
+
+	contentLengthStr := successfulTup.Res.Header.Get("Content-Length")
+	contentLength, err := strconv.ParseInt(contentLengthStr, 10, 64)
+	if err != nil {
+		contentLength = -1
+	}
+
 	reqID := r.Req.Context().Value(log.ContextreqIDKey).(string)
 	syncLogMsg := NewSyncLogMessageData(
 		r.Req.Method,
 		r.Req.Host,
 		successfulTup.Req.URL.Path,
 		successfulTup.Req.Host,
 		r.Req.Header.Get("User-Agent"),
-		successfulTup.Res.Header.Get("Content-Length"),
 		reqID,
-		errorMsg)
+		errorMsg,
+		contentLength)
 	logMsg, err := json.Marshal(syncLogMsg)
 	if err != nil {
 		return