Antonio Mika
·
2026-10-09
1package rsync
2
3import (
4 "bytes"
5 stdflate "compress/flate"
6 "errors"
7 "fmt"
8 "io"
9
10 "github.com/klauspost/compress/flate"
11)
12
13// compression identifies the negotiated token compression.
14type compression int
15
16const (
17 compressNone compression = iota
18 compressZlib
19 compressZlibx
20)
21
22// compressNames is our preference order for negotiation. zlib adds matched
23// blocks to the deflate history; zlibx does not.
24var compressNames = []struct {
25 name string
26 typ compression
27}{
28 {"zlibx", compressZlibx},
29 {"zlib", compressZlib},
30 {"none", compressNone},
31}
32
33func parseCompressName(name string) (compression, error) {
34 for _, c := range compressNames {
35 if c.name == name {
36 return c.typ, nil
37 }
38 }
39 return 0, fmt.Errorf("unknown compress name: %s", name)
40}
41
42// tokenSender emits one file's delta. token is a block index for a match, -1
43// at the end of the file, or -2 to send literal data without a token.
44// matched holds the matched block's data.
45type tokenSender interface {
46 send(token int32, literal, matched []byte) error
47}
48
49// tokenReceiver decodes one file's delta. recv returns a positive length
50// with literal data, -(block+1) for a match, or 0 at the end of the file.
51// see is called with each matched block's data.
52type tokenReceiver interface {
53 recv() (int32, []byte, error)
54 see(data []byte)
55}
56
57func newTokenSender(c *conn, comp compression, level int) tokenSender {
58 if comp == compressNone {
59 return &simpleTokenSender{c: c}
60 }
61 return &deflateTokenSender{c: c, zlib: comp == compressZlib, level: level, lastToken: -1}
62}
63
64func newTokenReceiver(c *conn, comp compression) tokenReceiver {
65 if comp == compressNone {
66 return &simpleTokenReceiver{c: c, buf: make([]byte, chunkSize)}
67 }
68 return &deflateTokenReceiver{c: c, zlib: comp == compressZlib, out: make([]byte, chunkSize)}
69}
70
71type simpleTokenSender struct{ c *conn }
72
73func (s *simpleTokenSender) send(token int32, literal, _ []byte) error {
74 for len(literal) > 0 {
75 n := min(len(literal), chunkSize)
76 if err := s.c.writeInt32(int32(n)); err != nil {
77 return err
78 }
79 if err := s.c.write(literal[:n]); err != nil {
80 return err
81 }
82 literal = literal[n:]
83 }
84 if token == -2 {
85 return nil
86 }
87 return s.c.writeInt32(-(token + 1))
88}
89
90type simpleTokenReceiver struct {
91 c *conn
92 buf []byte
93 residue int32
94}
95
96func (s *simpleTokenReceiver) recv() (int32, []byte, error) {
97 if s.residue == 0 {
98 i, err := s.c.readInt32()
99 if err != nil || i <= 0 {
100 return i, nil, err
101 }
102 s.residue = i
103 }
104 n := min(int32(len(s.buf)), s.residue)
105 s.residue -= n
106 if err := s.c.readFull(s.buf[:n]); err != nil {
107 return 0, nil, err
108 }
109 return n, s.buf[:n], nil
110}
111
112func (s *simpleTokenReceiver) see([]byte) {}
113
114// Flag bytes of the compressed token stream.
115const (
116 tokEndFlag = 0
117 tokLong = 0x20
118 tokRunLong = 0x21
119 tokDeflatedData = 0x40
120 tokRel = 0x80
121 tokRunRel = 0xc0
122 maxTokenIndex = 0x7ffffffe
123 deflateWindow = 32 * 1024
124)
125
126var syncMarker = []byte{0, 0, 0xff, 0xff}
127
128// deflateTokenSender produces rsync's compressed token stream: literal data
129// is raw deflate, sync-flushed before every token with the trailing
130// 00 00 ff ff marker stripped.
131type deflateTokenSender struct {
132 c *conn
133 zlib bool
134 level int
135
136 w *flate.Writer
137 out bytes.Buffer
138
139 lastToken, runStart, lastRunEnd int32
140 flushPending bool
141
142 history []byte
143 needDict bool
144}
145
146func (d *deflateTokenSender) send(token int32, literal, matched []byte) error {
147 switch {
148 case d.lastToken == -1:
149 if err := d.resetStream(); err != nil {
150 return err
151 }
152 d.lastRunEnd = 0
153 d.runStart = token
154 d.flushPending = false
155 case d.lastToken == -2:
156 d.runStart = token
157 case len(literal) != 0 || token != d.lastToken+1 || token >= d.runStart+65536:
158 if err := d.writeRun(); err != nil {
159 return err
160 }
161 d.lastRunEnd = d.lastToken
162 d.runStart = token
163 }
164 d.lastToken = token
165
166 if len(literal) != 0 || d.flushPending {
167 if err := d.deflate(literal, token != -2); err != nil {
168 return err
169 }
170 d.flushPending = token == -2
171 }
172
173 if token == -1 {
174 return d.c.writeByte(tokEndFlag)
175 }
176 if token != -2 && d.zlib {
177 d.addHistory(matched, d.c.protocol)
178 d.needDict = true
179 }
180 return nil
181}
182
183func (d *deflateTokenSender) resetStream() error {
184 d.history = d.history[:0]
185 d.needDict = false
186 d.out.Reset()
187 if d.w == nil {
188 w, err := flate.NewWriter(&d.out, d.level)
189 if err != nil {
190 return err
191 }
192 d.w = w
193 return nil
194 }
195 // Reset would reuse the last dictionary, so clear it explicitly.
196 d.w.ResetDict(&d.out, nil)
197 return nil
198}
199
200func (d *deflateTokenSender) writeRun() error {
201 r := d.runStart - d.lastRunEnd
202 n := d.lastToken - d.runStart
203 if r >= 0 && r <= 63 {
204 flag := byte(tokRel)
205 if n != 0 {
206 flag = tokRunRel
207 }
208 if err := d.c.writeByte(flag + byte(r)); err != nil {
209 return err
210 }
211 } else {
212 flag := byte(tokLong)
213 if n != 0 {
214 flag = tokRunLong
215 }
216 if err := d.c.writeByte(flag); err != nil {
217 return err
218 }
219 if err := d.c.writeInt32(d.runStart); err != nil {
220 return err
221 }
222 }
223 if n != 0 {
224 return d.c.write([]byte{byte(n), byte(n >> 8)})
225 }
226 return nil
227}
228
229func (d *deflateTokenSender) deflate(literal []byte, flush bool) error {
230 if d.needDict {
231 // The receiver's window holds the matched blocks too, so restart
232 // the compressor with the same history.
233 d.w.ResetDict(&d.out, d.window())
234 d.needDict = false
235 }
236 if len(literal) > 0 {
237 if _, err := d.w.Write(literal); err != nil {
238 return err
239 }
240 d.addHistory(literal, 31)
241 }
242 if flush {
243 if err := d.w.Flush(); err != nil {
244 return err
245 }
246 if !bytes.HasSuffix(d.out.Bytes(), syncMarker) {
247 return errors.New("rsync: deflate flush did not end in a sync marker")
248 }
249 d.out.Truncate(d.out.Len() - len(syncMarker))
250 }
251 data := d.out.Bytes()
252 for len(data) > 0 {
253 n := min(len(data), maxDataCount)
254 if err := d.c.write([]byte{byte(tokDeflatedData + n>>8), byte(n)}); err != nil {
255 return err
256 }
257 if err := d.c.write(data[:n]); err != nil {
258 return err
259 }
260 data = data[n:]
261 }
262 d.out.Reset()
263 return nil
264}
265
266// addHistory appends data to the deflate history. Before protocol 31 rsync
267// fed every 64K piece of a long matched block from its start, and both
268// sides must reproduce that.
269func (d *deflateTokenSender) addHistory(data []byte, protocol int) {
270 d.history = appendHistory(d.history, data, protocol)
271}
272
273func (d *deflateTokenSender) window() []byte {
274 if len(d.history) > deflateWindow {
275 return d.history[len(d.history)-deflateWindow:]
276 }
277 return d.history
278}
279
280func appendHistory(history, data []byte, protocol int) []byte {
281 toklen := len(data)
282 src := data
283 for toklen > 0 {
284 n := min(toklen, 0xffff)
285 history = append(history, src[:n]...)
286 if protocol >= 31 {
287 src = src[n:]
288 }
289 toklen -= n
290 }
291 if len(history) > 2*deflateWindow {
292 history = append(history[:0], history[len(history)-deflateWindow:]...)
293 }
294 return history
295}
296
297type deflateRecvState int
298
299const (
300 rInit deflateRecvState = iota
301 rIdle
302 rRunning
303 rInflating
304)
305
306// deflateTokenReceiver decodes the compressed token stream. Each run of
307// DEFLATED_DATA chunks is inflated as one segment: the source appends the
308// stripped sync marker when the next flag arrives and then reports EOF, and
309// the inflater is restarted with the current history for the next segment.
310type deflateTokenReceiver struct {
311 c *conn
312 zlib bool
313
314 state deflateRecvState
315 rxToken int32
316 rxRun int32
317 savedFlag int
318 hasSaved bool
319
320 src segmentSource
321 inflate io.ReadCloser
322 history []byte
323 out []byte
324}
325
326func (d *deflateTokenReceiver) recv() (int32, []byte, error) {
327 for {
328 switch d.state {
329 case rInit:
330 d.history = d.history[:0]
331 d.rxToken = 0
332 d.state = rIdle
333
334 case rIdle:
335 var flag int
336 if d.hasSaved {
337 flag, d.hasSaved = d.savedFlag, false
338 } else {
339 b, err := d.c.readByte()
340 if err != nil {
341 return 0, nil, err
342 }
343 flag = int(b)
344 }
345 if flag&0xC0 == tokDeflatedData {
346 if err := d.src.start(d.c, flag); err != nil {
347 return 0, nil, err
348 }
349 if d.inflate == nil {
350 d.inflate = stdflate.NewReaderDict(&d.src, d.window())
351 } else if err := d.inflate.(stdflate.Resetter).Reset(&d.src, d.window()); err != nil {
352 return 0, nil, err
353 }
354 d.state = rInflating
355 continue
356 }
357 if flag == tokEndFlag {
358 d.state = rInit
359 return 0, nil, nil
360 }
361 return d.tokenNum(flag)
362
363 case rInflating:
364 n, err := d.inflate.Read(d.out)
365 if n > 0 {
366 d.history = appendHistory(d.history, d.out[:n], 31)
367 return int32(n), d.out[:n], nil
368 }
369 if err == nil {
370 continue
371 }
372 if d.src.err != nil {
373 return 0, nil, d.src.err
374 }
375 if !errors.Is(err, io.ErrUnexpectedEOF) || !d.src.done() {
376 return 0, nil, fmt.Errorf("rsync: inflate failed: %w", err)
377 }
378 d.savedFlag, d.hasSaved = d.src.next, true
379 d.state = rIdle
380
381 case rRunning:
382 if d.rxRun <= 0 || d.rxToken >= maxTokenIndex {
383 return 0, nil, errors.New("rsync: invalid token number in compressed stream")
384 }
385 d.rxToken++
386 d.rxRun--
387 if d.rxRun == 0 {
388 d.state = rIdle
389 }
390 return -1 - d.rxToken, nil, nil
391 }
392 }
393}
394
395func (d *deflateTokenReceiver) tokenNum(flag int) (int32, []byte, error) {
396 invalid := errors.New("rsync: invalid token number in compressed stream")
397 if flag&tokRel != 0 {
398 incr := int32(flag & 0x3f)
399 if d.rxToken > maxTokenIndex-incr {
400 return 0, nil, invalid
401 }
402 d.rxToken += incr
403 flag >>= 6
404 } else {
405 v, err := d.c.readInt32()
406 if err != nil {
407 return 0, nil, err
408 }
409 if v < 0 || v > maxTokenIndex {
410 return 0, nil, invalid
411 }
412 d.rxToken = v
413 }
414 if flag&1 != 0 {
415 lo, err := d.c.readByte()
416 if err != nil {
417 return 0, nil, err
418 }
419 hi, err := d.c.readByte()
420 if err != nil {
421 return 0, nil, err
422 }
423 d.rxRun = int32(lo) + int32(hi)<<8
424 if d.rxRun <= 0 || d.rxToken > maxTokenIndex-d.rxRun {
425 return 0, nil, invalid
426 }
427 d.state = rRunning
428 }
429 return -1 - d.rxToken, nil, nil
430}
431
432func (d *deflateTokenReceiver) see(data []byte) {
433 if d.zlib {
434 d.history = appendHistory(d.history, data, d.c.protocol)
435 }
436}
437
438func (d *deflateTokenReceiver) window() []byte {
439 if len(d.history) > deflateWindow {
440 return d.history[len(d.history)-deflateWindow:]
441 }
442 return d.history
443}
444
445// segmentSource feeds one segment of DEFLATED_DATA chunks to the inflater,
446// reading further chunks from the connection on demand.
447type segmentSource struct {
448 c *conn
449 buf []byte
450 pos int
451 marker int // bytes of the sync marker handed out, -1 before the segment ends
452 next int // flag byte that ended the segment
453 err error
454 scratch [maxDataCount]byte
455}
456
457func (s *segmentSource) start(c *conn, flag int) error {
458 s.c = c
459 s.marker = -1
460 s.err = nil
461 return s.readChunk(flag)
462}
463
464func (s *segmentSource) readChunk(flag int) error {
465 lo, err := s.c.readByte()
466 if err != nil {
467 return err
468 }
469 n := (flag&0x3f)<<8 + int(lo)
470 if err := s.c.readFull(s.scratch[:n]); err != nil {
471 return err
472 }
473 s.buf = s.scratch[:n]
474 s.pos = 0
475 return nil
476}
477
478func (s *segmentSource) done() bool { return s.marker == len(syncMarker) }
479
480func (s *segmentSource) ReadByte() (byte, error) {
481 for {
482 if s.marker >= 0 {
483 if s.marker == len(syncMarker) {
484 return 0, io.EOF
485 }
486 b := syncMarker[s.marker]
487 s.marker++
488 return b, nil
489 }
490 if s.pos < len(s.buf) {
491 b := s.buf[s.pos]
492 s.pos++
493 return b, nil
494 }
495 b, err := s.c.readByte()
496 if err != nil {
497 s.err = err
498 return 0, err
499 }
500 if int(b)&0xC0 == tokDeflatedData {
501 if err := s.readChunk(int(b)); err != nil {
502 s.err = err
503 return 0, err
504 }
505 continue
506 }
507 s.next = int(b)
508 s.marker = 0
509 }
510}
511
512func (s *segmentSource) Read(p []byte) (int, error) {
513 n := 0
514 for n < len(p) {
515 if s.marker < 0 && s.pos < len(s.buf) {
516 m := copy(p[n:], s.buf[s.pos:])
517 s.pos += m
518 n += m
519 continue
520 }
521 if n > 0 {
522 return n, nil
523 }
524 b, err := s.ReadByte()
525 if err != nil {
526 return n, err
527 }
528 p[n] = b
529 n++
530 }
531 return n, nil
532}