File: dependencyproxy.go

package info (click to toggle)
gitlab 17.6.5-19
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 629,368 kB
  • sloc: ruby: 1,915,304; javascript: 557,307; sql: 60,639; xml: 6,509; sh: 4,567; makefile: 1,239; python: 406
file content (309 lines) | stat: -rw-r--r-- 8,898 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
// Package dependencyproxy provides functionality for handling dependency proxy operations
package dependencyproxy

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"gitlab.com/gitlab-org/labkit/log"

	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper/fail"
	"gitlab.com/gitlab-org/gitlab/workhorse/internal/senddata"
	"gitlab.com/gitlab-org/gitlab/workhorse/internal/transport"
	"gitlab.com/gitlab-org/gitlab/workhorse/internal/upload"
)

const dialTimeout = 10 * time.Second
const responseHeaderTimeout = 10 * time.Second
const uploadRequestGracePeriod = 60 * time.Second

var defaultTransportOptions = []transport.Option{transport.WithDialTimeout(dialTimeout), transport.WithResponseHeaderTimeout(responseHeaderTimeout)}

type cacheKey struct {
	ssrfFilter     bool
	allowLocalhost bool
	allowedURIs    string
}

var httpClients sync.Map

// Injector provides functionality for injecting dependencies
type Injector struct {
	senddata.Prefix
	uploadHandler upload.BodyUploadHandler
}

type entryParams struct {
	URL             string
	Headers         http.Header
	ResponseHeaders http.Header
	UploadConfig    uploadConfig
	SSRFFilter      bool
	AllowLocalhost  bool
	AllowedURIs     []string
}

type uploadConfig struct {
	Headers                  http.Header
	Method                   string
	URL                      string
	AuthorizedUploadResponse authorizeUploadResponse
}

type authorizeUploadResponse struct {
	TempPath            string
	RemoteObject        api.RemoteObject
	MaximumSize         int64
	UploadHashFunctions []string
}

func (u *uploadConfig) ExtractUploadAuthorizeFields() *api.Response {
	tempPath := u.AuthorizedUploadResponse.TempPath
	remoteID := u.AuthorizedUploadResponse.RemoteObject.RemoteTempObjectID

	if tempPath == "" && remoteID == "" {
		return nil
	}

	return &api.Response{
		TempPath:            tempPath,
		RemoteObject:        u.AuthorizedUploadResponse.RemoteObject,
		MaximumSize:         u.AuthorizedUploadResponse.MaximumSize,
		UploadHashFunctions: u.AuthorizedUploadResponse.UploadHashFunctions,
	}
}

type nullResponseWriter struct {
	header http.Header
	status int
}

func (nullResponseWriter) Write(p []byte) (int, error) {
	return len(p), nil
}

func (w *nullResponseWriter) Header() http.Header {
	return w.header
}

func (w *nullResponseWriter) WriteHeader(status int) {
	if w.status == 0 {
		w.status = status
	}
}

// NewInjector creates a new instance of Injector
func NewInjector() *Injector {
	return &Injector{Prefix: "send-dependency:"}
}

// SetUploadHandler sets the upload handler for the Injector
func (p *Injector) SetUploadHandler(uploadHandler upload.BodyUploadHandler) {
	p.uploadHandler = uploadHandler
}

// Inject performs the injection of dependencies
func (p *Injector) Inject(w http.ResponseWriter, r *http.Request, sendData string) {
	params, err := p.unpackParams(sendData)
	if err != nil {
		fail.Request(w, r, err)
		return
	}

	dependencyResponse, err := p.fetchURL(r.Context(), params)
	if err != nil {
		handleFetchError(w, r, err)
		return
	}
	defer func() { _ = dependencyResponse.Body.Close() }()

	if dependencyResponse.StatusCode >= 400 {
		handleErrorResponse(w, dependencyResponse)
		return
	}

	w.Header().Set("Content-Length", dependencyResponse.Header.Get("Content-Length"))

	teeReader := io.TeeReader(dependencyResponse.Body, w)
	// upload request context should follow the r context + a grace period
	ctx, cancel := context.WithCancel(context.WithoutCancel(r.Context()))
	defer cancel()

	stop := context.AfterFunc(r.Context(), func() {
		t := time.AfterFunc(uploadRequestGracePeriod, cancel) // call cancel function after 60 seconds

		context.AfterFunc(ctx, func() {
			if !t.Stop() { // if ctx is canceled and time still running, we stop the timer
				<-t.C // drain the channel because it's recommended in the docs: https://pkg.go.dev/time#Timer.Stop
			}
		})
	})
	defer stop()
	saveFileRequest, err := p.newUploadRequest(ctx, params, r, teeReader)
	if err != nil {
		fail.Request(w, r, fmt.Errorf("dependency proxy: failed to create request: %w", err))
	}

	forwardHeaders(dependencyResponse.Header, saveFileRequest)

	p.forwardHeadersToResponse(w, dependencyResponse.Header, params.ResponseHeaders)

	// workhorse hijack overwrites the Content-Type header, but we need this header value
	saveFileRequest.Header.Set("Workhorse-Proxy-Content-Type", dependencyResponse.Header.Get("Content-Type"))
	saveFileRequest.ContentLength = dependencyResponse.ContentLength

	nrw := &nullResponseWriter{header: make(http.Header)}
	apiResponse := params.UploadConfig.ExtractUploadAuthorizeFields()
	if apiResponse != nil {
		p.uploadHandler.ServeHTTPWithAPIResponse(nrw, saveFileRequest, apiResponse)
	} else {
		p.uploadHandler.ServeHTTP(nrw, saveFileRequest)
	}

	if nrw.status != http.StatusOK {
		fields := log.Fields{"code": nrw.status}

		fail.Request(nrw, saveFileRequest, fmt.Errorf("dependency proxy: failed to upload file"), fail.WithFields(fields))
	}
}

func handleFetchError(w http.ResponseWriter, r *http.Request, err error) {
	status := http.StatusBadGateway
	if os.IsTimeout(err) {
		status = http.StatusGatewayTimeout
	}
	fail.Request(w, r, err, fail.WithStatus(status))
}

func handleErrorResponse(w http.ResponseWriter, dependencyResponse *http.Response) {
	w.WriteHeader(dependencyResponse.StatusCode)
	_, _ = io.Copy(w, dependencyResponse.Body) // swallow errors for investigation, see https://gitlab.com/gitlab-org/gitlab/-/issues/459952.
}

// forwardHeaders forwards headers from the dependency response to the saveFileRequest.
func forwardHeaders(dependencyHeader http.Header, saveFileRequest *http.Request) {
	for key, values := range dependencyHeader {
		saveFileRequest.Header.Del(key)
		for _, value := range values {
			saveFileRequest.Header.Add(key, value)
		}
	}
}
func (p *Injector) fetchURL(ctx context.Context, params *entryParams) (*http.Response, error) {
	r, err := http.NewRequestWithContext(ctx, "GET", params.URL, nil)
	if err != nil {
		return nil, fmt.Errorf("dependency proxy: failed to fetch dependency: %w", err)
	}
	r.Header = params.Headers

	return cachedClient(params).Do(r)
}

func (p *Injector) newUploadRequest(ctx context.Context, params *entryParams, originalRequest *http.Request, body io.Reader) (*http.Request, error) {
	method := p.uploadMethodFrom(params)
	uploadURL := p.uploadURLFrom(params, originalRequest)
	request, err := http.NewRequestWithContext(ctx, method, uploadURL, body)
	if err != nil {
		return nil, err
	}

	request.Header = originalRequest.Header.Clone()

	for key, values := range params.UploadConfig.Headers {
		request.Header.Del(key)
		for _, value := range values {
			request.Header.Add(key, value)
		}
	}

	return request, nil
}

func (p *Injector) forwardHeadersToResponse(w http.ResponseWriter, headers ...http.Header) {
	for _, h := range headers {
		for key, values := range h {
			w.Header().Del(key)
			for _, v := range values {
				w.Header().Add(key, v)
			}
		}
	}
}

func (p *Injector) unpackParams(sendData string) (*entryParams, error) {
	var params entryParams
	if err := p.Unpack(&params, sendData); err != nil {
		return nil, fmt.Errorf("dependency proxy: unpack sendData: %w", err)
	}

	if err := p.validateParams(&params); err != nil {
		return nil, fmt.Errorf("dependency proxy: invalid params: %w", err)
	}

	return &params, nil
}

func (p *Injector) validateParams(params *entryParams) error {
	var uploadMethod = params.UploadConfig.Method
	if uploadMethod != "" && uploadMethod != http.MethodPost && uploadMethod != http.MethodPut {
		return fmt.Errorf("invalid upload method %s", uploadMethod)
	}

	var uploadURL = params.UploadConfig.URL
	if uploadURL != "" {
		if _, err := url.ParseRequestURI(uploadURL); err != nil {
			return fmt.Errorf("invalid upload url %w", err)
		}
	}

	return nil
}

func (p *Injector) uploadMethodFrom(params *entryParams) string {
	if params.UploadConfig.Method != "" {
		return params.UploadConfig.Method
	}
	return http.MethodPost
}

func (p *Injector) uploadURLFrom(params *entryParams, originalRequest *http.Request) string {
	if params.UploadConfig.URL != "" {
		return params.UploadConfig.URL
	}

	return originalRequest.URL.String() + "/upload"
}

func cachedClient(params *entryParams) *http.Client {
	key := cacheKey{
		allowLocalhost: params.AllowLocalhost,
		ssrfFilter:     params.SSRFFilter,
		allowedURIs:    strings.Join(params.AllowedURIs, ","),
	}
	cachedClient, found := httpClients.Load(key)
	if found {
		return cachedClient.(*http.Client)
	}

	options := defaultTransportOptions

	if params.SSRFFilter {
		options = append(options, transport.WithSSRFFilter(params.AllowLocalhost, params.AllowedURIs))
	}

	client := &http.Client{
		Transport: transport.NewRestrictedTransport(options...),
	}

	httpClients.Store(key, client)

	return client
}