MM-13606 Remove consumeAndClose and clean up integration response handling (#10066)

* MM-13606 Remove consumeAndClose

* Allow overriding HTTPService's request timeout

* MM-13606 Clean up integration response handling

* Properly close httptest servers

* Address feedback

* Only call buf.Bytes when necessary

* Properly check for errors in doOutgoingWebhookRequest

* Add comment explaining ignored ioutil.ReadAll errors

app/command.go

@@ -5,6 +5,7 @@ package app
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -160,7 +161,6 @@ func (a *App) ExecuteCommand(args *model.CommandArgs) (*model.CommandResponse, *
 	trigger := parts[0][1:]
 	trigger = strings.ToLower(trigger)
 	message := strings.Join(parts[1:], " ")
-	provider := GetCommandProvider(trigger)
 
 	clientTriggerId, triggerId, appErr := model.GenerateTriggerId(args.UserId, a.AsymmetricSigningKey())
 	if appErr != nil {
@@ -169,24 +169,52 @@ func (a *App) ExecuteCommand(args *model.CommandArgs) (*model.CommandResponse, *
 
 	args.TriggerId = triggerId
 
-	if provider != nil {
-		if cmd := provider.GetCommand(a, args.T); cmd != nil {
-			response := provider.DoCommand(a, args, message)
-			return a.HandleCommandResponse(cmd, args, response, true)
-		}
+	cmd, response := a.tryExecuteBuiltInCommand(args, trigger, message)
+	if cmd != nil && response != nil {
+		return a.HandleCommandResponse(cmd, args, response, true)
 	}
 
-	cmd, response, appErr := a.ExecutePluginCommand(args)
+	cmd, response, appErr = a.tryExecutePluginCommand(args)
 	if appErr != nil {
 		return nil, appErr
-	}
-	if cmd != nil {
+	} else if cmd != nil && response != nil {
 		response.TriggerId = clientTriggerId
 		return a.HandleCommandResponse(cmd, args, response, true)
 	}
 
+	cmd, response, appErr = a.tryExecuteCustomCommand(args, trigger, message)
+	if appErr != nil {
+		return nil, appErr
+	} else if cmd != nil && response != nil {
+		response.TriggerId = clientTriggerId
+		return a.HandleCommandResponse(cmd, args, response, false)
+	}
+
+	return nil, model.NewAppError("command", "api.command.execute_command.not_found.app_error", map[string]interface{}{"Trigger": trigger}, "", http.StatusNotFound)
+}
+
+// tryExecutePluginCommand attempts to run a built in command based on the given arguments. If no such command can be
+// found, returns nil for all arguments.
+func (a *App) tryExecuteBuiltInCommand(args *model.CommandArgs, trigger string, message string) (*model.Command, *model.CommandResponse) {
+	provider := GetCommandProvider(trigger)
+	if provider == nil {
+		return nil, nil
+	}
+
+	cmd := provider.GetCommand(a, args.T)
+	if cmd == nil {
+		return nil, nil
+	}
+
+	return cmd, provider.DoCommand(a, args, message)
+}
+
+// tryExecuteCustomCommand attempts to run a custom command based on the given arguments. If no such command can be
+// found, returns nil for all arguments.
+func (a *App) tryExecuteCustomCommand(args *model.CommandArgs, trigger string, message string) (*model.Command, *model.CommandResponse, *model.AppError) {
+	// Handle custom commands
 	if !*a.Config().ServiceSettings.EnableCommands {
-		return nil, model.NewAppError("ExecuteCommand", "api.command.disabled.app_error", nil, "", http.StatusNotImplemented)
+		return nil, nil, model.NewAppError("ExecuteCommand", "api.command.disabled.app_error", nil, "", http.StatusNotImplemented)
 	}
 
 	chanChan := a.Srv.Store.Channel().Get(args.ChannelId, true)
@@ -195,98 +223,121 @@ func (a *App) ExecuteCommand(args *model.CommandArgs) (*model.CommandResponse, *
 
 	result := <-a.Srv.Store.Command().GetByTeam(args.TeamId)
 	if result.Err != nil {
-		return nil, result.Err
+		return nil, nil, result.Err
 	}
 
 	tr := <-teamChan
 	if tr.Err != nil {
-		return nil, tr.Err
+		return nil, nil, tr.Err
 	}
 	team := tr.Data.(*model.Team)
 
 	ur := <-userChan
 	if ur.Err != nil {
-		return nil, ur.Err
+		return nil, nil, ur.Err
 	}
 	user := ur.Data.(*model.User)
 
 	cr := <-chanChan
 	if cr.Err != nil {
-		return nil, cr.Err
+		return nil, nil, cr.Err
 	}
 	channel := cr.Data.(*model.Channel)
 
+	var cmd *model.Command
+
 	teamCmds := result.Data.([]*model.Command)
-	for _, cmd := range teamCmds {
-		if trigger == cmd.Trigger {
-			mlog.Debug(fmt.Sprintf(utils.T("api.command.execute_command.debug"), trigger, args.UserId))
-
-			p := url.Values{}
-			p.Set("token", cmd.Token)
-
-			p.Set("team_id", cmd.TeamId)
-			p.Set("team_domain", team.Name)
-
-			p.Set("channel_id", args.ChannelId)
-			p.Set("channel_name", channel.Name)
-
-			p.Set("user_id", args.UserId)
-			p.Set("user_name", user.Username)
-
-			p.Set("command", "/"+trigger)
-			p.Set("text", message)
-
-			p.Set("trigger_id", triggerId)
-
-			hook, appErr := a.CreateCommandWebhook(cmd.Id, args)
-			if appErr != nil {
-				return nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": trigger}, appErr.Error(), http.StatusInternalServerError)
-			}
-			p.Set("response_url", args.SiteURL+"/hooks/commands/"+hook.Id)
-
-			var req *http.Request
-			if cmd.Method == model.COMMAND_METHOD_GET {
-				req, _ = http.NewRequest(http.MethodGet, cmd.URL, nil)
-
-				if req.URL.RawQuery != "" {
-					req.URL.RawQuery += "&"
-				}
-				req.URL.RawQuery += p.Encode()
-			} else {
-				req, _ = http.NewRequest(http.MethodPost, cmd.URL, strings.NewReader(p.Encode()))
-			}
-
-			req.Header.Set("Accept", "application/json")
-			req.Header.Set("Authorization", "Token "+cmd.Token)
-			if cmd.Method == model.COMMAND_METHOD_POST {
-				req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-			}
-
-			resp, err := a.HTTPService.MakeClient(false).Do(req)
-			if err != nil {
-				return nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": trigger}, err.Error(), http.StatusInternalServerError)
-			}
-			if resp.StatusCode != http.StatusOK {
-				defer resp.Body.Close()
-				body, _ := ioutil.ReadAll(resp.Body)
-				return nil, model.NewAppError("command", "api.command.execute_command.failed_resp.app_error", map[string]interface{}{"Trigger": trigger, "Status": resp.Status}, string(body), http.StatusInternalServerError)
-			}
-
-			response, err := model.CommandResponseFromHTTPBody(resp.Header.Get("Content-Type"), resp.Body)
-			if err != nil {
-				return nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": trigger}, err.Error(), http.StatusInternalServerError)
-			}
-			if response == nil {
-				return nil, model.NewAppError("command", "api.command.execute_command.failed_empty.app_error", map[string]interface{}{"Trigger": trigger}, "", http.StatusInternalServerError)
-			}
-
-			response.TriggerId = clientTriggerId
-
-			return a.HandleCommandResponse(cmd, args, response, false)
+	for _, teamCmd := range teamCmds {
+		if trigger == teamCmd.Trigger {
+			cmd = teamCmd
 		}
 	}
 
-	return nil, model.NewAppError("command", "api.command.execute_command.not_found.app_error", map[string]interface{}{"Trigger": trigger}, "", http.StatusNotFound)
+	if cmd == nil {
+		return nil, nil, nil
+	}
+
+	mlog.Debug(fmt.Sprintf(utils.T("api.command.execute_command.debug"), trigger, args.UserId))
+
+	p := url.Values{}
+	p.Set("token", cmd.Token)
+
+	p.Set("team_id", cmd.TeamId)
+	p.Set("team_domain", team.Name)
+
+	p.Set("channel_id", args.ChannelId)
+	p.Set("channel_name", channel.Name)
+
+	p.Set("user_id", args.UserId)
+	p.Set("user_name", user.Username)
+
+	p.Set("command", "/"+trigger)
+	p.Set("text", message)
+
+	p.Set("trigger_id", args.TriggerId)
+
+	hook, appErr := a.CreateCommandWebhook(cmd.Id, args)
+	if appErr != nil {
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": trigger}, appErr.Error(), http.StatusInternalServerError)
+	}
+	p.Set("response_url", args.SiteURL+"/hooks/commands/"+hook.Id)
+
+	return a.doCommandRequest(cmd, p)
+}
+
+func (a *App) doCommandRequest(cmd *model.Command, p url.Values) (*model.Command, *model.CommandResponse, *model.AppError) {
+	// Prepare the request
+	var req *http.Request
+	var err error
+	if cmd.Method == model.COMMAND_METHOD_GET {
+		req, err = http.NewRequest(http.MethodGet, cmd.URL, nil)
+	} else {
+		req, err = http.NewRequest(http.MethodPost, cmd.URL, strings.NewReader(p.Encode()))
+	}
+
+	if err != nil {
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": cmd.Trigger}, err.Error(), http.StatusInternalServerError)
+	}
+
+	if cmd.Method == model.COMMAND_METHOD_GET {
+		if req.URL.RawQuery != "" {
+			req.URL.RawQuery += "&"
+		}
+		req.URL.RawQuery += p.Encode()
+	}
+
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Authorization", "Token "+cmd.Token)
+	if cmd.Method == model.COMMAND_METHOD_POST {
+		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+
+	// Send the request
+	resp, err := a.HTTPService.MakeClient(false).Do(req)
+	if err != nil {
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": cmd.Trigger}, err.Error(), http.StatusInternalServerError)
+	}
+
+	defer resp.Body.Close()
+
+	// Handle the response
+	body := io.LimitReader(resp.Body, MaxIntegrationResponseSize)
+
+	if resp.StatusCode != http.StatusOK {
+		// Ignore the error below because the resulting string will just be the empty string if bodyBytes is nil
+		bodyBytes, _ := ioutil.ReadAll(body)
+
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed_resp.app_error", map[string]interface{}{"Trigger": cmd.Trigger, "Status": resp.Status}, string(bodyBytes), http.StatusInternalServerError)
+	}
+
+	response, err := model.CommandResponseFromHTTPBody(resp.Header.Get("Content-Type"), body)
+	if err != nil {
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]interface{}{"Trigger": cmd.Trigger}, err.Error(), http.StatusInternalServerError)
+	} else if response == nil {
+		return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed_empty.app_error", map[string]interface{}{"Trigger": cmd.Trigger}, "", http.StatusInternalServerError)
+	}
+
+	return cmd, response, nil
 }
 
 func (a *App) HandleCommandResponse(command *model.Command, args *model.CommandArgs, response *model.CommandResponse, builtIn bool) (*model.CommandResponse, *model.AppError) {

app/command_test.go

@@ -4,11 +4,19 @@
 package app
 
 import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/services/httpservice"
 )
 
 func TestMoveCommand(t *testing.T) {
@@ -257,3 +265,99 @@ func TestHandleCommandResponse(t *testing.T) {
 	_, err = th.App.HandleCommandResponse(command, args, resp, builtIn)
 	assert.Nil(t, err)
 }
+
+func TestDoCommandRequest(t *testing.T) {
+	th := Setup().InitBasic()
+	defer th.TearDown()
+
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		cfg.ServiceSettings.AllowedUntrustedInternalConnections = model.NewString("127.0.0.1")
+		cfg.ServiceSettings.EnableCommands = model.NewBool(true)
+	})
+
+	t.Run("with a valid text response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, strings.NewReader("Hello, World!"))
+		}))
+		defer server.Close()
+
+		_, resp, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.Nil(t, err)
+
+		assert.NotNil(t, resp)
+		assert.Equal(t, "Hello, World!", resp.Text)
+	})
+
+	t.Run("with a valid json response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("Content-Type", "application/json")
+
+			io.Copy(w, strings.NewReader(`{"text": "Hello, World!"}`))
+		}))
+		defer server.Close()
+
+		_, resp, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.Nil(t, err)
+
+		assert.NotNil(t, resp)
+		assert.Equal(t, "Hello, World!", resp.Text)
+	})
+
+	t.Run("with a large text response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, InfiniteReader{})
+		}))
+		defer server.Close()
+
+		// Since we limit the length of the response, no error will be returned and resp.Text will be a finite string
+
+		_, resp, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.Nil(t, err)
+		require.NotNil(t, resp)
+	})
+
+	t.Run("with a large, valid json response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("Content-Type", "application/json")
+
+			io.Copy(w, io.MultiReader(strings.NewReader(`{"text": "`), InfiniteReader{}, strings.NewReader(`"}`)))
+		}))
+		defer server.Close()
+
+		_, _, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.NotNil(t, err)
+		require.Equal(t, "api.command.execute_command.failed.app_error", err.Id)
+	})
+
+	t.Run("with a large, invalid json response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("Content-Type", "application/json")
+
+			io.Copy(w, InfiniteReader{})
+		}))
+		defer server.Close()
+
+		_, _, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.NotNil(t, err)
+		require.Equal(t, "api.command.execute_command.failed.app_error", err.Id)
+	})
+
+	t.Run("with a slow response", func(t *testing.T) {
+		timeout := 100 * time.Millisecond
+
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(timeout + time.Millisecond)
+			io.Copy(w, strings.NewReader(`{"text": "Hello, World!"}`))
+		}))
+		defer server.Close()
+
+		th.App.HTTPService.(*httpservice.HTTPServiceImpl).RequestTimeout = timeout
+		defer func() {
+			th.App.HTTPService.(*httpservice.HTTPServiceImpl).RequestTimeout = httpservice.RequestTimeout
+		}()
+
+		_, _, err := th.App.doCommandRequest(&model.Command{URL: server.URL}, url.Values{})
+		require.NotNil(t, err)
+		require.Equal(t, "api.command.execute_command.failed.app_error", err.Id)
+	})
+}

app/integration_action.go

@@ -71,13 +71,12 @@ func (a *App) DoPostAction(postId, actionId, userId, selectedOption string) (str
 	}
 
 	resp, err := a.DoActionRequest(action.Integration.URL, request.ToJson())
-	if resp != nil {
-		defer consumeAndClose(resp)
-	}
 	if err != nil {
 		return "", err
 	}
 
+	defer resp.Body.Close()
+
 	var response model.PostActionIntegrationResponse
 	if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
 		return "", model.NewAppError("DoPostAction", "api.post.do_action.action_integration.app_error", nil, "err="+err.Error(), http.StatusBadRequest)
@@ -184,14 +183,12 @@ func (a *App) SubmitInteractiveDialog(request model.SubmitDialogRequest) (*model
 	}
 
 	resp, err := a.DoActionRequest(url, b)
-	if resp != nil {
-		defer consumeAndClose(resp)
-	}
-
 	if err != nil {
 		return nil, err
 	}
 
+	defer resp.Body.Close()
+
 	var response model.SubmitDialogResponse
 	if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
 		// Don't fail, an empty response is acceptable

app/notification_push.go

@@ -286,10 +286,9 @@ func (a *App) sendToPushProxy(msg model.PushNotification, session *model.Session
 		return
 	}
 
+	defer resp.Body.Close()
+
 	pushResponse := model.PushResponseFromJson(resp.Body)
-	if resp.Body != nil {
-		consumeAndClose(resp)
-	}
 
 	if pushResponse[model.PUSH_STATUS] == model.PUSH_STATUS_REMOVE {
 		mlog.Info(fmt.Sprintf("Device was reported as removed for UserId=%v SessionId=%v removing push for this session", session.UserId, session.Id), mlog.String("user_id", session.UserId))

app/oauth.go

@@ -731,9 +731,9 @@ func (a *App) AuthorizeOAuthUser(w http.ResponseWriter, r *http.Request, service
 
 	stateProps := model.MapFromJson(strings.NewReader(stateStr))
 
-	expectedToken, err := a.GetOAuthStateToken(stateProps["token"])
-	if err != nil {
-		return nil, "", stateProps, err
+	expectedToken, appErr := a.GetOAuthStateToken(stateProps["token"])
+	if appErr != nil {
+		return nil, "", stateProps, appErr
 	}
 
 	stateEmail := stateProps["email"]
@@ -752,8 +752,9 @@ func (a *App) AuthorizeOAuthUser(w http.ResponseWriter, r *http.Request, service
 		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.invalid_state.app_error", nil, "", http.StatusBadRequest)
 	}
 
-	if err = a.DeleteToken(expectedToken); err != nil {
-		mlog.Error(err.Error())
+	appErr = a.DeleteToken(expectedToken)
+	if appErr != nil {
+		mlog.Error(appErr.Error())
 	}
 
 	httpCookie := &http.Cookie{
@@ -783,30 +784,26 @@ func (a *App) AuthorizeOAuthUser(w http.ResponseWriter, r *http.Request, service
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 	req.Header.Set("Accept", "application/json")
 
-	resp, serviceErr := a.HTTPService.MakeClient(true).Do(req)
-	if serviceErr != nil {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.token_failed.app_error", nil, serviceErr.Error(), http.StatusInternalServerError)
+	resp, err := a.HTTPService.MakeClient(true).Do(req)
+	if err != nil {
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.token_failed.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
+	defer resp.Body.Close()
 
-	bodyBytes, readErr := ioutil.ReadAll(resp.Body)
-	if readErr != nil {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.token_failed.app_error", nil, readErr.Error(), http.StatusInternalServerError)
-	}
-	resp.Body = ioutil.NopCloser(bytes.NewBuffer(bodyBytes))
-
-	ar := model.AccessResponseFromJson(resp.Body)
-	consumeAndClose(resp)
+	var buf bytes.Buffer
+	tee := io.TeeReader(resp.Body, &buf)
+	ar := model.AccessResponseFromJson(tee)
 
 	if ar == nil || resp.StatusCode != http.StatusOK {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.bad_response.app_error", nil, "response_body="+string(bodyBytes), http.StatusInternalServerError)
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.bad_response.app_error", nil, "response_body="+buf.String(), http.StatusInternalServerError)
 	}
 
 	if strings.ToLower(ar.TokenType) != model.ACCESS_TOKEN_TYPE {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.bad_token.app_error", nil, "token_type="+ar.TokenType+", response_body="+string(bodyBytes), http.StatusInternalServerError)
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.bad_token.app_error", nil, "token_type="+ar.TokenType+", response_body="+buf.String(), http.StatusInternalServerError)
 	}
 
 	if len(ar.AccessToken) == 0 {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.missing.app_error", nil, "response_body="+string(bodyBytes), http.StatusInternalServerError)
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.missing.app_error", nil, "response_body="+buf.String(), http.StatusInternalServerError)
 	}
 
 	p = url.Values{}
@@ -820,25 +817,27 @@ func (a *App) AuthorizeOAuthUser(w http.ResponseWriter, r *http.Request, service
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("Authorization", "Bearer "+ar.AccessToken)
 
-	resp, serviceErr = a.HTTPService.MakeClient(true).Do(req)
-	if serviceErr != nil {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.service.app_error", map[string]interface{}{"Service": service}, serviceErr.Error(), http.StatusInternalServerError)
-	}
+	resp, err = a.HTTPService.MakeClient(true).Do(req)
+	if err != nil {
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.service.app_error", map[string]interface{}{"Service": service}, err.Error(), http.StatusInternalServerError)
+	} else if resp.StatusCode != http.StatusOK {
+		defer resp.Body.Close()
 
-	bodyBytes, readErr = ioutil.ReadAll(resp.Body)
-	if readErr != nil {
-		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.token_failed.app_error", nil, readErr.Error(), http.StatusInternalServerError)
-	}
-
-	if resp.StatusCode != http.StatusOK {
+		// Ignore the error below because the resulting string will just be the empty string if bodyBytes is nil
+		bodyBytes, _ := ioutil.ReadAll(resp.Body)
 		bodyString := string(bodyBytes)
+
 		mlog.Error("Error getting OAuth user: " + bodyString)
+
 		if service == model.SERVICE_GITLAB && resp.StatusCode == http.StatusForbidden && strings.Contains(bodyString, "Terms of Service") {
+			// Return a nicer error when the user hasn't accepted GitLab's terms of service
 			return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "oauth.gitlab.tos.error", nil, "", http.StatusBadRequest)
 		}
+
+		return nil, "", stateProps, model.NewAppError("AuthorizeOAuthUser", "api.user.authorize_oauth_user.response.app_error", nil, "response_body="+bodyString, http.StatusInternalServerError)
 	}
 
-	resp.Body = ioutil.NopCloser(bytes.NewBuffer(bodyBytes))
+	// Note that resp.Body is not closed here, so it must be closed by the caller
 	return resp.Body, teamId, stateProps, nil
 }
 

app/opengraph.go

@@ -20,7 +20,7 @@ func (a *App) GetOpenGraphMetadata(requestURL string) *opengraph.OpenGraph {
 		mlog.Error("GetOpenGraphMetadata request failed", mlog.String("requestURL", requestURL), mlog.Any("err", err))
 		return nil
 	}
-	defer consumeAndClose(res)
+	defer res.Body.Close()
 
 	return a.ParseOpenGraphMetadata(requestURL, res.Body, res.Header.Get("Content-Type"))
 }

app/plugin_commands.go

@@ -90,7 +90,9 @@ func (a *App) PluginCommandsForTeam(teamId string) []*model.Command {
 	return commands
 }
 
-func (a *App) ExecutePluginCommand(args *model.CommandArgs) (*model.Command, *model.CommandResponse, *model.AppError) {
+// tryExecutePluginCommand attempts to run a command provided by a plugin based on the given arguments. If no such
+// command can be found, returns nil for all arguments.
+func (a *App) tryExecutePluginCommand(args *model.CommandArgs) (*model.Command, *model.CommandResponse, *model.AppError) {
 	parts := strings.Split(args.Command, " ")
 	trigger := parts[0][1:]
 	trigger = strings.ToLower(trigger)

app/post_metadata.go

@@ -339,7 +339,8 @@ func (a *App) getLinkMetadata(requestURL string, useCache bool) (*opengraph.Open
 	if err != nil {
 		return nil, nil, err
 	}
-	defer consumeAndClose(res)
+
+	defer res.Body.Close()
 
 	// Parse the data
 	og, image, err := a.parseLinkMetadata(requestURL, res.Body, res.Header.Get("Content-Type"))

app/security_update_check.go

@@ -81,8 +81,9 @@ func (s *Server) DoSecurityUpdateCheck() {
 					return
 				}
 
+				defer res.Body.Close()
+
 				bulletins := model.SecurityBulletinsFromJson(res.Body)
-				consumeAndClose(res)
 
 				for _, bulletin := range bulletins {
 					if bulletin.AppliesToVersion == model.CurrentVersion {

app/server.go

@@ -8,8 +8,6 @@ import (
 	"crypto/ecdsa"
 	"crypto/tls"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"net/url"
@@ -596,14 +594,6 @@ func (a *App) OriginChecker() func(*http.Request) bool {
 	return nil
 }
 
-// This is required to re-use the underlying connection and not take up file descriptors
-func consumeAndClose(r *http.Response) {
-	if r.Body != nil {
-		io.Copy(ioutil.Discard, r.Body)
-		r.Body.Close()
-	}
-}
-
 func runSecurityJob(s *Server) {
 	doSecurity(s)
 	model.CreateRecurringTask("Security", func() {

app/webhook.go

@@ -20,6 +20,8 @@ import (
 const (
 	TRIGGERWORDS_EXACT_MATCH = 0
 	TRIGGERWORDS_STARTS_WITH = 1
+
+	MaxIntegrationResponseSize = 1024 * 1024 // Posts can be <100KB at most, so this is likely more than enough
 )
 
 func (a *App) handleWebhookEvents(post *model.Post, team *model.Team, channel *model.Channel, user *model.User) *model.AppError {
@@ -101,55 +103,70 @@ func (a *App) TriggerWebhook(payload *model.OutgoingWebhookPayload, hook *model.
 		contentType = "application/x-www-form-urlencoded"
 	}
 
-	for _, url := range hook.CallbackURLs {
-		a.Srv.Go(func(url string) func() {
-			return func() {
-				req, _ := http.NewRequest("POST", url, body)
-				req.Header.Set("Content-Type", contentType)
-				req.Header.Set("Accept", "application/json")
-				if resp, err := a.HTTPService.MakeClient(false).Do(req); err != nil {
-					mlog.Error(fmt.Sprintf("Event POST failed, err=%s", err.Error()))
-				} else {
-					defer consumeAndClose(resp)
+	for i := range hook.CallbackURLs {
+		// Get the callback URL by index to properly capture it for the go func
+		url := hook.CallbackURLs[i]
 
-					webhookResp := model.OutgoingWebhookResponseFromJson(resp.Body)
+		a.Srv.Go(func() {
+			webhookResp, err := a.doOutgoingWebhookRequest(url, body, contentType)
+			if err != nil {
+				mlog.Error(fmt.Sprintf("Event POST failed, err=%s", err.Error()))
+				return
+			}
 
-					if webhookResp != nil && (webhookResp.Text != nil || len(webhookResp.Attachments) > 0) {
-						postRootId := ""
-						if webhookResp.ResponseType == model.OUTGOING_HOOK_RESPONSE_TYPE_COMMENT {
-							postRootId = post.Id
-						}
-						if len(webhookResp.Props) == 0 {
-							webhookResp.Props = make(model.StringInterface)
-						}
-						webhookResp.Props["webhook_display_name"] = hook.DisplayName
+			if webhookResp != nil && (webhookResp.Text != nil || len(webhookResp.Attachments) > 0) {
+				postRootId := ""
+				if webhookResp.ResponseType == model.OUTGOING_HOOK_RESPONSE_TYPE_COMMENT {
+					postRootId = post.Id
+				}
+				if len(webhookResp.Props) == 0 {
+					webhookResp.Props = make(model.StringInterface)
+				}
+				webhookResp.Props["webhook_display_name"] = hook.DisplayName
 
-						text := ""
-						if webhookResp.Text != nil {
-							text = a.ProcessSlackText(*webhookResp.Text)
-						}
-						webhookResp.Attachments = a.ProcessSlackAttachments(webhookResp.Attachments)
-						// attachments is in here for slack compatibility
-						if len(webhookResp.Attachments) > 0 {
-							webhookResp.Props["attachments"] = webhookResp.Attachments
-						}
-						if a.Config().ServiceSettings.EnablePostUsernameOverride && hook.Username != "" && webhookResp.Username == "" {
-							webhookResp.Username = hook.Username
-						}
+				text := ""
+				if webhookResp.Text != nil {
+					text = a.ProcessSlackText(*webhookResp.Text)
+				}
+				webhookResp.Attachments = a.ProcessSlackAttachments(webhookResp.Attachments)
+				// attachments is in here for slack compatibility
+				if len(webhookResp.Attachments) > 0 {
+					webhookResp.Props["attachments"] = webhookResp.Attachments
+				}
+				if a.Config().ServiceSettings.EnablePostUsernameOverride && hook.Username != "" && webhookResp.Username == "" {
+					webhookResp.Username = hook.Username
+				}
 
-						if a.Config().ServiceSettings.EnablePostIconOverride && hook.IconURL != "" && webhookResp.IconURL == "" {
-							webhookResp.IconURL = hook.IconURL
-						}
-						if _, err := a.CreateWebhookPost(hook.CreatorId, channel, text, webhookResp.Username, webhookResp.IconURL, webhookResp.Props, webhookResp.Type, postRootId); err != nil {
-							mlog.Error(fmt.Sprintf("Failed to create response post, err=%v", err))
-						}
-					}
+				if a.Config().ServiceSettings.EnablePostIconOverride && hook.IconURL != "" && webhookResp.IconURL == "" {
+					webhookResp.IconURL = hook.IconURL
+				}
+				if _, err := a.CreateWebhookPost(hook.CreatorId, channel, text, webhookResp.Username, webhookResp.IconURL, webhookResp.Props, webhookResp.Type, postRootId); err != nil {
+					mlog.Error(fmt.Sprintf("Failed to create response post, err=%v", err))
 				}
 			}
-		}(url))
+		})
 	}
 }
 
+func (a *App) doOutgoingWebhookRequest(url string, body io.Reader, contentType string) (*model.OutgoingWebhookResponse, error) {
+	req, err := http.NewRequest("POST", url, body)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Content-Type", contentType)
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := a.HTTPService.MakeClient(false).Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	defer resp.Body.Close()
+
+	return model.OutgoingWebhookResponseFromJson(io.LimitReader(resp.Body, MaxIntegrationResponseSize))
+}
+
 func SplitWebhookPost(post *model.Post, maxPostSize int) ([]*model.Post, *model.AppError) {
 	splits := make([]*model.Post, 0)
 	remainingText := post.Message

app/webhook_test.go

@@ -4,17 +4,19 @@
 package app
 
 import (
-	"strings"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
+	"encoding/json"
+	"io"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
 	"time"
 
 	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/services/httpservice"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestCreateIncomingWebhookForChannel(t *testing.T) {
@@ -652,3 +654,91 @@ func TestTriggerOutGoingWebhookWithUsernameAndIconURL(t *testing.T) {
 	}
 
 }
+
+type InfiniteReader struct {
+	Prefix string
+}
+
+func (r InfiniteReader) Read(p []byte) (n int, err error) {
+	for i := range p {
+		p[i] = 'a'
+	}
+
+	return len(p), nil
+}
+
+func TestDoOutgoingWebhookRequest(t *testing.T) {
+	th := Setup().InitBasic()
+	defer th.TearDown()
+
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		cfg.ServiceSettings.AllowedUntrustedInternalConnections = model.NewString("127.0.0.1")
+		cfg.ServiceSettings.EnableOutgoingWebhooks = true
+	})
+
+	t.Run("with a valid response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, strings.NewReader(`{"text": "Hello, World!"}`))
+		}))
+		defer server.Close()
+
+		resp, err := th.App.doOutgoingWebhookRequest(server.URL, strings.NewReader(""), "application/json")
+		require.Nil(t, err)
+
+		assert.NotNil(t, resp)
+		assert.NotNil(t, resp.Text)
+		assert.Equal(t, "Hello, World!", *resp.Text)
+	})
+
+	t.Run("with an invalid response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, strings.NewReader("aaaaaaaa"))
+		}))
+		defer server.Close()
+
+		_, err := th.App.doOutgoingWebhookRequest(server.URL, strings.NewReader(""), "application/json")
+		require.NotNil(t, err)
+		require.IsType(t, &json.SyntaxError{}, err)
+	})
+
+	t.Run("with a large, valid response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, io.MultiReader(strings.NewReader(`{"text": "`), InfiniteReader{}, strings.NewReader(`"}`)))
+		}))
+		defer server.Close()
+
+		_, err := th.App.doOutgoingWebhookRequest(server.URL, strings.NewReader(""), "application/json")
+		require.NotNil(t, err)
+		require.Equal(t, io.ErrUnexpectedEOF, err)
+	})
+
+	t.Run("with a large, invalid response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			io.Copy(w, InfiniteReader{})
+		}))
+		defer server.Close()
+
+		_, err := th.App.doOutgoingWebhookRequest(server.URL, strings.NewReader(""), "application/json")
+		require.NotNil(t, err)
+		require.IsType(t, &json.SyntaxError{}, err)
+	})
+
+	t.Run("with a slow response", func(t *testing.T) {
+		timeout := 100 * time.Millisecond
+
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(timeout + time.Millisecond)
+			io.Copy(w, strings.NewReader(`{"text": "Hello, World!"}`))
+		}))
+		defer server.Close()
+
+		th.App.HTTPService.(*httpservice.HTTPServiceImpl).RequestTimeout = timeout
+		defer func() {
+			th.App.HTTPService.(*httpservice.HTTPServiceImpl).RequestTimeout = httpservice.RequestTimeout
+		}()
+
+		_, err := th.App.doOutgoingWebhookRequest(server.URL, strings.NewReader(""), "application/json")
+		require.NotNil(t, err)
+		require.IsType(t, &url.Error{}, err)
+	})
+}

i18n/en.json

@@ -2126,6 +2126,10 @@
     "id": "api.user.authorize_oauth_user.missing.app_error",
     "translation": "Missing access token"
   },
+  {
+    "id": "api.user.authorize_oauth_user.response.app_error",
+    "translation": "Received invalid response from OAuth service provider"
+  },
   {
     "id": "api.user.authorize_oauth_user.service.app_error",
     "translation": "Token request to {{.Service}} failed"

model/outgoing_webhook.go

@@ -109,10 +109,10 @@ func (o *OutgoingWebhookResponse) ToJson() string {
 	return string(b)
 }
 
-func OutgoingWebhookResponseFromJson(data io.Reader) *OutgoingWebhookResponse {
+func OutgoingWebhookResponseFromJson(data io.Reader) (*OutgoingWebhookResponse, error) {
 	var o *OutgoingWebhookResponse
-	json.NewDecoder(data).Decode(&o)
-	return o
+	err := json.NewDecoder(data).Decode(&o)
+	return o, err
 }
 
 func (o *OutgoingWebhook) IsValid() *AppError {

model/outgoing_webhook_test.go

@@ -202,7 +202,7 @@ func TestOutgoingWebhookResponseJson(t *testing.T) {
 	o.Text = NewString("some text")
 
 	json := o.ToJson()
-	ro := OutgoingWebhookResponseFromJson(strings.NewReader(json))
+	ro, _ := OutgoingWebhookResponseFromJson(strings.NewReader(json))
 
 	if *o.Text != *ro.Text {
 		t.Fatal("Text does not match")

services/httpservice/client.go

@@ -126,10 +126,3 @@ func NewTransport(enableInsecureConnections bool, allowHost func(host string) bo
 		},
 	}
 }
-
-func NewHTTPClient(transport http.RoundTripper) *http.Client {
-	return &http.Client{
-		Transport: transport,
-		Timeout:   RequestTimeout,
-	}
-}

services/httpservice/client_test.go

@@ -139,3 +139,9 @@ func TestUserAgentIsSet(t *testing.T) {
 	}
 	client.Do(req)
 }
+
+func NewHTTPClient(transport http.RoundTripper) *http.Client {
+	return &http.Client{
+		Transport: transport,
+	}
+}

services/httpservice/httpservice.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/mattermost/mattermost-server/services/configservice"
 )
@@ -20,7 +21,7 @@ type HTTPService interface {
 	// MakeTransport returns a RoundTripper that is suitable for making requests to external resources. The default
 	// implementation provides:
 	// - A shorter timeout for dial and TLS handshake (defined as constant "ConnectTimeout")
-	// - A timeout for end-to-end requests (defined as constant "RequestTimeout")
+	// - A timeout for end-to-end requests
 	// - A Mattermost-specific user agent header
 	// - Additional security for untrusted and insecure connections
 	MakeTransport(trustURLs bool) http.RoundTripper
@@ -28,14 +29,22 @@ type HTTPService interface {
 
 type HTTPServiceImpl struct {
 	configService configservice.ConfigService
+
+	RequestTimeout time.Duration
 }
 
 func MakeHTTPService(configService configservice.ConfigService) HTTPService {
-	return &HTTPServiceImpl{configService}
+	return &HTTPServiceImpl{
+		configService,
+		RequestTimeout,
+	}
 }
 
 func (h *HTTPServiceImpl) MakeClient(trustURLs bool) *http.Client {
-	return NewHTTPClient(h.MakeTransport(trustURLs))
+	return &http.Client{
+		Transport: h.MakeTransport(trustURLs),
+		Timeout:   h.RequestTimeout,
+	}
 }
 
 func (h *HTTPServiceImpl) MakeTransport(trustURLs bool) http.RoundTripper {