groupware: add response peeking

pkg/jmap/http.go

@@ -57,6 +57,8 @@ const (
 	logDuration        = "duration"
 	logBodyTruncated   = "truncated"
 	logAuthenticatorId = "auth-id"
+
+	responseBodyPeekSize = 2 * 1024
 )
 
 // Record JMAP HTTP execution events that may occur, e.g. using metrics.
@@ -207,32 +209,46 @@ func (h *HttpJmapClient) beforeRequest(_ context.Context, logger *log.Logger, _
 	return nil
 }
 
-type httpResponseReadCloser struct {
+type readCloser struct {
 	io.Reader
-	body io.ReadCloser
+	closer io.ReadCloser
 }
 
-func (c httpResponseReadCloser) Close() error {
-	if c.body != nil {
-		return c.body.Close()
+func (c readCloser) Close() error {
+	if c.closer != nil {
+		return c.closer.Close()
 	}
 	return nil
 }
 
-func (h *HttpJmapClient) response(_ context.Context, logger *log.Logger, _ string, endpoint string, duration time.Duration, req *http.Request, resp *http.Response) (io.ReadCloser, error) {
+func (h *HttpJmapClient) response(peekSize int64, _ context.Context, logger *log.Logger, _ string, endpoint string, duration time.Duration, req *http.Request, resp *http.Response) (io.ReadCloser, []byte, bool, error) {
+	whole := resp.Body
+	peek := []byte{}
+	truncated := false
+	var err error
+
+	if peekSize > 0 {
+		whole, peek, truncated, err = peekResponse(resp.Body, peekSize, func(err error) { logger.Error().Err(err).Msg("failed to close response body") }) //NOSONAR
+		if err != nil {
+			return whole, peek, truncated, err
+		}
+	}
 	l := logger.Trace()
 	if h.traceResponses && l.Enabled() {
-		return dumpHttpResponse(req, resp, h.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) {
+		body, err := dumpHttpResponse(req, resp, whole, h.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) {
 			l.Str(logProto, logProtoJmap).Str(logType, logTypeResponse).Str(logEndpoint, endpoint).
 				Str(logMethod, method).Str(logUri, uri).
 				Str(logHttpStatus, log.SafeString(resp.Status)).Int(logHttpStatusCode, resp.StatusCode).
 				Dur(logDuration, duration).
 				Msg(content)
-		}, func(err error) {
-			logger.Error().Err(err).Msg("failed to close response body") //NOSONAR
-		})
+		}, func(err error) { logger.Error().Err(err).Msg("failed to close response body") }) //NOSONAR
+		if err != nil {
+			return body, peek, truncated, err
+		} else {
+			return body, peek, truncated, nil
+		}
 	} else {
-		return resp.Body, nil
+		return whole, peek, truncated, nil
 	}
 }
 
@@ -280,7 +296,7 @@ func (h *HttpJmapClient) GetSession(ctx context.Context, sessionUrl *url.URL, us
 	}
 
 	// dump the response regardless of the status code
-	body, err := h.response(ctx, logger, username, endpoint, duration, req, res)
+	body, _, _, err := h.response(responseBodyPeekSize, ctx, logger, username, endpoint, duration, req, res)
 
 	// since we are not returning a stream from this function, we have to close the response body
 	// before leaving the scope of the function
@@ -295,6 +311,7 @@ func (h *HttpJmapClient) GetSession(ctx context.Context, sessionUrl *url.URL, us
 		logger.Error().Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).Msg("HTTP response status code is not 200")
 		return SessionResponse{}, jmapError(fmt.Errorf("JMAP API response status is %v", res.Status), JmapErrorServerResponse)
 	}
+
 	h.listener.OnSuccessfulRequest(endpoint, Operation("GetSession"), res.StatusCode)
 
 	if err != nil {
@@ -366,7 +383,7 @@ func (h *HttpJmapClient) Command(operation Operation, request Request, ctx Conte
 	// caller won't be able to read the response; instead, it is up to the caller to close the
 	// ReadCloser we are returning from here
 
-	body, err := h.response(cotx, logger, session.Username, endpoint, duration, req, res)
+	body, _, _, err := h.response(responseBodyPeekSize, cotx, logger, session.Username, endpoint, duration, req, res)
 	language := Language(res.Header.Get("Content-Language")) //NOSONAR
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to read response body")
@@ -377,7 +394,12 @@ func (h *HttpJmapClient) Command(operation Operation, request Request, ctx Conte
 	if res.StatusCode < 200 || res.StatusCode > 299 {
 		h.listener.OnFailedRequestWithStatus(endpoint, operation, res.StatusCode)
 		logger.Error().Str(logEndpoint, endpoint).Str(logHttpStatus, log.SafeString(res.Status)).Msg("HTTP response status code is not 2xx") //NOSONAR
-		return nil, language, jmapError(fmt.Errorf("JMAP server responsed with '%s'", res.Status), JmapErrorServerResponse)
+		switch res.StatusCode {
+		case http.StatusUnauthorized:
+			return nil, language, jmapError(fmt.Errorf("JMAP server responsed with '%s'", res.Status), JmapErrorAuthenticationFailed)
+		default:
+			return nil, language, jmapError(fmt.Errorf("JMAP server responsed with '%s'", res.Status), JmapErrorServerResponse)
+		}
 	}
 
 	h.listener.OnSuccessfulRequest(endpoint, operation, res.StatusCode)
@@ -425,7 +447,7 @@ func (h *HttpJmapClient) UploadBinary(uploadUrl string, operation Operation, end
 	// caller won't be able to read the response; instead, it is up to the caller to close the
 	// ReadCloser we are returning from here
 
-	responseBody, err := h.response(cotx, logger, session.Username, endpoint, duration, req, res)
+	responseBody, _, _, err := h.response(responseBodyPeekSize, cotx, logger, session.Username, endpoint, duration, req, res)
 
 	// since we are not returning a stream from this function, we have to close the response body
 	// before leaving the scope of the function
@@ -497,7 +519,7 @@ func (h *HttpJmapClient) DownloadBinary(downloadUrl string, operation Operation,
 	// caller won't be able to read the response; instead, it is up to the caller to close the
 	// ReadCloser we are returning from here
 
-	responseBody, err := h.response(cotx, logger, session.Username, endpoint, duration, req, res)
+	responseBody, _, _, err := h.response(responseBodyPeekSize, cotx, logger, session.Username, endpoint, duration, req, res)
 
 	language := Language(res.Header.Get("Content-Language"))
 	if err != nil {
@@ -634,7 +656,7 @@ func (w *HttpWsClientFactory) connect(operation Operation, ctx context.Context,
 	{
 		l := logger.Trace()
 		if w.traceWsResponses && l.Enabled() {
-			body, err = dumpHttpResponse(nil, res, w.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) {
+			body, err = dumpHttpResponse(nil, res, res.Body, w.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) {
 				l.Str(logProto, logProtoJmap).Str(logType, logTypeResponse).Str(logEndpoint, endpoint).
 					Str(logMethod, method).Str(logUri, uri).
 					Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).

pkg/jmap/http_test.go

@@ -0,0 +1,146 @@
+package jmap
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPeekResponse(t *testing.T) {
+	require := require.New(t)
+	n := 6
+	text := "hello, world"
+
+	require.Less(n, len(text))
+	body, peek, truncated, err := peekResponse(io.NopCloser(strings.NewReader(text)), int64(n), func(err error) {
+		require.NoError(err)
+	})
+	require.NoError(err)
+	require.True(truncated)
+	require.Equal(text[:n], string(peek))
+	buf := new(strings.Builder)
+	_, err = io.Copy(buf, body)
+	require.NoError(err)
+	require.Equal(text, buf.String())
+}
+
+func TestDumpHttpResponse(t *testing.T) {
+	require := require.New(t)
+	n := 6
+	text := "hello, world"
+
+	resp := &http.Response{
+		Status:     "200 OK",
+		StatusCode: 200,
+		Header:     http.Header{},
+		Body:       nil,
+	}
+
+	body, err := dumpHttpResponse(nil, resp, io.NopCloser(strings.NewReader(text)), int64(n), func(method, uri, content string, truncated bool) {
+		require.True(truncated)
+		require.Equal(fmt.Sprintf("200 OK\n%s", text[:n]), content)
+	}, func(err error) {
+		require.NoError(err)
+	})
+	require.NoError(err)
+	buf := new(strings.Builder)
+	_, err = io.Copy(buf, body)
+	require.NoError(err)
+	require.Equal(text, buf.String())
+}
+
+type testNopReadCloser struct {
+	io.Reader
+	closed bool
+}
+
+func (c *testNopReadCloser) Close() error {
+	c.closed = true
+	return nil
+}
+
+func TestResponse(t *testing.T) {
+	text := "hello, world"
+
+	logger := log.NewLogger(log.Level("trace"))
+
+	for _, tt := range []struct {
+		name               string
+		peekSize           int
+		expectedTruncation bool
+		expectedPeek       string
+	}{
+		{"without peek", 0, false, ""},
+		{"with truncated peek", 4, true, "hell"},
+		{"with full peek", 1024, false, text},
+	} {
+		for _, tr := range []struct {
+			name      string
+			traceSize int
+		}{
+			{"with tracing", 256},
+			{"without tracing", 0},
+		} {
+			for _, b := range []struct {
+				name     string
+				withBody bool
+			}{
+				{"with a body", true},
+				{"without a body", false},
+			} {
+				t.Run(fmt.Sprintf("%s %s %s %s", t.Name(), tt.name, tr.name, b.name), func(t *testing.T) {
+					require := require.New(t)
+
+					var closer *testNopReadCloser
+					var source io.ReadCloser
+					if b.withBody {
+						closer = &testNopReadCloser{closed: false, Reader: strings.NewReader(text)}
+						source = closer
+					} else {
+						closer = nil
+						source = http.NoBody
+					}
+
+					resp := &http.Response{
+						Status:     "200 OK",
+						StatusCode: 200,
+						Header:     http.Header{},
+						Body:       source,
+					}
+
+					client := HttpJmapClient{
+						traceResponses:           true,
+						traceMaxResponseBodySize: int64(tr.traceSize),
+					}
+					body, peek, truncated, err := client.response(int64(tt.peekSize), context.TODO(), &logger, "username", "https://stalwart", time.Duration(1*time.Second), nil, resp)
+					require.NoError(err)
+
+					if b.withBody {
+						require.Equal(tt.expectedPeek, string(peek))
+						require.Equal(tt.expectedTruncation, truncated)
+						buf := new(strings.Builder)
+						_, err = io.Copy(buf, body)
+						require.NoError(err)
+						require.Equal(text, buf.String())
+					} else {
+						require.Empty(peek)
+						require.False(truncated)
+					}
+
+					body.Close()
+
+					if closer != nil {
+						require.True(closer.closed)
+					}
+				})
+			}
+		}
+	}
+}

pkg/jmap/tools.go

@@ -483,10 +483,71 @@ func dumpHttpRequest(req *http.Request, maxBodySize int64, closure func(method s
 	return nil
 }
 
-func dumpHttpResponse(req *http.Request, resp *http.Response, maxBodySize int64,
+type nullReader struct {
+}
+
+func (r nullReader) Read(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (r nullReader) Close() error {
+	return nil
+}
+
+var _ io.ReadCloser = nullReader{}
+
+func multiReader(peekBuffer *bytes.Buffer, body io.Reader) io.Reader {
+	readers := []io.Reader{}
+	if peekBuffer != nil {
+		readers = append(readers, peekBuffer)
+	}
+	if body != nil && body != http.NoBody {
+		readers = append(readers, body)
+	}
+	switch len(readers) {
+	case 0:
+		return http.NoBody
+	case 1:
+		return readers[0]
+	default:
+		return io.MultiReader(readers...)
+	}
+}
+
+func peekResponse(body io.ReadCloser, size int64,
+	closeErrorClosure func(err error)) (io.ReadCloser, []byte, bool, error) {
+	var peekBuffer *bytes.Buffer = nil
+	truncated := false
+	peek := []byte{}
+	if body != nil && body != http.NoBody && size > 0 {
+		peekBuffer = new(bytes.Buffer)
+		limitedReader := io.LimitReader(body, size)
+		tee := io.TeeReader(limitedReader, peekBuffer)
+		if _, err := io.Copy(io.Discard, tee); err != nil {
+			if err := body.Close(); err != nil {
+				closeErrorClosure(err)
+			}
+			return nil, peek, false, err
+		}
+
+		if int64(peekBuffer.Len()) >= size {
+			truncated = true
+		}
+
+		peek = peekBuffer.Bytes()
+	}
+
+	return readCloser{
+		closer: body,
+		Reader: multiReader(peekBuffer, body),
+	}, peek, truncated, nil
+}
+
+func dumpHttpResponse(req *http.Request, resp *http.Response, body io.ReadCloser, maxBodySize int64,
 	closure func(method string, uri string, content string, truncated bool),
 	closeErrorClosure func(err error)) (io.ReadCloser, error) {
 	var logBuilder bytes.Buffer
+	fmt.Fprintf(&logBuilder, "%s\n", resp.Status)
 	for key, values := range resp.Header {
 		for _, value := range values {
 			fmt.Fprintf(&logBuilder, "%s: %s\n", key, value)
@@ -495,15 +556,13 @@ func dumpHttpResponse(req *http.Request, resp *http.Response, maxBodySize int64,
 
 	var peekBuffer *bytes.Buffer = nil
 	truncated := false
-	if maxBodySize > 0 {
+	if body != nil && body != http.NoBody && maxBodySize > 0 {
 		peekBuffer = new(bytes.Buffer)
-		limitedReader := io.LimitReader(resp.Body, maxBodySize)
+		limitedReader := io.LimitReader(body, maxBodySize)
 		tee := io.TeeReader(limitedReader, peekBuffer)
 		if _, err := io.Copy(io.Discard, tee); err != nil {
-			if resp.Body != nil {
-				if err := resp.Body.Close(); err != nil {
-					closeErrorClosure(err)
-				}
+			if err := body.Close(); err != nil {
+				closeErrorClosure(err)
 			}
 			return nil, err
 		}
@@ -522,11 +581,11 @@ func dumpHttpResponse(req *http.Request, resp *http.Response, maxBodySize int64,
 	}
 
 	if peekBuffer != nil {
-		return httpResponseReadCloser{
-			body:   resp.Body,
-			Reader: io.MultiReader(peekBuffer, resp.Body),
+		return readCloser{
+			closer: body,
+			Reader: multiReader(peekBuffer, body),
 		}, nil
 	} else {
-		return resp.Body, nil
+		return body, nil
 	}
 }

services/groupware/pkg/groupware/api_index_test.go

@@ -1,10 +1,10 @@
 package groupware
 
 import (
-	"encoding/json"
-	"net/http"
 	"testing"
 
+	"github.com/opencloud-eu/opencloud/pkg/jmap"
+	"github.com/opencloud-eu/opencloud/pkg/structs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -12,22 +12,16 @@ func TestGroupwareIndex(t *testing.T) {
 	require := require.New(t)
 	g, err := newGroupwareTest(t)
 	require.NoError(err)
-	require.NotEmpty(g.Users)
 
-	client := &http.Client{}
-	for _, u := range g.Users {
-		req, err := http.NewRequest(http.MethodGet, g.BaseURL, nil)
-		require.NoError(err)
-		req.SetBasicAuth(u.Name, u.Password)
-		resp, err := client.Do(req)
-		require.NoError(err)
-		require.Equal(200, resp.StatusCode)
-		defer resp.Body.Close()
-		index := IndexResponse{}
-		err = json.NewDecoder(resp.Body).Decode(&index)
-		require.NoError(err)
-		require.Len(index.Accounts, 1)
-		require.Equal(u.Email, index.Accounts[0].Name)
-		require.Len(index.Accounts[0].Identities, 2) // email + alias
-	}
+	index := IndexResponse{}
+	u := gget(g, "/", &index)
+	require.Len(index.Accounts, 1)
+	require.Equal(u.Email, index.Accounts[0].Name)
+	require.Len(index.Accounts[0].Identities, 2) // email + alias
+	require.Len(structs.Filter(index.Accounts[0].Identities, func(i jmap.Identity) bool {
+		return i.Email == u.Email
+	}), 1)
+	require.Len(structs.Filter(index.Accounts[0].Identities, func(i jmap.Identity) bool {
+		return i.Email == u.Alias
+	}), 1)
 }

services/groupware/pkg/groupware/groupware_integration_test.go

@@ -1,6 +1,7 @@
 package groupware
 
 import (
+	"encoding/json"
 	"fmt"
 	"math/rand"
 	"net"
@@ -12,6 +13,8 @@ import (
 
 	userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	"github.com/go-chi/chi/v5"
+	"github.com/go-chi/chi/v5/middleware"
+	"github.com/google/uuid"
 	"github.com/opencloud-eu/opencloud/pkg/cors"
 	"github.com/opencloud-eu/opencloud/pkg/jmaptest"
 	opencloudmiddleware "github.com/opencloud-eu/opencloud/pkg/middleware"
@@ -30,10 +33,31 @@ func init() {
 }
 
 type GroupwareTest struct {
+	t       *testing.T
 	BaseURL string
 	Users   []jmaptest.User
 }
 
+func gget[T any](g GroupwareTest, path string, result *T) jmaptest.User {
+	u, err := url.JoinPath(g.BaseURL, path)
+	require.NoError(g.t, err)
+	req, err := http.NewRequest(http.MethodGet, u, nil)
+	require.NoError(g.t, err)
+	user := g.Users[rand.Intn(len(g.Users))] // pick a user at random
+	req.SetBasicAuth(user.Name, user.Password)
+	rid := uuid.New().String()
+	req.Header.Add("X-Request-Id", rid)
+	req.Header.Add("Trace-Id", rid)
+	client := http.Client{}
+	resp, err := client.Do(req)
+	require.NoError(g.t, err)
+	require.Equal(g.t, 200, resp.StatusCode)
+	defer resp.Body.Close()
+	err = json.NewDecoder(resp.Body).Decode(result)
+	require.NoError(g.t, err)
+	return user
+}
+
 func newGroupwareTest(t *testing.T) (GroupwareTest, error) {
 	require := require.New(t)
 	s, err := jmaptest.NewStalwartTest(t)
@@ -136,6 +160,7 @@ func newGroupwareTest(t *testing.T) (GroupwareTest, error) {
 	}
 
 	m.Use(
+		middleware.RequestID,
 		opencloudmiddleware.Cors(
 			cors.Logger(logger),
 			cors.AllowedOrigins(config.HTTP.CORS.AllowedOrigins),
@@ -166,6 +191,7 @@ func newGroupwareTest(t *testing.T) (GroupwareTest, error) {
 	}
 
 	return GroupwareTest{
+		t:       t,
 		BaseURL: baseURL,
 		Users:   s.Users,
 	}, nil