1 package p9p
2 
3 import (
4 	"bufio"
5 	"encoding/binary"
6 	"fmt"
7 	"io"
8 	"io/ioutil"
9 	"log"
10 	"net"
11 	"time"
12 
13 	"context"
14 )
15 
16 // Channel defines the operations necessary to implement a 9p message channel
17 // interface. Typically, message channels do no protocol processing except to
18 // send and receive message frames.
19 type Channel interface {
20 	// ReadFcall reads one fcall frame into the provided fcall structure. The
21 	// Fcall may be cleared whether there is an error or not. If the operation
22 	// is successful, the contents of the fcall will be populated in the
23 	// argument. ReadFcall cannot be called concurrently with other calls to
24 	// ReadFcall. This both to preserve message ordering and to allow lockless
25 	// buffer reusage.
26 	ReadFcall(ctx context.Context, fcall *Fcall) error
27 
28 	// WriteFcall writes the provided fcall to the channel. WriteFcall cannot
29 	// be called concurrently with other calls to WriteFcall.
30 	WriteFcall(ctx context.Context, fcall *Fcall) error
31 
32 	// MSize returns the current msize for the channel.
33 	MSize() int
34 
35 	// SetMSize sets the maximum message size for the channel. This must never
36 	// be called currently with ReadFcall or WriteFcall.
37 	SetMSize(msize int)
38 }
39 
40 // NewChannel returns a new channel to read and write Fcalls with the provided
41 // connection and message size.
42 func NewChannel(conn net.Conn, msize int) Channel {
43 	return newChannel(conn, codec9p{}, msize)
44 }
45 
46 const (
47 	defaultRWTimeout = 30 * time.Second // default read/write timeout if not set in context
48 )
49 
50 // channel provides bidirectional protocol framing for 9p over net.Conn.
51 // Operations are not thread-safe but reads and writes may be carried out
52 // concurrently, supporting separate read and write loops.
53 //
54 // Lifecyle
55 //
56 // A connection, or message channel abstraction, has a lifecycle delineated by
57 // Tversion/Rversion request response cycles. For now, this is part of the
58 // channel itself but doesn't necessarily influence the channels state, except
59 // the msize. Visually, it might look something like this:
60 //
61 // 	[Established] -> [Version] -> [Session] -> [Version]---+
62 //	                     ^                                 |
63 // 	                     |_________________________________|
64 //
65 // The connection is established, then we negotiate a version, run a session,
66 // then negotiate a version and so on. For most purposes, we are likely going
67 // to terminate the connection after the session but we may want to support
68 // connection pooling. Pooling may result in possible security leaks if the
69 // connections are shared among contexts, since the version is negotiated at
70 // the start of the session. To avoid this, we can actually use a "tombstone"
71 // version message which clears the server's session state without starting a
72 // new session. The next version message would then prepare the session
73 // without leaking any Fid's.
74 type channel struct {
75 	conn   net.Conn
76 	codec  Codec
77 	brd    *bufio.Reader
78 	bwr    *bufio.Writer
79 	closed chan struct{}
80 	msize  int
81 	rdbuf  []byte
82 }
83 
84 func newChannel(conn net.Conn, codec Codec, msize int) *channel {
85 	return &channel{
86 		conn:   conn,
87 		codec:  codec,
88 		brd:    bufio.NewReaderSize(conn, msize), // msize may not be optimal buffer size
89 		bwr:    bufio.NewWriterSize(conn, msize),
90 		closed: make(chan struct{}),
91 		msize:  msize,
92 		rdbuf:  make([]byte, msize),
93 	}
94 }
95 
96 func (ch *channel) MSize() int {
97 	return ch.msize
98 }
99 
100 // setmsize resizes the buffers for use with a separate msize. This call must
101 // be protected by a mutex or made before passing to other goroutines.
102 func (ch *channel) SetMSize(msize int) {
103 	// NOTE(stevvooe): We cannot safely resize the buffered reader and writer.
104 	// Proceed assuming that original size is sufficient.
105 
106 	ch.msize = msize
107 	if msize < len(ch.rdbuf) {
108 		// just change the cap
109 		ch.rdbuf = ch.rdbuf[:msize]
110 		return
111 	}
112 
113 	ch.rdbuf = make([]byte, msize)
114 }
115 
116 // ReadFcall reads the next message from the channel into fcall.
117 func (ch *channel) ReadFcall(ctx context.Context, fcall *Fcall) error {
118 	select {
119 	case <-ctx.Done():
120 		return ctx.Err()
121 	case <-ch.closed:
122 		return ErrClosed
123 	default:
124 	}
125 
126 	deadline, ok := ctx.Deadline()
127 	if !ok {
128 		deadline = time.Now().Add(defaultRWTimeout)
129 	}
130 
131 	if err := ch.conn.SetReadDeadline(deadline); err != nil {
132 		log.Printf("transport: error setting read deadline on %v: %v", ch.conn.RemoteAddr(), err)
133 	}
134 
135 	n, err := readmsg(ch.brd, ch.rdbuf)
136 	if err != nil {
137 		// TODO(stevvooe): There may be more we can do here to detect partial
138 		// reads. For now, we just propagate the error untouched.
139 		return err
140 	}
141 
142 	if n > len(ch.rdbuf) {
143 		// TODO(stevvooe): Make this error detectable and respond with error
144 		// message.
145 		return fmt.Errorf("message too large for buffer: %v > %v ", n, len(ch.rdbuf))
146 	}
147 
148 	// clear out the fcall
149 	*fcall = Fcall{}
150 	if err := ch.codec.Unmarshal(ch.rdbuf[:n], fcall); err != nil {
151 		return err
152 	}
153 
154 	return nil
155 }
156 
157 func (ch *channel) WriteFcall(ctx context.Context, fcall *Fcall) error {
158 	select {
159 	case <-ctx.Done():
160 		return ctx.Err()
161 	case <-ch.closed:
162 		return ErrClosed
163 	default:
164 	}
165 
166 	deadline, ok := ctx.Deadline()
167 	if !ok {
168 		deadline = time.Now().Add(defaultRWTimeout)
169 	}
170 
171 	if err := ch.conn.SetWriteDeadline(deadline); err != nil {
172 		log.Printf("transport: error setting read deadline on %v: %v", ch.conn.RemoteAddr(), err)
173 	}
174 
175 	p, err := ch.codec.Marshal(fcall)
176 	if err != nil {
177 		return err
178 	}
179 
180 	if err := sendmsg(ch.bwr, p); err != nil {
181 		return err
182 	}
183 
184 	return ch.bwr.Flush()
185 }
186 
187 // readmsg reads a 9p message into p from rd, ensuring that all bytes are
188 // consumed from the size header. If the size header indicates the message is
189 // larger than p, the entire message will be discarded, leaving a truncated
190 // portion in p. Any error should be treated as a framing error unless n is
191 // zero. The caller must check that n is less than or equal to len(p) to
192 // ensure that a valid message has been read.
193 func readmsg(rd io.Reader, p []byte) (n int, err error) {
194 	var msize uint32
195 
196 	if err := binary.Read(rd, binary.LittleEndian, &msize); err != nil {
197 		return 0, err
198 	}
199 
200 	n += binary.Size(msize)
201 	mbody := int(msize) - 4
202 
203 	if mbody < len(p) {
204 		p = p[:mbody]
205 	}
206 
207 	np, err := io.ReadFull(rd, p)
208 	if err != nil {
209 		return np + n, err
210 	}
211 	n += np
212 
213 	if mbody > len(p) {
214 		// message has been read up to len(p) but we must consume the entire
215 		// message. This is an error condition but is non-fatal if we can
216 		// consume msize bytes.
217 		nn, err := io.CopyN(ioutil.Discard, rd, int64(mbody-len(p)))
218 		n += int(nn)
219 		if err != nil {
220 			return n, err
221 		}
222 	}
223 
224 	return n, nil
225 }
226 
227 // sendmsg writes a message of len(p) to wr with a 9p size header. All errors
228 // should be considered terminal.
229 func sendmsg(wr io.Writer, p []byte) error {
230 	size := uint32(len(p) + 4) // message size plus 4-bytes for size.
231 	if err := binary.Write(wr, binary.LittleEndian, size); err != nil {
232 		return err
233 	}
234 
235 	// This assume partial writes to wr aren't possible. Not sure if this
236 	// valid. Matters during timeout retries.
237 	if n, err := wr.Write(p); err != nil {
238 		return err
239 	} else if n < len(p) {
240 		return io.ErrShortWrite
241 	}
242 
243 	return nil
244 }