Antonio Mika
·
2026-10-09
1package rsync
2
3import (
4 "bufio"
5 "encoding/binary"
6 "errors"
7 "fmt"
8 "io"
9 "sync"
10)
11
12// remoteExitError is returned when the peer aborts with MSG_ERROR_EXIT.
13type remoteExitError struct{ code int }
14
15func (e *remoteExitError) Error() string {
16 return fmt.Sprintf("remote side exited with code %d", e.code)
17}
18
19// muxWriter queues protocol data and messages in order and writes them from
20// its own goroutine. Once multiplexing is on, data goes out as MSG_DATA
21// frames. Writers of data block when too much is queued; messages never
22// block, so the receiver can report progress while the generator waits on
23// the peer.
24type muxWriter struct {
25 mu sync.Mutex
26 cond *sync.Cond
27 w io.Writer
28 queue []muxItem
29 queued int
30 kick bool
31 busy bool
32 multiplex bool
33 written int64
34 err error
35}
36
37type muxItem struct {
38 tag byte
39 data []byte
40}
41
42const (
43 muxBatchSize = 32 * 1024
44 muxMaxQueued = 256 * 1024
45)
46
47func newMuxWriter(w io.Writer) *muxWriter {
48 m := &muxWriter{w: w}
49 m.cond = sync.NewCond(&m.mu)
50 go m.pump()
51 return m
52}
53
54func (m *muxWriter) Write(p []byte) (int, error) {
55 m.mu.Lock()
56 defer m.mu.Unlock()
57 for m.err == nil && m.queued >= muxMaxQueued {
58 m.kick = true
59 m.cond.Broadcast()
60 m.cond.Wait()
61 }
62 if m.err != nil {
63 return 0, m.err
64 }
65 if n := len(m.queue); n > 0 && m.queue[n-1].tag == msgData && len(m.queue[n-1].data) < muxBatchSize {
66 m.queue[n-1].data = append(m.queue[n-1].data, p...)
67 } else {
68 m.queue = append(m.queue, muxItem{tag: msgData, data: append([]byte(nil), p...)})
69 }
70 m.queued += len(p)
71 if m.queued >= muxBatchSize {
72 m.kick = true
73 m.cond.Broadcast()
74 }
75 return len(p), nil
76}
77
78// WriteMsg queues an out-of-band message behind any pending data.
79func (m *muxWriter) WriteMsg(tag byte, p []byte) error {
80 m.mu.Lock()
81 defer m.mu.Unlock()
82 if m.err != nil {
83 return m.err
84 }
85 if !m.multiplex {
86 return errors.New("rsync: message sent before multiplexing started")
87 }
88 m.queue = append(m.queue, muxItem{tag: tag, data: append([]byte(nil), p...)})
89 m.kick = true
90 m.cond.Broadcast()
91 return nil
92}
93
94// Kick asks for queued output to go out without waiting for it.
95func (m *muxWriter) Kick() error {
96 m.mu.Lock()
97 defer m.mu.Unlock()
98 if len(m.queue) > 0 {
99 m.kick = true
100 m.cond.Broadcast()
101 }
102 return m.err
103}
104
105// Flush waits until everything queued has been written.
106func (m *muxWriter) Flush() error {
107 m.mu.Lock()
108 defer m.mu.Unlock()
109 for m.err == nil && (len(m.queue) > 0 || m.busy) {
110 m.kick = true
111 m.cond.Broadcast()
112 m.cond.Wait()
113 }
114 return m.err
115}
116
117// errMuxClosed is returned to writers that outlive Close.
118var errMuxClosed = errors.New("rsync: write after close")
119
120// Close flushes and stops the writer goroutine. Later writes fail.
121func (m *muxWriter) Close() error {
122 err := m.Flush()
123 m.mu.Lock()
124 if m.err == nil {
125 m.err = errMuxClosed
126 }
127 m.cond.Broadcast()
128 m.mu.Unlock()
129 return err
130}
131
132func (m *muxWriter) setMultiplex(on bool) error {
133 if err := m.Flush(); err != nil {
134 return err
135 }
136 m.mu.Lock()
137 m.multiplex = on
138 m.mu.Unlock()
139 return nil
140}
141
142func (m *muxWriter) pump() {
143 m.mu.Lock()
144 defer m.mu.Unlock()
145 for {
146 for m.err == nil && (!m.kick || len(m.queue) == 0) {
147 m.cond.Wait()
148 }
149 if m.err != nil {
150 m.queue = nil
151 m.cond.Broadcast()
152 return
153 }
154 items := m.queue
155 m.queue = nil
156 m.queued = 0
157 m.kick = false
158 m.busy = true
159 multiplex := m.multiplex
160 m.cond.Broadcast()
161 m.mu.Unlock()
162
163 var n int64
164 var err error
165 for _, it := range items {
166 var k int64
167 k, err = writeItem(m.w, it, multiplex)
168 n += k
169 if err != nil {
170 break
171 }
172 }
173
174 m.mu.Lock()
175 m.busy = false
176 m.written += n
177 if err != nil {
178 m.err = err
179 }
180 m.cond.Broadcast()
181 }
182}
183
184func writeItem(w io.Writer, it muxItem, multiplex bool) (int64, error) {
185 if !multiplex {
186 n, err := w.Write(it.data)
187 return int64(n), err
188 }
189 var total int64
190 data := it.data
191 for {
192 chunk := data[:min(len(data), 0xFFFFFF)]
193 var hdr [4]byte
194 binary.LittleEndian.PutUint32(hdr[:], uint32(mplexBase+int(it.tag))<<24|uint32(len(chunk)))
195 n, err := w.Write(append(hdr[:], chunk...))
196 total += int64(n)
197 if err != nil {
198 return total, err
199 }
200 data = data[len(chunk):]
201 if len(data) == 0 {
202 return total, nil
203 }
204 }
205}
206
207func (m *muxWriter) bytesWritten() int64 {
208 m.mu.Lock()
209 defer m.mu.Unlock()
210 n := m.written + int64(m.queued)
211 for _, it := range m.queue {
212 if it.tag != msgData {
213 n += int64(len(it.data)) + 4
214 }
215 }
216 return n
217}
218
219// flushingReader sends our pending output before it blocks on the peer,
220// which is what keeps request/response exchanges from stalling.
221type flushingReader struct {
222 r io.Reader
223 flush func() error
224 read int64
225}
226
227func (f *flushingReader) Read(p []byte) (int, error) {
228 if err := f.flush(); err != nil {
229 return 0, err
230 }
231 n, err := f.r.Read(p)
232 f.read += int64(n)
233 return n, err
234}
235
236// muxReader returns the data stream of a possibly multiplexed input and
237// hands every other message to onMsg.
238type muxReader struct {
239 r *bufio.Reader
240 src *flushingReader
241 multiplex bool
242 remaining int
243 onMsg func(tag byte, data []byte) error
244}
245
246func (m *muxReader) fill() error {
247 for m.remaining == 0 {
248 var hdr [4]byte
249 if _, err := io.ReadFull(m.r, hdr[:]); err != nil {
250 return err
251 }
252 v := binary.LittleEndian.Uint32(hdr[:])
253 tag := int(v>>24) - mplexBase
254 size := int(v & 0xFFFFFF)
255 if tag == msgData {
256 m.remaining = size
257 continue
258 }
259 if tag < 0 {
260 return fmt.Errorf("rsync: unexpected tag %d", tag)
261 }
262 data := make([]byte, size)
263 if _, err := io.ReadFull(m.r, data); err != nil {
264 return err
265 }
266 if err := m.onMsg(byte(tag), data); err != nil {
267 return err
268 }
269 }
270 return nil
271}
272
273func (m *muxReader) Read(p []byte) (int, error) {
274 if len(p) == 0 {
275 return 0, nil
276 }
277 if !m.multiplex {
278 return m.r.Read(p)
279 }
280 if err := m.fill(); err != nil {
281 return 0, err
282 }
283 n, err := m.r.Read(p[:min(len(p), m.remaining)])
284 m.remaining -= n
285 return n, err
286}
287
288func (m *muxReader) ReadByte() (byte, error) {
289 if !m.multiplex {
290 return m.r.ReadByte()
291 }
292 if err := m.fill(); err != nil {
293 return 0, err
294 }
295 b, err := m.r.ReadByte()
296 if err == nil {
297 m.remaining--
298 }
299 return b, err
300}
301
302// ndxState holds the delta-encoding state of one direction of write_ndx.
303type ndxState struct {
304 prevPositive int32
305 prevNegative int32
306}
307
308func newNdxState() ndxState { return ndxState{prevPositive: -1, prevNegative: 1} }
309
310// conn carries the rsync protocol over the session's stdin and stdout.
311type conn struct {
312 protocol int
313 r *muxReader
314 w *muxWriter
315 ndxIn ndxState
316 ndxOut ndxState
317 rbuf [16]byte
318 wbuf [16]byte
319}
320
321func newConn(rw io.ReadWriter) *conn {
322 w := newMuxWriter(rw)
323 src := &flushingReader{r: rw, flush: w.Kick}
324 c := &conn{
325 r: &muxReader{r: bufio.NewReaderSize(src, 64*1024), src: src},
326 w: w,
327 ndxIn: newNdxState(),
328 ndxOut: newNdxState(),
329 }
330 c.r.onMsg = func(tag byte, data []byte) error {
331 return fmt.Errorf("rsync: unexpected message %d", tag)
332 }
333 return c
334}
335
336func (c *conn) bytesRead() int64 { return c.r.src.read }
337func (c *conn) bytesWritten() int64 { return c.w.bytesWritten() }
338func (c *conn) flush() error { return c.w.Flush() }
339func (c *conn) close() error { return c.w.Close() }
340
341func (c *conn) readFull(p []byte) error {
342 _, err := io.ReadFull(c.r, p)
343 if errors.Is(err, io.EOF) {
344 return io.ErrUnexpectedEOF
345 }
346 return err
347}
348
349func (c *conn) readByte() (byte, error) {
350 b, err := c.r.ReadByte()
351 if errors.Is(err, io.EOF) {
352 return 0, io.ErrUnexpectedEOF
353 }
354 return b, err
355}
356
357func (c *conn) readInt32() (int32, error) {
358 if err := c.readFull(c.rbuf[:4]); err != nil {
359 return 0, err
360 }
361 return int32(binary.LittleEndian.Uint32(c.rbuf[:4])), nil
362}
363
364func (c *conn) readShortint() (uint16, error) {
365 if err := c.readFull(c.rbuf[:2]); err != nil {
366 return 0, err
367 }
368 return binary.LittleEndian.Uint16(c.rbuf[:2]), nil
369}
370
371// readLongint reads the pre-30 64-bit encoding: an int32, or -1 followed by
372// the full 64-bit value.
373func (c *conn) readLongint() (int64, error) {
374 v, err := c.readInt32()
375 if err != nil || v != -1 {
376 return int64(v), err
377 }
378 if err := c.readFull(c.rbuf[:8]); err != nil {
379 return 0, err
380 }
381 return int64(binary.LittleEndian.Uint64(c.rbuf[:8])), nil
382}
383
384// intByteExtra maps the high bits of a varint's first byte to the number of
385// bytes that follow it.
386var intByteExtra = [64]int{
387 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
388 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
389 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
390 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6,
391}
392
393func (c *conn) readVarint() (int32, error) {
394 ch, err := c.readByte()
395 if err != nil {
396 return 0, err
397 }
398 var b [5]byte
399 extra := intByteExtra[ch/4]
400 if extra == 0 {
401 return int32(ch), nil
402 }
403 if extra >= len(b) {
404 return 0, errors.New("rsync: overflow in varint")
405 }
406 if err := c.readFull(b[:extra]); err != nil {
407 return 0, err
408 }
409 bit := byte(1) << (8 - extra)
410 b[extra] = ch & (bit - 1)
411 return int32(binary.LittleEndian.Uint32(b[:4])), nil
412}
413
414func (c *conn) readVarlong(minBytes int) (int64, error) {
415 var b2 [8]byte
416 if err := c.readFull(b2[:minBytes]); err != nil {
417 return 0, err
418 }
419 var b [9]byte
420 copy(b[:], b2[1:minBytes])
421 extra := intByteExtra[b2[0]/4]
422 if extra == 0 {
423 b[minBytes-1] = b2[0]
424 return int64(binary.LittleEndian.Uint64(b[:8])), nil
425 }
426 if minBytes+extra > len(b) {
427 return 0, errors.New("rsync: overflow in varlong")
428 }
429 if err := c.readFull(b[minBytes-1 : minBytes-1+extra]); err != nil {
430 return 0, err
431 }
432 bit := byte(1) << (8 - extra)
433 b[minBytes+extra-1] = b2[0] & (bit - 1)
434 return int64(binary.LittleEndian.Uint64(b[:8])), nil
435}
436
437func (c *conn) readVarint30() (int32, error) {
438 if c.protocol < 30 {
439 return c.readInt32()
440 }
441 return c.readVarint()
442}
443
444func (c *conn) readVarlong30(minBytes int) (int64, error) {
445 if c.protocol < 30 {
446 return c.readLongint()
447 }
448 return c.readVarlong(minBytes)
449}
450
451func (c *conn) readVstring(max int) (string, error) {
452 b, err := c.readByte()
453 if err != nil {
454 return "", err
455 }
456 n := int(b)
457 if n&0x80 != 0 {
458 lo, err := c.readByte()
459 if err != nil {
460 return "", err
461 }
462 n = (n&^0x80)*0x100 + int(lo)
463 }
464 if n >= max {
465 return "", fmt.Errorf("rsync: over-long vstring received (%d > %d)", n, max-1)
466 }
467 buf := make([]byte, n)
468 if err := c.readFull(buf); err != nil {
469 return "", err
470 }
471 return string(buf), nil
472}
473
474// readNdx reads a file-list index, which protocol 30 delta-encodes.
475func (c *conn) readNdx() (int32, error) {
476 if c.protocol < 30 {
477 return c.readInt32()
478 }
479 b0, err := c.readByte()
480 if err != nil {
481 return 0, err
482 }
483 prev := &c.ndxIn.prevPositive
484 switch b0 {
485 case 0:
486 return ndxDone, nil
487 case 0xFF:
488 prev = &c.ndxIn.prevNegative
489 if b0, err = c.readByte(); err != nil {
490 return 0, err
491 }
492 }
493 var unum uint32
494 if b0 == 0xFE {
495 var b [4]byte
496 if err := c.readFull(b[:2]); err != nil {
497 return 0, err
498 }
499 if b[0]&0x80 != 0 {
500 b[3] = b[0] &^ 0x80
501 b[0] = b[1]
502 if err := c.readFull(b[1:3]); err != nil {
503 return 0, err
504 }
505 unum = binary.LittleEndian.Uint32(b[:])
506 } else {
507 unum = uint32(b[0])<<8 + uint32(b[1]) + uint32(*prev)
508 }
509 } else {
510 unum = uint32(b0) + uint32(*prev)
511 }
512 if unum > 0x7FFFFFFF {
513 return 0, fmt.Errorf("rsync: invalid file index %d", unum)
514 }
515 num := int32(unum)
516 *prev = num
517 if prev == &c.ndxIn.prevNegative {
518 num = -num
519 }
520 return num, nil
521}
522
523func (c *conn) write(p []byte) error {
524 _, err := c.w.Write(p)
525 return err
526}
527
528func (c *conn) writeByte(b byte) error {
529 return c.write([]byte{b})
530}
531
532func (c *conn) writeInt32(v int32) error {
533 var b [4]byte
534 binary.LittleEndian.PutUint32(b[:], uint32(v))
535 return c.write(b[:])
536}
537
538func (c *conn) writeShortint(v uint16) error {
539 var b [2]byte
540 binary.LittleEndian.PutUint16(b[:], v)
541 return c.write(b[:])
542}
543
544func (c *conn) writeLongint(v int64) error {
545 if v >= 0 && v <= 0x7FFFFFFF {
546 return c.writeInt32(int32(v))
547 }
548 var b [12]byte
549 binary.LittleEndian.PutUint32(b[:4], 0xFFFFFFFF)
550 binary.LittleEndian.PutUint64(b[4:], uint64(v))
551 return c.write(b[:])
552}
553
554func appendVarint(dst []byte, x int32) []byte {
555 var b [5]byte
556 binary.LittleEndian.PutUint32(b[1:], uint32(x))
557 cnt := 4
558 for cnt > 1 && b[cnt] == 0 {
559 cnt--
560 }
561 bit := byte(1) << (7 - cnt + 1)
562 switch {
563 case b[cnt] >= bit:
564 cnt++
565 b[0] = ^(bit - 1)
566 case cnt > 1:
567 b[0] = b[cnt] | ^(bit*2 - 1)
568 default:
569 b[0] = b[1]
570 }
571 return append(dst, b[:cnt]...)
572}
573
574func appendVarlong(dst []byte, x int64, minBytes int) []byte {
575 var b [9]byte
576 binary.LittleEndian.PutUint64(b[1:], uint64(x))
577 cnt := 8
578 for cnt > minBytes && b[cnt] == 0 {
579 cnt--
580 }
581 bit := byte(1) << (7 - cnt + minBytes)
582 switch {
583 case b[cnt] >= bit:
584 cnt++
585 b[0] = ^(bit - 1)
586 case cnt > minBytes:
587 b[0] = b[cnt] | ^(bit*2 - 1)
588 default:
589 b[0] = b[cnt]
590 }
591 return append(dst, b[:cnt]...)
592}
593
594func (c *conn) writeVarint(v int32) error {
595 return c.write(appendVarint(c.wbuf[:0], v))
596}
597
598func (c *conn) writeVarlong(v int64, minBytes int) error {
599 return c.write(appendVarlong(c.wbuf[:0], v, minBytes))
600}
601
602func (c *conn) writeVarint30(v int32) error {
603 if c.protocol < 30 {
604 return c.writeInt32(v)
605 }
606 return c.writeVarint(v)
607}
608
609func (c *conn) writeVarlong30(v int64, minBytes int) error {
610 if c.protocol < 30 {
611 return c.writeLongint(v)
612 }
613 return c.writeVarlong(v, minBytes)
614}
615
616func (c *conn) writeVstring(s string) error {
617 n := len(s)
618 if n > 0x7FFF {
619 return fmt.Errorf("rsync: attempting to send over-long vstring (%d > %d)", n, 0x7FFF)
620 }
621 var hdr []byte
622 if n > 0x7F {
623 hdr = []byte{byte(n/0x100 + 0x80), byte(n)}
624 } else {
625 hdr = []byte{byte(n)}
626 }
627 if err := c.write(hdr); err != nil {
628 return err
629 }
630 return c.write([]byte(s))
631}
632
633func (c *conn) writeNdx(ndx int32) error {
634 if c.protocol < 30 {
635 return c.writeInt32(ndx)
636 }
637 var b [6]byte
638 cnt := 0
639 var diff int32
640 switch {
641 case ndx >= 0:
642 diff = ndx - c.ndxOut.prevPositive
643 c.ndxOut.prevPositive = ndx
644 case ndx == ndxDone:
645 return c.writeByte(0)
646 default:
647 b[cnt] = 0xFF
648 cnt++
649 ndx = -ndx
650 diff = ndx - c.ndxOut.prevNegative
651 c.ndxOut.prevNegative = ndx
652 }
653 switch {
654 case diff > 0 && diff < 0xFE:
655 b[cnt] = byte(diff)
656 cnt++
657 case diff < 0 || diff > 0x7FFF:
658 b[cnt] = 0xFE
659 b[cnt+1] = byte(ndx>>24) | 0x80
660 b[cnt+2] = byte(ndx)
661 b[cnt+3] = byte(ndx >> 8)
662 b[cnt+4] = byte(ndx >> 16)
663 cnt += 5
664 default:
665 b[cnt] = 0xFE
666 b[cnt+1] = byte(diff >> 8)
667 b[cnt+2] = byte(diff)
668 cnt += 3
669 }
670 return c.write(b[:cnt])
671}
672
673func (c *conn) writeMsg(tag byte, data []byte) error {
674 return c.w.WriteMsg(tag, data)
675}
676
677func (c *conn) writeMsgInt(tag byte, v int32) error {
678 var b [4]byte
679 binary.LittleEndian.PutUint32(b[:], uint32(v))
680 return c.w.WriteMsg(tag, b[:])
681}