Cloak/internal/multiplex/stream.go

220 lines
5.1 KiB
Go
Raw Normal View History

2018-10-05 22:44:20 +00:00
package multiplex
import (
"errors"
2019-08-14 09:04:27 +00:00
"io"
2019-07-23 10:06:49 +00:00
"net"
"time"
2019-08-02 14:45:33 +00:00
log "github.com/sirupsen/logrus"
2018-10-05 22:44:20 +00:00
"sync"
2018-10-23 19:47:58 +00:00
"sync/atomic"
2018-10-05 22:44:20 +00:00
)
var ErrBrokenStream = errors.New("broken stream")
2018-10-05 22:44:20 +00:00
type Stream struct {
id uint32
session *Session
recvBuf recvBuffer
2019-08-05 12:32:53 +00:00
nextSendSeq uint64
2018-10-07 17:09:45 +00:00
2020-04-12 00:34:21 +00:00
writingM sync.Mutex
2018-11-23 23:57:35 +00:00
2019-08-20 21:43:04 +00:00
// atomic
2019-07-28 10:58:45 +00:00
closed uint32
2019-08-04 09:38:49 +00:00
2020-04-11 23:49:49 +00:00
// only alloc when writing to the stream
allocIdempot sync.Once
obfsBuf []byte
// we assign each stream a fixed underlying TCP connection to utilise order guarantee provided by TCP itself
// so that frameSorter should have few to none ooo frames to deal with
// overall the streams in a session should be uniformly distributed across all connections
2019-08-20 21:43:04 +00:00
// This is not used in unordered connection mode
assignedConnId uint32
rfTimeout time.Duration
2018-10-05 22:44:20 +00:00
}
2020-04-12 11:51:00 +00:00
func makeStream(sesh *Session, id uint32) *Stream {
var recvBuf recvBuffer
2019-08-14 09:04:27 +00:00
if sesh.Unordered {
recvBuf = NewDatagramBuffer()
2019-08-14 09:04:27 +00:00
} else {
recvBuf = NewStreamBuffer()
2019-08-14 09:04:27 +00:00
}
2019-08-05 12:32:53 +00:00
2018-10-05 22:44:20 +00:00
stream := &Stream{
2020-04-12 11:51:00 +00:00
id: id,
session: sesh,
recvBuf: recvBuf,
2018-10-05 22:44:20 +00:00
}
2019-08-05 12:32:53 +00:00
2018-10-05 22:44:20 +00:00
return stream
}
2019-07-28 10:58:45 +00:00
func (s *Stream) isClosed() bool { return atomic.LoadUint32(&s.closed) == 1 }
2019-08-30 19:43:04 +00:00
func (s *Stream) writeFrame(frame Frame) error {
toBeClosed, err := s.recvBuf.Write(frame)
if toBeClosed {
err = s.passiveClose()
if errors.Is(err, errRepeatStreamClosing) {
log.Debug(err)
return nil
}
return err
}
return err
}
2019-08-05 13:33:20 +00:00
2019-08-20 21:43:04 +00:00
// Read implements io.Read
2019-07-28 10:58:45 +00:00
func (s *Stream) Read(buf []byte) (n int, err error) {
2019-08-06 10:19:47 +00:00
//log.Tracef("attempting to read from stream %v", s.id)
2018-10-16 20:13:19 +00:00
if len(buf) == 0 {
2020-04-08 17:16:54 +00:00
return 0, nil
2018-10-05 22:44:20 +00:00
}
2019-08-14 09:04:27 +00:00
2019-08-30 19:39:23 +00:00
n, err = s.recvBuf.Read(buf)
2020-04-13 15:39:19 +00:00
log.Tracef("%v read from stream %v with err %v", n, s.id, err)
2019-08-30 19:39:23 +00:00
if err == io.EOF {
return n, ErrBrokenStream
2018-10-16 20:13:19 +00:00
}
2019-08-30 19:39:23 +00:00
return
2020-04-12 11:43:24 +00:00
}
2019-08-30 19:39:23 +00:00
2020-04-12 11:43:24 +00:00
func (s *Stream) WriteTo(w io.Writer) (int64, error) {
// will keep writing until the underlying buffer is closed
2020-04-12 15:34:49 +00:00
n, err := s.recvBuf.WriteTo(w)
2020-04-12 22:01:30 +00:00
log.Tracef("%v read from stream %v with err %v", n, s.id, err)
2020-04-12 15:34:49 +00:00
if err == io.EOF {
return n, ErrBrokenStream
}
return n, nil
2018-10-05 22:44:20 +00:00
}
2020-04-13 21:48:28 +00:00
func (s *Stream) sendFrame(f *Frame, framePayloadOffset int) error {
2020-04-12 22:01:30 +00:00
var cipherTextLen int
2020-04-13 21:48:28 +00:00
cipherTextLen, err := s.session.Obfs(f, s.obfsBuf, framePayloadOffset)
2020-04-12 22:01:30 +00:00
if err != nil {
return err
}
_, err = s.session.sb.send(s.obfsBuf[:cipherTextLen], &s.assignedConnId)
2020-04-13 15:39:19 +00:00
log.Tracef("%v sent to remote through stream %v with err %v. seq: %v", len(f.Payload), s.id, err, f.Seq)
2020-04-12 22:01:30 +00:00
if err != nil {
if err == errBrokenSwitchboard {
s.session.SetTerminalMsg(err.Error())
s.session.passiveClose()
}
return err
}
return nil
}
2019-08-20 21:43:04 +00:00
// Write implements io.Write
2019-07-28 10:58:45 +00:00
func (s *Stream) Write(in []byte) (n int, err error) {
2020-04-12 00:34:21 +00:00
s.writingM.Lock()
defer s.writingM.Unlock()
2019-07-28 10:58:45 +00:00
if s.isClosed() {
return 0, ErrBrokenStream
2018-10-05 22:44:20 +00:00
}
2020-04-12 00:34:21 +00:00
if s.obfsBuf == nil {
s.obfsBuf = make([]byte, s.session.SendBufferSize)
}
for n < len(in) {
var framePayload []byte
if len(in)-n <= s.session.maxStreamUnitWrite {
framePayload = in[n:]
} else {
2020-04-11 21:37:15 +00:00
if s.session.Unordered { // no splitting
err = io.ErrShortBuffer
return
}
framePayload = in[n : s.session.maxStreamUnitWrite+n]
}
2020-04-13 15:39:19 +00:00
f := &Frame{
StreamID: s.id,
Seq: s.nextSendSeq,
Closing: C_NOOP,
Payload: framePayload,
}
s.nextSendSeq++
2020-04-13 21:48:28 +00:00
err = s.sendFrame(f, 0)
if err != nil {
2020-04-12 22:01:30 +00:00
return
2019-10-08 22:11:16 +00:00
}
2020-04-12 22:01:30 +00:00
n += len(framePayload)
}
return
}
2018-10-05 22:44:20 +00:00
2020-04-12 22:01:30 +00:00
func (s *Stream) ReadFrom(r io.Reader) (n int64, err error) {
if s.obfsBuf == nil {
s.obfsBuf = make([]byte, s.session.SendBufferSize)
}
for {
if s.rfTimeout != 0 {
if rder, ok := r.(net.Conn); !ok {
log.Warn("ReadFrom timeout is set but reader doesn't implement SetReadDeadline")
} else {
rder.SetReadDeadline(time.Now().Add(s.rfTimeout))
}
}
2020-04-12 22:01:30 +00:00
read, er := r.Read(s.obfsBuf[HEADER_LEN : HEADER_LEN+s.session.maxStreamUnitWrite])
if er != nil {
return n, er
}
if s.isClosed() {
2020-04-13 15:39:19 +00:00
return n, ErrBrokenStream
}
s.writingM.Lock()
f := &Frame{
StreamID: s.id,
Seq: s.nextSendSeq,
Closing: C_NOOP,
Payload: s.obfsBuf[HEADER_LEN : HEADER_LEN+read],
2020-04-12 22:01:30 +00:00
}
2020-04-13 15:39:19 +00:00
s.nextSendSeq++
2020-04-13 21:48:28 +00:00
err = s.sendFrame(f, HEADER_LEN)
2020-04-13 15:39:19 +00:00
s.writingM.Unlock()
if err != nil {
return
}
2020-04-12 22:01:30 +00:00
n += int64(read)
}
2018-10-05 22:44:20 +00:00
}
func (s *Stream) passiveClose() error {
return s.session.closeStream(s, false)
2018-10-27 22:35:46 +00:00
}
// active close. Close locally and tell the remote that this stream is being closed
2019-07-28 10:58:45 +00:00
func (s *Stream) Close() error {
2020-04-13 15:39:19 +00:00
s.writingM.Lock()
defer s.writingM.Unlock()
return s.session.closeStream(s, true)
2018-10-23 19:47:58 +00:00
}
2019-07-23 10:06:49 +00:00
// the following functions are purely for implementing net.Conn interface.
// they are not used
var errNotImplemented = errors.New("Not implemented")
2019-07-28 10:58:45 +00:00
func (s *Stream) LocalAddr() net.Addr { return s.session.addrs.Load().([]net.Addr)[0] }
func (s *Stream) RemoteAddr() net.Addr { return s.session.addrs.Load().([]net.Addr)[1] }
2019-07-23 10:06:49 +00:00
// TODO: implement the following
2019-07-28 10:58:45 +00:00
func (s *Stream) SetDeadline(t time.Time) error { return errNotImplemented }
func (s *Stream) SetWriteToTimeout(d time.Duration) { s.recvBuf.SetWriteToTimeout(d) }
2020-04-09 17:56:17 +00:00
func (s *Stream) SetReadDeadline(t time.Time) error { s.recvBuf.SetReadDeadline(t); return nil }
func (s *Stream) SetReadFromTimeout(d time.Duration) { s.rfTimeout = d }
2019-07-28 10:58:45 +00:00
func (s *Stream) SetWriteDeadline(t time.Time) error { return errNotImplemented }