| 1 | diff --git a/internal/blob/blob.go b/internal/blob/blob.go |
| 2 | index 6e79cbc..588d035 100644 |
| 3 | --- a/internal/blob/blob.go |
| 4 | +++ b/internal/blob/blob.go |
| 5 | @@ -99,7 +99,7 @@ type Service struct { |
| 6 | multipartFinalizing map[string]bool |
| 7 | // verifyMultipartHook is used by package tests to pause finalization |
| 8 | // without allocating a GiB fixture. It is nil in production. |
| 9 | - verifyMultipartHook func(string, string, int64) error |
| 10 | + verifyMultipartHook func(context.Context, string, string, int64) error |
| 11 | } |
| 12 | |
| 13 | var ErrBlocked = errors.New("blocked: this blob was removed by a moderator") |
| 14 | diff --git a/internal/blob/multipart.go b/internal/blob/multipart.go |
| 15 | index 0b744e2..398da74 100644 |
| 16 | --- a/internal/blob/multipart.go |
| 17 | +++ b/internal/blob/multipart.go |
| 18 | @@ -340,7 +340,7 @@ func (s *Service) commitMultipart(ctx context.Context, part multipartRequest, ch |
| 19 | if s.verifyMultipartHook != nil { |
| 20 | verify = s.verifyMultipartHook |
| 21 | } |
| 22 | - if err := verify(candidate.path, part.hash, part.length); err != nil { |
| 23 | + if err := verify(ctx, candidate.path, part.hash, part.length); err != nil { |
| 24 | s.multipartMu.Lock() |
| 25 | delete(s.multipartFinalizing, part.id) |
| 26 | // A canceled request must not throw away an otherwise complete session: |
| 27 | @@ -428,16 +428,34 @@ func (s *Service) recordMultipartRanges(ctx context.Context, id string, ranges [ |
| 28 | return err |
| 29 | }) |
| 30 | } |
| 31 | -func verifyMultipart(path, hash string, size int64) error { |
| 32 | + |
| 33 | +// verifyMultipart hashes the assembled file outside the tenant lock. It checks |
| 34 | +// the request context between reads so a disconnected client stops a long |
| 35 | +// hash instead of holding the session until the hash completes. |
| 36 | +func verifyMultipart(ctx context.Context, path, hash string, size int64) error { |
| 37 | file, err := os.Open(path) |
| 38 | if err != nil { |
| 39 | return err |
| 40 | } |
| 41 | defer file.Close() |
| 42 | hasher := sha256.New() |
| 43 | - count, err := io.Copy(hasher, file) |
| 44 | - if err != nil { |
| 45 | - return err |
| 46 | + var count int64 |
| 47 | + buffer := make([]byte, 1<<20) |
| 48 | + for { |
| 49 | + if err := ctx.Err(); err != nil { |
| 50 | + return err |
| 51 | + } |
| 52 | + n, readErr := file.Read(buffer) |
| 53 | + if n > 0 { |
| 54 | + hasher.Write(buffer[:n]) |
| 55 | + count += int64(n) |
| 56 | + } |
| 57 | + if errors.Is(readErr, io.EOF) { |
| 58 | + break |
| 59 | + } |
| 60 | + if readErr != nil { |
| 61 | + return readErr |
| 62 | + } |
| 63 | } |
| 64 | if count != size || hex.EncodeToString(hasher.Sum(nil)) != hash { |
| 65 | return ErrHashMismatch |
| 66 | diff --git a/internal/blob/multipart_review_test.go b/internal/blob/multipart_review_test.go |
| 67 | index 594cef1..ff3e901 100644 |
| 68 | --- a/internal/blob/multipart_review_test.go |
| 69 | +++ b/internal/blob/multipart_review_test.go |
| 70 | @@ -114,10 +114,10 @@ func TestMultipartReviewStalledHashDoesNotBlockOrdinaryUpload(t *testing.T) { |
| 71 | sha := hex.EncodeToString(digest[:]) |
| 72 | started := make(chan struct{}) |
| 73 | release := make(chan struct{}) |
| 74 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 75 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 76 | close(started) |
| 77 | <-release |
| 78 | - return verifyMultipart(path, hash, size) |
| 79 | + return verifyMultipart(ctx, path, hash, size) |
| 80 | } |
| 81 | result := make(chan *httptest.ResponseRecorder, 1) |
| 82 | r := multipartReviewRequest(sha, int64(len(body)), 0, body) |
| 83 | @@ -160,10 +160,10 @@ func TestMultipartReviewStalledHashDoesNotBlockAnotherSession(t *testing.T) { |
| 84 | release := make(chan struct{}) |
| 85 | var unblock sync.Once |
| 86 | defer unblock.Do(func() { close(release) }) |
| 87 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 88 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 89 | close(started) |
| 90 | <-release |
| 91 | - return verifyMultipart(path, hash, size) |
| 92 | + return verifyMultipart(ctx, path, hash, size) |
| 93 | } |
| 94 | first := make(chan *httptest.ResponseRecorder, 1) |
| 95 | go func() { first <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }() |
| 96 | @@ -207,10 +207,10 @@ func TestMultipartReviewFinalizingSessionRejectsConcurrentChunk(t *testing.T) { |
| 97 | sha := hex.EncodeToString(digest[:]) |
| 98 | started := make(chan struct{}) |
| 99 | release := make(chan struct{}) |
| 100 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 101 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 102 | close(started) |
| 103 | <-release |
| 104 | - return verifyMultipart(path, hash, size) |
| 105 | + return verifyMultipart(ctx, path, hash, size) |
| 106 | } |
| 107 | first := make(chan *httptest.ResponseRecorder, 1) |
| 108 | go func() { first <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }() |
| 109 | @@ -236,10 +236,10 @@ func TestMultipartReviewCleanupDoesNotRemoveActiveFinalization(t *testing.T) { |
| 110 | sha := hex.EncodeToString(digest[:]) |
| 111 | started := make(chan struct{}) |
| 112 | release := make(chan struct{}) |
| 113 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 114 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 115 | close(started) |
| 116 | <-release |
| 117 | - return verifyMultipart(path, hash, size) |
| 118 | + return verifyMultipart(ctx, path, hash, size) |
| 119 | } |
| 120 | result := make(chan *httptest.ResponseRecorder, 1) |
| 121 | go func() { result <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }() |
| 122 | @@ -278,10 +278,10 @@ func TestMultipartReviewPolicyRecheckedAfterHash(t *testing.T) { |
| 123 | sha := hex.EncodeToString(digest[:]) |
| 124 | started := make(chan struct{}) |
| 125 | release := make(chan struct{}) |
| 126 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 127 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 128 | close(started) |
| 129 | <-release |
| 130 | - return verifyMultipart(path, hash, size) |
| 131 | + return verifyMultipart(ctx, path, hash, size) |
| 132 | } |
| 133 | s.config.Limits = func() Limits { return Limits{MaxFileBytes: int64(len(body))} } |
| 134 | result := make(chan *httptest.ResponseRecorder, 1) |
| 135 | @@ -304,10 +304,10 @@ func TestMultipartReviewCanceledHashCanBeRetried(t *testing.T) { |
| 136 | sha := hex.EncodeToString(digest[:]) |
| 137 | started := make(chan struct{}) |
| 138 | release := make(chan struct{}) |
| 139 | - s.verifyMultipartHook = func(path, hash string, size int64) error { |
| 140 | + s.verifyMultipartHook = func(ctx context.Context, path, hash string, size int64) error { |
| 141 | close(started) |
| 142 | <-release |
| 143 | - return verifyMultipart(path, hash, size) |
| 144 | + return verifyMultipart(ctx, path, hash, size) |
| 145 | } |
| 146 | ctx, cancel := context.WithCancel(context.Background()) |
| 147 | r := multipartReviewRequest(sha, int64(len(body)), 0, body).WithContext(ctx) |
| 148 | diff --git a/internal/gitrelay/pr_budget_proxy.go b/internal/gitrelay/pr_budget_proxy.go |
| 149 | index 9da3093..f2433d4 100644 |
| 150 | --- a/internal/gitrelay/pr_budget_proxy.go |
| 151 | +++ b/internal/gitrelay/pr_budget_proxy.go |
| 152 | @@ -3,6 +3,10 @@ package gitrelay |
| 153 | import ( |
| 154 | "bufio" |
| 155 | "context" |
| 156 | + "crypto/rand" |
| 157 | + "crypto/subtle" |
| 158 | + "encoding/base64" |
| 159 | + "encoding/hex" |
| 160 | "errors" |
| 161 | "io" |
| 162 | "net" |
| 163 | @@ -29,7 +33,11 @@ func (b *byteBudget) take(wanted, limit int64) int64 { |
| 164 | } |
| 165 | |
| 166 | type repairProxy struct { |
| 167 | - listener net.Listener |
| 168 | + listener net.Listener |
| 169 | + // credential is the Proxy-Authorization value Git presents. The listener |
| 170 | + // is loopback only, but other local processes cannot borrow the tunnel. |
| 171 | + credential string |
| 172 | + url string |
| 173 | ctx context.Context |
| 174 | cancel context.CancelFunc |
| 175 | budget *byteBudget |
| 176 | @@ -47,14 +55,22 @@ func newRepairProxy(ctx context.Context, targetIP, targetHost, targetPort string |
| 177 | if err != nil { |
| 178 | return nil, err |
| 179 | } |
| 180 | + var secret [16]byte |
| 181 | + if _, err := rand.Read(secret[:]); err != nil { |
| 182 | + _ = listener.Close() |
| 183 | + return nil, err |
| 184 | + } |
| 185 | proxyCtx, cancel := context.WithCancel(ctx) |
| 186 | - p := &repairProxy{listener: listener, ctx: proxyCtx, cancel: cancel, budget: budget, limit: limit, targetIP: targetIP, targetHost: targetHost, targetPort: targetPort, connections: make(map[net.Conn]struct{})} |
| 187 | + token := hex.EncodeToString(secret[:]) |
| 188 | + credential := "Basic " + base64.StdEncoding.EncodeToString([]byte("tiny:"+token)) |
| 189 | + p := &repairProxy{listener: listener, credential: credential, ctx: proxyCtx, cancel: cancel, budget: budget, limit: limit, targetIP: targetIP, targetHost: targetHost, targetPort: targetPort, connections: make(map[net.Conn]struct{})} |
| 190 | + p.url = "http://tiny:" + token + "@" + listener.Addr().String() |
| 191 | context.AfterFunc(proxyCtx, func() { _ = p.Close() }) |
| 192 | go p.serve() |
| 193 | return p, nil |
| 194 | } |
| 195 | |
| 196 | -func (p *repairProxy) URL() string { return "http://" + p.listener.Addr().String() } |
| 197 | +func (p *repairProxy) URL() string { return p.url } |
| 198 | |
| 199 | func (p *repairProxy) Close() error { |
| 200 | p.closeOnce.Do(func() { |
| 201 | @@ -109,6 +125,10 @@ func (p *repairProxy) handle(client net.Conn) { |
| 202 | if request.Method != http.MethodConnect && request.Method != http.MethodGet && request.Method != http.MethodPost { |
| 203 | return |
| 204 | } |
| 205 | + if subtle.ConstantTimeCompare([]byte(request.Header.Get("Proxy-Authorization")), []byte(p.credential)) != 1 { |
| 206 | + _, _ = io.WriteString(client, "HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic realm=\"tiny\"\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") |
| 207 | + return |
| 208 | + } |
| 209 | host := request.URL.Hostname() |
| 210 | port := request.URL.Port() |
| 211 | if port == "" { |