session.go 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287
  1. package quic
  2. import (
  3. "bytes"
  4. "context"
  5. "crypto/tls"
  6. "errors"
  7. "fmt"
  8. "io"
  9. "net"
  10. "reflect"
  11. "sync"
  12. "time"
  13. "github.com/lucas-clemente/quic-go/internal/ackhandler"
  14. "github.com/lucas-clemente/quic-go/internal/congestion"
  15. "github.com/lucas-clemente/quic-go/internal/flowcontrol"
  16. "github.com/lucas-clemente/quic-go/internal/handshake"
  17. "github.com/lucas-clemente/quic-go/internal/protocol"
  18. "github.com/lucas-clemente/quic-go/internal/qerr"
  19. "github.com/lucas-clemente/quic-go/internal/utils"
  20. "github.com/lucas-clemente/quic-go/internal/wire"
  21. )
  22. type unpacker interface {
  23. Unpack(hdr *wire.Header, data []byte) (*unpackedPacket, error)
  24. }
  25. type streamGetter interface {
  26. GetOrOpenReceiveStream(protocol.StreamID) (receiveStreamI, error)
  27. GetOrOpenSendStream(protocol.StreamID) (sendStreamI, error)
  28. }
  29. type streamManager interface {
  30. GetOrOpenSendStream(protocol.StreamID) (sendStreamI, error)
  31. GetOrOpenReceiveStream(protocol.StreamID) (receiveStreamI, error)
  32. OpenStream() (Stream, error)
  33. OpenUniStream() (SendStream, error)
  34. OpenStreamSync() (Stream, error)
  35. OpenUniStreamSync() (SendStream, error)
  36. AcceptStream() (Stream, error)
  37. AcceptUniStream() (ReceiveStream, error)
  38. DeleteStream(protocol.StreamID) error
  39. UpdateLimits(*handshake.TransportParameters) error
  40. HandleMaxStreamsFrame(*wire.MaxStreamsFrame) error
  41. CloseWithError(error)
  42. }
  43. type cryptoStreamHandler interface {
  44. RunHandshake() error
  45. ChangeConnectionID(protocol.ConnectionID) error
  46. io.Closer
  47. ConnectionState() tls.ConnectionState
  48. }
  49. type receivedPacket struct {
  50. remoteAddr net.Addr
  51. rcvTime time.Time
  52. data []byte
  53. buffer *packetBuffer
  54. }
  55. func (p *receivedPacket) Clone() *receivedPacket {
  56. return &receivedPacket{
  57. remoteAddr: p.remoteAddr,
  58. rcvTime: p.rcvTime,
  59. data: p.data,
  60. buffer: p.buffer,
  61. }
  62. }
  63. type closeError struct {
  64. err error
  65. remote bool
  66. sendClose bool
  67. }
  68. var errCloseForRecreating = errors.New("closing session in order to recreate it")
  69. // A Session is a QUIC session
  70. type session struct {
  71. sessionRunner sessionRunner
  72. destConnID protocol.ConnectionID
  73. origDestConnID protocol.ConnectionID // if the server sends a Retry, this is the connection ID we used initially
  74. srcConnID protocol.ConnectionID
  75. perspective protocol.Perspective
  76. initialVersion protocol.VersionNumber // if version negotiation is performed, this is the version we initially tried
  77. version protocol.VersionNumber
  78. config *Config
  79. conn connection
  80. streamsMap streamManager
  81. rttStats *congestion.RTTStats
  82. cryptoStreamManager *cryptoStreamManager
  83. sentPacketHandler ackhandler.SentPacketHandler
  84. receivedPacketHandler ackhandler.ReceivedPacketHandler
  85. framer framer
  86. windowUpdateQueue *windowUpdateQueue
  87. connFlowController flowcontrol.ConnectionFlowController
  88. unpacker unpacker
  89. frameParser wire.FrameParser
  90. packer packer
  91. cryptoStreamHandler cryptoStreamHandler
  92. receivedPackets chan *receivedPacket
  93. sendingScheduled chan struct{}
  94. closeOnce sync.Once
  95. closed utils.AtomicBool
  96. // closeChan is used to notify the run loop that it should terminate
  97. closeChan chan closeError
  98. connectionClosePacket *packedPacket
  99. packetsReceivedAfterClose int
  100. ctx context.Context
  101. ctxCancel context.CancelFunc
  102. undecryptablePackets []*receivedPacket
  103. clientHelloWritten <-chan struct{}
  104. handshakeCompleteChan chan struct{} // is closed when the handshake completes
  105. handshakeComplete bool
  106. receivedRetry bool
  107. receivedFirstPacket bool
  108. receivedFirstForwardSecurePacket bool
  109. sessionCreationTime time.Time
  110. // The idle timeout is set based on the max of the time we received the last packet...
  111. lastPacketReceivedTime time.Time
  112. // ... and the time we sent a new ack-eliciting packet after receiving a packet.
  113. firstAckElicitingPacketAfterIdleSentTime time.Time
  114. // pacingDeadline is the time when the next packet should be sent
  115. pacingDeadline time.Time
  116. peerParams *handshake.TransportParameters
  117. timer *utils.Timer
  118. // keepAlivePingSent stores whether a Ping frame was sent to the peer or not
  119. // it is reset as soon as we receive a packet from the peer
  120. keepAlivePingSent bool
  121. logger utils.Logger
  122. }
  123. var _ Session = &session{}
  124. var _ streamSender = &session{}
  125. var newSession = func(
  126. conn connection,
  127. runner sessionRunner,
  128. clientDestConnID protocol.ConnectionID,
  129. destConnID protocol.ConnectionID,
  130. srcConnID protocol.ConnectionID,
  131. conf *Config,
  132. tlsConf *tls.Config,
  133. params *handshake.TransportParameters,
  134. logger utils.Logger,
  135. v protocol.VersionNumber,
  136. ) (quicSession, error) {
  137. s := &session{
  138. conn: conn,
  139. sessionRunner: runner,
  140. config: conf,
  141. srcConnID: srcConnID,
  142. destConnID: destConnID,
  143. perspective: protocol.PerspectiveServer,
  144. handshakeCompleteChan: make(chan struct{}),
  145. logger: logger,
  146. version: v,
  147. }
  148. s.preSetup()
  149. s.sentPacketHandler = ackhandler.NewSentPacketHandler(0, s.rttStats, s.logger)
  150. s.streamsMap = newStreamsMap(
  151. s,
  152. s.newFlowController,
  153. uint64(s.config.MaxIncomingStreams),
  154. uint64(s.config.MaxIncomingUniStreams),
  155. s.perspective,
  156. s.version,
  157. )
  158. s.framer = newFramer(s.streamsMap, s.version)
  159. initialStream := newCryptoStream()
  160. handshakeStream := newCryptoStream()
  161. oneRTTStream := newPostHandshakeCryptoStream(s.framer)
  162. cs, err := handshake.NewCryptoSetupServer(
  163. initialStream,
  164. handshakeStream,
  165. oneRTTStream,
  166. clientDestConnID,
  167. conn.RemoteAddr(),
  168. params,
  169. s.processTransportParameters,
  170. tlsConf,
  171. logger,
  172. )
  173. if err != nil {
  174. return nil, err
  175. }
  176. s.cryptoStreamHandler = cs
  177. s.packer = newPacketPacker(
  178. s.destConnID,
  179. s.srcConnID,
  180. initialStream,
  181. handshakeStream,
  182. s.sentPacketHandler,
  183. s.RemoteAddr(),
  184. cs,
  185. s.framer,
  186. s.receivedPacketHandler,
  187. s.perspective,
  188. s.version,
  189. )
  190. s.cryptoStreamManager = newCryptoStreamManager(cs, initialStream, handshakeStream, oneRTTStream)
  191. if err := s.postSetup(); err != nil {
  192. return nil, err
  193. }
  194. s.unpacker = newPacketUnpacker(cs, s.version)
  195. return s, nil
  196. }
  197. // declare this as a variable, such that we can it mock it in the tests
  198. var newClientSession = func(
  199. conn connection,
  200. runner sessionRunner,
  201. destConnID protocol.ConnectionID,
  202. srcConnID protocol.ConnectionID,
  203. conf *Config,
  204. tlsConf *tls.Config,
  205. initialPacketNumber protocol.PacketNumber,
  206. params *handshake.TransportParameters,
  207. initialVersion protocol.VersionNumber,
  208. logger utils.Logger,
  209. v protocol.VersionNumber,
  210. ) (quicSession, error) {
  211. s := &session{
  212. conn: conn,
  213. sessionRunner: runner,
  214. config: conf,
  215. srcConnID: srcConnID,
  216. destConnID: destConnID,
  217. perspective: protocol.PerspectiveClient,
  218. handshakeCompleteChan: make(chan struct{}),
  219. logger: logger,
  220. initialVersion: initialVersion,
  221. version: v,
  222. }
  223. s.preSetup()
  224. s.sentPacketHandler = ackhandler.NewSentPacketHandler(initialPacketNumber, s.rttStats, s.logger)
  225. initialStream := newCryptoStream()
  226. handshakeStream := newCryptoStream()
  227. oneRTTStream := newPostHandshakeCryptoStream(s.framer)
  228. cs, clientHelloWritten, err := handshake.NewCryptoSetupClient(
  229. initialStream,
  230. handshakeStream,
  231. oneRTTStream,
  232. s.destConnID,
  233. conn.RemoteAddr(),
  234. params,
  235. s.processTransportParameters,
  236. tlsConf,
  237. logger,
  238. )
  239. if err != nil {
  240. return nil, err
  241. }
  242. s.clientHelloWritten = clientHelloWritten
  243. s.cryptoStreamHandler = cs
  244. s.cryptoStreamManager = newCryptoStreamManager(cs, initialStream, handshakeStream, oneRTTStream)
  245. s.unpacker = newPacketUnpacker(cs, s.version)
  246. s.streamsMap = newStreamsMap(
  247. s,
  248. s.newFlowController,
  249. uint64(s.config.MaxIncomingStreams),
  250. uint64(s.config.MaxIncomingUniStreams),
  251. s.perspective,
  252. s.version,
  253. )
  254. s.framer = newFramer(s.streamsMap, s.version)
  255. s.packer = newPacketPacker(
  256. s.destConnID,
  257. s.srcConnID,
  258. initialStream,
  259. handshakeStream,
  260. s.sentPacketHandler,
  261. s.RemoteAddr(),
  262. cs,
  263. s.framer,
  264. s.receivedPacketHandler,
  265. s.perspective,
  266. s.version,
  267. )
  268. return s, s.postSetup()
  269. }
  270. func (s *session) preSetup() {
  271. s.frameParser = wire.NewFrameParser(s.version)
  272. s.rttStats = &congestion.RTTStats{}
  273. s.receivedPacketHandler = ackhandler.NewReceivedPacketHandler(s.rttStats, s.logger, s.version)
  274. s.connFlowController = flowcontrol.NewConnectionFlowController(
  275. protocol.InitialMaxData,
  276. protocol.ByteCount(s.config.MaxReceiveConnectionFlowControlWindow),
  277. s.onHasConnectionWindowUpdate,
  278. s.rttStats,
  279. s.logger,
  280. )
  281. }
  282. func (s *session) postSetup() error {
  283. s.receivedPackets = make(chan *receivedPacket, protocol.MaxSessionUnprocessedPackets)
  284. s.closeChan = make(chan closeError, 1)
  285. s.sendingScheduled = make(chan struct{}, 1)
  286. s.undecryptablePackets = make([]*receivedPacket, 0, protocol.MaxUndecryptablePackets)
  287. s.ctx, s.ctxCancel = context.WithCancel(context.Background())
  288. s.timer = utils.NewTimer()
  289. now := time.Now()
  290. s.lastPacketReceivedTime = now
  291. s.sessionCreationTime = now
  292. s.windowUpdateQueue = newWindowUpdateQueue(s.streamsMap, s.connFlowController, s.framer.QueueControlFrame)
  293. return nil
  294. }
  295. // run the session main loop
  296. func (s *session) run() error {
  297. defer s.ctxCancel()
  298. go func() {
  299. if err := s.cryptoStreamHandler.RunHandshake(); err != nil {
  300. s.closeLocal(err)
  301. return
  302. }
  303. close(s.handshakeCompleteChan)
  304. }()
  305. if s.perspective == protocol.PerspectiveClient {
  306. select {
  307. case <-s.clientHelloWritten:
  308. s.scheduleSending()
  309. case closeErr := <-s.closeChan:
  310. // put the close error back into the channel, so that the run loop can receive it
  311. s.closeChan <- closeErr
  312. }
  313. }
  314. var closeErr closeError
  315. runLoop:
  316. for {
  317. // Close immediately if requested
  318. select {
  319. case closeErr = <-s.closeChan:
  320. break runLoop
  321. case <-s.handshakeCompleteChan:
  322. s.handleHandshakeComplete()
  323. default:
  324. }
  325. s.maybeResetTimer()
  326. select {
  327. case closeErr = <-s.closeChan:
  328. break runLoop
  329. case <-s.timer.Chan():
  330. s.timer.SetRead()
  331. // We do all the interesting stuff after the switch statement, so
  332. // nothing to see here.
  333. case <-s.sendingScheduled:
  334. // We do all the interesting stuff after the switch statement, so
  335. // nothing to see here.
  336. case p := <-s.receivedPackets:
  337. // Only reset the timers if this packet was actually processed.
  338. // This avoids modifying any state when handling undecryptable packets,
  339. // which could be injected by an attacker.
  340. if wasProcessed := s.handlePacketImpl(p); !wasProcessed {
  341. continue
  342. }
  343. case <-s.handshakeCompleteChan:
  344. s.handleHandshakeComplete()
  345. }
  346. now := time.Now()
  347. if timeout := s.sentPacketHandler.GetAlarmTimeout(); !timeout.IsZero() && timeout.Before(now) {
  348. // This could cause packets to be retransmitted.
  349. // Check it before trying to send packets.
  350. if err := s.sentPacketHandler.OnAlarm(); err != nil {
  351. s.closeLocal(err)
  352. }
  353. }
  354. var pacingDeadline time.Time
  355. if s.pacingDeadline.IsZero() { // the timer didn't have a pacing deadline set
  356. pacingDeadline = s.sentPacketHandler.TimeUntilSend()
  357. }
  358. if s.config.KeepAlive && !s.keepAlivePingSent && s.handshakeComplete && s.firstAckElicitingPacketAfterIdleSentTime.IsZero() && time.Since(s.lastPacketReceivedTime) >= s.peerParams.IdleTimeout/2 {
  359. // send a PING frame since there is no activity in the session
  360. s.logger.Debugf("Sending a keep-alive ping to keep the connection alive.")
  361. s.framer.QueueControlFrame(&wire.PingFrame{})
  362. s.keepAlivePingSent = true
  363. } else if !pacingDeadline.IsZero() && now.Before(pacingDeadline) {
  364. // If we get to this point before the pacing deadline, we should wait until that deadline.
  365. // This can happen when scheduleSending is called, or a packet is received.
  366. // Set the timer and restart the run loop.
  367. s.pacingDeadline = pacingDeadline
  368. continue
  369. }
  370. if !s.handshakeComplete && now.Sub(s.sessionCreationTime) >= s.config.HandshakeTimeout {
  371. s.destroy(qerr.TimeoutError("Handshake did not complete in time"))
  372. continue
  373. }
  374. if s.handshakeComplete && now.Sub(s.idleTimeoutStartTime()) >= s.config.IdleTimeout {
  375. s.destroy(qerr.TimeoutError("No recent network activity"))
  376. continue
  377. }
  378. if err := s.sendPackets(); err != nil {
  379. s.closeLocal(err)
  380. }
  381. }
  382. s.handleCloseError(closeErr)
  383. s.closed.Set(true)
  384. s.logger.Infof("Connection %s closed.", s.srcConnID)
  385. s.cryptoStreamHandler.Close()
  386. return closeErr.err
  387. }
  388. func (s *session) Context() context.Context {
  389. return s.ctx
  390. }
  391. func (s *session) ConnectionState() tls.ConnectionState {
  392. return s.cryptoStreamHandler.ConnectionState()
  393. }
  394. func (s *session) maybeResetTimer() {
  395. var deadline time.Time
  396. if s.config.KeepAlive && s.handshakeComplete && !s.keepAlivePingSent {
  397. deadline = s.idleTimeoutStartTime().Add(s.peerParams.IdleTimeout / 2)
  398. } else {
  399. deadline = s.idleTimeoutStartTime().Add(s.config.IdleTimeout)
  400. }
  401. if ackAlarm := s.receivedPacketHandler.GetAlarmTimeout(); !ackAlarm.IsZero() {
  402. deadline = utils.MinTime(deadline, ackAlarm)
  403. }
  404. if lossTime := s.sentPacketHandler.GetAlarmTimeout(); !lossTime.IsZero() {
  405. deadline = utils.MinTime(deadline, lossTime)
  406. }
  407. if !s.handshakeComplete {
  408. handshakeDeadline := s.sessionCreationTime.Add(s.config.HandshakeTimeout)
  409. deadline = utils.MinTime(deadline, handshakeDeadline)
  410. }
  411. if !s.pacingDeadline.IsZero() {
  412. deadline = utils.MinTime(deadline, s.pacingDeadline)
  413. }
  414. s.timer.Reset(deadline)
  415. }
  416. func (s *session) idleTimeoutStartTime() time.Time {
  417. return utils.MaxTime(s.lastPacketReceivedTime, s.firstAckElicitingPacketAfterIdleSentTime)
  418. }
  419. func (s *session) handleHandshakeComplete() {
  420. s.handshakeComplete = true
  421. s.handshakeCompleteChan = nil // prevent this case from ever being selected again
  422. s.sessionRunner.OnHandshakeComplete(s)
  423. // The client completes the handshake first (after sending the CFIN).
  424. // We need to make sure they learn about the peer completing the handshake,
  425. // in order to stop retransmitting handshake packets.
  426. // They will stop retransmitting handshake packets when receiving the first forward-secure packet.
  427. // We need to make sure that an ack-eliciting 1-RTT packet is sent,
  428. // independent from the application protocol.
  429. if s.perspective == protocol.PerspectiveServer {
  430. s.queueControlFrame(&wire.PingFrame{})
  431. s.sentPacketHandler.SetHandshakeComplete()
  432. }
  433. }
  434. func (s *session) handlePacketImpl(rp *receivedPacket) bool {
  435. var counter uint8
  436. var lastConnID protocol.ConnectionID
  437. var processed bool
  438. data := rp.data
  439. p := rp
  440. for len(data) > 0 {
  441. if counter > 0 {
  442. p = p.Clone()
  443. p.data = data
  444. }
  445. hdr, packetData, rest, err := wire.ParsePacket(p.data, s.srcConnID.Len())
  446. if err != nil {
  447. s.logger.Debugf("error parsing packet: %s", err)
  448. break
  449. }
  450. if counter > 0 && !hdr.DestConnectionID.Equal(lastConnID) {
  451. s.logger.Debugf("coalesced packet has different destination connection ID: %s, expected %s", hdr.DestConnectionID, lastConnID)
  452. break
  453. }
  454. lastConnID = hdr.DestConnectionID
  455. if counter > 0 {
  456. p.buffer.Split()
  457. }
  458. counter++
  459. // only log if this actually a coalesced packet
  460. if s.logger.Debug() && (counter > 1 || len(rest) > 0) {
  461. s.logger.Debugf("Parsed a coalesced packet. Part %d: %d bytes. Remaining: %d bytes.", counter, len(packetData), len(rest))
  462. }
  463. p.data = packetData
  464. if wasProcessed := s.handleSinglePacket(p, hdr); wasProcessed {
  465. processed = true
  466. }
  467. data = rest
  468. }
  469. p.buffer.MaybeRelease()
  470. return processed
  471. }
  472. func (s *session) handleSinglePacket(p *receivedPacket, hdr *wire.Header) bool /* was the packet successfully processed */ {
  473. var wasQueued bool
  474. defer func() {
  475. // Put back the packet buffer if the packet wasn't queued for later decryption.
  476. if !wasQueued {
  477. p.buffer.Decrement()
  478. }
  479. }()
  480. if hdr.Type == protocol.PacketTypeRetry {
  481. return s.handleRetryPacket(p, hdr)
  482. }
  483. // The server can change the source connection ID with the first Handshake packet.
  484. // After this, all packets with a different source connection have to be ignored.
  485. if s.receivedFirstPacket && hdr.IsLongHeader && !hdr.SrcConnectionID.Equal(s.destConnID) {
  486. s.logger.Debugf("Dropping packet with unexpected source connection ID: %s (expected %s)", hdr.SrcConnectionID, s.destConnID)
  487. return false
  488. }
  489. // drop 0-RTT packets
  490. if hdr.Type == protocol.PacketType0RTT {
  491. return false
  492. }
  493. packet, err := s.unpacker.Unpack(hdr, p.data)
  494. if err != nil {
  495. if err == handshake.ErrOpenerNotYetAvailable {
  496. // Sealer for this encryption level not yet available.
  497. // Try again later.
  498. wasQueued = true
  499. s.tryQueueingUndecryptablePacket(p)
  500. return false
  501. }
  502. // This might be a packet injected by an attacker.
  503. // Drop it.
  504. s.logger.Debugf("Dropping packet that could not be unpacked. Unpack error: %s", err)
  505. return false
  506. }
  507. if s.logger.Debug() {
  508. s.logger.Debugf("<- Reading packet %#x (%d bytes) for connection %s, %s", packet.packetNumber, len(p.data), hdr.DestConnectionID, packet.encryptionLevel)
  509. packet.hdr.Log(s.logger)
  510. }
  511. if err := s.handleUnpackedPacket(packet, p.rcvTime); err != nil {
  512. s.closeLocal(err)
  513. return false
  514. }
  515. return true
  516. }
  517. func (s *session) handleRetryPacket(p *receivedPacket, hdr *wire.Header) bool /* was this a valid Retry */ {
  518. if s.perspective == protocol.PerspectiveServer {
  519. s.logger.Debugf("Ignoring Retry.")
  520. return false
  521. }
  522. if s.receivedFirstPacket {
  523. s.logger.Debugf("Ignoring Retry, since we already received a packet.")
  524. return false
  525. }
  526. (&wire.ExtendedHeader{Header: *hdr}).Log(s.logger)
  527. if !hdr.OrigDestConnectionID.Equal(s.destConnID) {
  528. s.logger.Debugf("Ignoring spoofed Retry. Original Destination Connection ID: %s, expected: %s", hdr.OrigDestConnectionID, s.destConnID)
  529. return false
  530. }
  531. if hdr.SrcConnectionID.Equal(s.destConnID) {
  532. s.logger.Debugf("Ignoring Retry, since the server didn't change the Source Connection ID.")
  533. return false
  534. }
  535. // If a token is already set, this means that we already received a Retry from the server.
  536. // Ignore this Retry packet.
  537. if s.receivedRetry {
  538. s.logger.Debugf("Ignoring Retry, since a Retry was already received.")
  539. return false
  540. }
  541. s.logger.Debugf("<- Received Retry")
  542. s.logger.Debugf("Switching destination connection ID to: %s", hdr.SrcConnectionID)
  543. s.origDestConnID = s.destConnID
  544. s.destConnID = hdr.SrcConnectionID
  545. s.receivedRetry = true
  546. if err := s.sentPacketHandler.ResetForRetry(); err != nil {
  547. s.closeLocal(err)
  548. return false
  549. }
  550. s.cryptoStreamHandler.ChangeConnectionID(s.destConnID)
  551. s.packer.SetToken(hdr.Token)
  552. s.packer.ChangeDestConnectionID(s.destConnID)
  553. s.scheduleSending()
  554. return true
  555. }
  556. func (s *session) handleUnpackedPacket(packet *unpackedPacket, rcvTime time.Time) error {
  557. if len(packet.data) == 0 {
  558. return qerr.Error(qerr.ProtocolViolation, "empty packet")
  559. }
  560. // The server can change the source connection ID with the first Handshake packet.
  561. if s.perspective == protocol.PerspectiveClient && !s.receivedFirstPacket && packet.hdr.IsLongHeader && !packet.hdr.SrcConnectionID.Equal(s.destConnID) {
  562. s.logger.Debugf("Received first packet. Switching destination connection ID to: %s", packet.hdr.SrcConnectionID)
  563. s.destConnID = packet.hdr.SrcConnectionID
  564. s.packer.ChangeDestConnectionID(s.destConnID)
  565. }
  566. s.receivedFirstPacket = true
  567. s.lastPacketReceivedTime = rcvTime
  568. s.firstAckElicitingPacketAfterIdleSentTime = time.Time{}
  569. s.keepAlivePingSent = false
  570. // The client completes the handshake first (after sending the CFIN).
  571. // We know that the server completed the handshake as soon as we receive a forward-secure packet.
  572. if s.perspective == protocol.PerspectiveClient {
  573. if !s.receivedFirstForwardSecurePacket && packet.encryptionLevel == protocol.Encryption1RTT {
  574. s.receivedFirstForwardSecurePacket = true
  575. s.sentPacketHandler.SetHandshakeComplete()
  576. }
  577. }
  578. r := bytes.NewReader(packet.data)
  579. var isAckEliciting bool
  580. for {
  581. frame, err := s.frameParser.ParseNext(r, packet.encryptionLevel)
  582. if err != nil {
  583. return err
  584. }
  585. if frame == nil {
  586. break
  587. }
  588. if ackhandler.IsFrameAckEliciting(frame) {
  589. isAckEliciting = true
  590. }
  591. if err := s.handleFrame(frame, packet.packetNumber, packet.encryptionLevel); err != nil {
  592. return err
  593. }
  594. }
  595. if err := s.receivedPacketHandler.ReceivedPacket(packet.packetNumber, packet.encryptionLevel, rcvTime, isAckEliciting); err != nil {
  596. return err
  597. }
  598. return nil
  599. }
  600. func (s *session) handleFrame(f wire.Frame, pn protocol.PacketNumber, encLevel protocol.EncryptionLevel) error {
  601. var err error
  602. wire.LogFrame(s.logger, f, false)
  603. switch frame := f.(type) {
  604. case *wire.CryptoFrame:
  605. err = s.handleCryptoFrame(frame, encLevel)
  606. case *wire.StreamFrame:
  607. err = s.handleStreamFrame(frame, encLevel)
  608. case *wire.AckFrame:
  609. err = s.handleAckFrame(frame, pn, encLevel)
  610. case *wire.ConnectionCloseFrame:
  611. s.closeRemote(qerr.Error(frame.ErrorCode, frame.ReasonPhrase))
  612. case *wire.ResetStreamFrame:
  613. err = s.handleResetStreamFrame(frame)
  614. case *wire.MaxDataFrame:
  615. s.handleMaxDataFrame(frame)
  616. case *wire.MaxStreamDataFrame:
  617. err = s.handleMaxStreamDataFrame(frame)
  618. case *wire.MaxStreamsFrame:
  619. err = s.handleMaxStreamsFrame(frame)
  620. case *wire.DataBlockedFrame:
  621. case *wire.StreamDataBlockedFrame:
  622. case *wire.StreamsBlockedFrame:
  623. case *wire.StopSendingFrame:
  624. err = s.handleStopSendingFrame(frame)
  625. case *wire.PingFrame:
  626. case *wire.PathChallengeFrame:
  627. s.handlePathChallengeFrame(frame)
  628. case *wire.PathResponseFrame:
  629. // since we don't send PATH_CHALLENGEs, we don't expect PATH_RESPONSEs
  630. err = errors.New("unexpected PATH_RESPONSE frame")
  631. case *wire.NewTokenFrame:
  632. case *wire.NewConnectionIDFrame:
  633. case *wire.RetireConnectionIDFrame:
  634. // since we don't send new connection IDs, we don't expect retirements
  635. err = errors.New("unexpected RETIRE_CONNECTION_ID frame")
  636. default:
  637. err = fmt.Errorf("unexpected frame type: %s", reflect.ValueOf(&frame).Elem().Type().Name())
  638. }
  639. return err
  640. }
  641. // handlePacket is called by the server with a new packet
  642. func (s *session) handlePacket(p *receivedPacket) {
  643. if s.closed.Get() {
  644. s.handlePacketAfterClosed(p)
  645. }
  646. // Discard packets once the amount of queued packets is larger than
  647. // the channel size, protocol.MaxSessionUnprocessedPackets
  648. select {
  649. case s.receivedPackets <- p:
  650. default:
  651. }
  652. }
  653. func (s *session) handlePacketAfterClosed(p *receivedPacket) {
  654. s.packetsReceivedAfterClose++
  655. if s.connectionClosePacket == nil {
  656. return
  657. }
  658. // exponential backoff
  659. // only send a CONNECTION_CLOSE for the 1st, 2nd, 4th, 8th, 16th, ... packet arriving
  660. for n := s.packetsReceivedAfterClose; n > 1; n = n / 2 {
  661. if n%2 != 0 {
  662. return
  663. }
  664. }
  665. s.logger.Debugf("Received %d packets after sending CONNECTION_CLOSE. Retransmitting.", s.packetsReceivedAfterClose)
  666. if err := s.conn.Write(s.connectionClosePacket.raw); err != nil {
  667. s.logger.Debugf("Error retransmitting CONNECTION_CLOSE: %s", err)
  668. }
  669. }
  670. func (s *session) handleCryptoFrame(frame *wire.CryptoFrame, encLevel protocol.EncryptionLevel) error {
  671. encLevelChanged, err := s.cryptoStreamManager.HandleCryptoFrame(frame, encLevel)
  672. if err != nil {
  673. return err
  674. }
  675. s.logger.Debugf("Handled crypto frame at level %s. encLevelChanged: %t", encLevel, encLevelChanged)
  676. if encLevelChanged {
  677. s.tryDecryptingQueuedPackets()
  678. }
  679. return nil
  680. }
  681. func (s *session) handleStreamFrame(frame *wire.StreamFrame, encLevel protocol.EncryptionLevel) error {
  682. // TODO(#1261): implement strict rules for frames types in unencrypted packets
  683. if encLevel < protocol.Encryption1RTT {
  684. return qerr.Error(qerr.ProtocolViolation, fmt.Sprintf("received unencrypted stream data on stream %d", frame.StreamID))
  685. }
  686. str, err := s.streamsMap.GetOrOpenReceiveStream(frame.StreamID)
  687. if err != nil {
  688. return err
  689. }
  690. if str == nil {
  691. // Stream is closed and already garbage collected
  692. // ignore this StreamFrame
  693. return nil
  694. }
  695. return str.handleStreamFrame(frame)
  696. }
  697. func (s *session) handleMaxDataFrame(frame *wire.MaxDataFrame) {
  698. s.connFlowController.UpdateSendWindow(frame.ByteOffset)
  699. }
  700. func (s *session) handleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) error {
  701. str, err := s.streamsMap.GetOrOpenSendStream(frame.StreamID)
  702. if err != nil {
  703. return err
  704. }
  705. if str == nil {
  706. // stream is closed and already garbage collected
  707. return nil
  708. }
  709. str.handleMaxStreamDataFrame(frame)
  710. return nil
  711. }
  712. func (s *session) handleMaxStreamsFrame(frame *wire.MaxStreamsFrame) error {
  713. return s.streamsMap.HandleMaxStreamsFrame(frame)
  714. }
  715. func (s *session) handleResetStreamFrame(frame *wire.ResetStreamFrame) error {
  716. str, err := s.streamsMap.GetOrOpenReceiveStream(frame.StreamID)
  717. if err != nil {
  718. return err
  719. }
  720. if str == nil {
  721. // stream is closed and already garbage collected
  722. return nil
  723. }
  724. return str.handleResetStreamFrame(frame)
  725. }
  726. func (s *session) handleStopSendingFrame(frame *wire.StopSendingFrame) error {
  727. str, err := s.streamsMap.GetOrOpenSendStream(frame.StreamID)
  728. if err != nil {
  729. return err
  730. }
  731. if str == nil {
  732. // stream is closed and already garbage collected
  733. return nil
  734. }
  735. str.handleStopSendingFrame(frame)
  736. return nil
  737. }
  738. func (s *session) handlePathChallengeFrame(frame *wire.PathChallengeFrame) {
  739. s.queueControlFrame(&wire.PathResponseFrame{Data: frame.Data})
  740. }
  741. func (s *session) handleAckFrame(frame *wire.AckFrame, pn protocol.PacketNumber, encLevel protocol.EncryptionLevel) error {
  742. if err := s.sentPacketHandler.ReceivedAck(frame, pn, encLevel, s.lastPacketReceivedTime); err != nil {
  743. return err
  744. }
  745. if encLevel == protocol.Encryption1RTT {
  746. s.receivedPacketHandler.IgnoreBelow(s.sentPacketHandler.GetLowestPacketNotConfirmedAcked())
  747. }
  748. return nil
  749. }
  750. // closeLocal closes the session and send a CONNECTION_CLOSE containing the error
  751. func (s *session) closeLocal(e error) {
  752. s.closeOnce.Do(func() {
  753. if e == nil {
  754. s.logger.Infof("Closing session.")
  755. } else {
  756. s.logger.Errorf("Closing session with error: %s", e)
  757. }
  758. s.sessionRunner.Retire(s.srcConnID)
  759. s.closeChan <- closeError{err: e, sendClose: true, remote: false}
  760. })
  761. }
  762. // destroy closes the session without sending the error on the wire
  763. func (s *session) destroy(e error) {
  764. s.closeOnce.Do(func() {
  765. if nerr, ok := e.(net.Error); ok && nerr.Timeout() {
  766. s.logger.Errorf("Destroying session %s: %s", s.destConnID, e)
  767. } else {
  768. s.logger.Errorf("Destroying session %s with error: %s", s.destConnID, e)
  769. }
  770. s.sessionRunner.Remove(s.srcConnID)
  771. s.closeChan <- closeError{err: e, sendClose: false, remote: false}
  772. })
  773. }
  774. // closeForRecreating closes the session in order to recreate it immediately afterwards
  775. // It returns the first packet number that should be used in the new session.
  776. func (s *session) closeForRecreating() protocol.PacketNumber {
  777. s.destroy(errCloseForRecreating)
  778. nextPN, _ := s.sentPacketHandler.PeekPacketNumber(protocol.EncryptionInitial)
  779. return nextPN
  780. }
  781. func (s *session) closeRemote(e error) {
  782. s.closeOnce.Do(func() {
  783. s.logger.Errorf("Peer closed session with error: %s", e)
  784. s.sessionRunner.Remove(s.srcConnID)
  785. s.closeChan <- closeError{err: e, remote: true}
  786. })
  787. }
  788. // Close the connection. It sends a qerr.NoError.
  789. // It waits until the run loop has stopped before returning
  790. func (s *session) Close() error {
  791. s.closeLocal(nil)
  792. <-s.ctx.Done()
  793. return nil
  794. }
  795. func (s *session) CloseWithError(code protocol.ApplicationErrorCode, e error) error {
  796. s.closeLocal(qerr.Error(qerr.ErrorCode(code), e.Error()))
  797. <-s.ctx.Done()
  798. return nil
  799. }
  800. func (s *session) handleCloseError(closeErr closeError) {
  801. if closeErr.err == nil {
  802. closeErr.err = qerr.NoError
  803. }
  804. var quicErr *qerr.QuicError
  805. var ok bool
  806. if quicErr, ok = closeErr.err.(*qerr.QuicError); !ok {
  807. quicErr = qerr.ToQuicError(closeErr.err)
  808. }
  809. s.streamsMap.CloseWithError(quicErr)
  810. if !closeErr.sendClose {
  811. return
  812. }
  813. // If this is a remote close we're done here
  814. if closeErr.remote {
  815. return
  816. }
  817. if err := s.sendConnectionClose(quicErr); err != nil {
  818. s.logger.Debugf("Error sending CONNECTION_CLOSE: %s", err)
  819. }
  820. }
  821. func (s *session) processTransportParameters(data []byte) {
  822. var params *handshake.TransportParameters
  823. var err error
  824. switch s.perspective {
  825. case protocol.PerspectiveClient:
  826. params, err = s.processTransportParametersForClient(data)
  827. case protocol.PerspectiveServer:
  828. params, err = s.processTransportParametersForServer(data)
  829. }
  830. if err != nil {
  831. s.closeLocal(err)
  832. return
  833. }
  834. s.logger.Debugf("Received Transport Parameters: %s", params)
  835. s.peerParams = params
  836. if err := s.streamsMap.UpdateLimits(params); err != nil {
  837. s.closeLocal(err)
  838. return
  839. }
  840. s.packer.HandleTransportParameters(params)
  841. s.frameParser.SetAckDelayExponent(params.AckDelayExponent)
  842. s.connFlowController.UpdateSendWindow(params.InitialMaxData)
  843. if params.StatelessResetToken != nil {
  844. s.sessionRunner.AddResetToken(*params.StatelessResetToken, s)
  845. }
  846. }
  847. func (s *session) processTransportParametersForClient(data []byte) (*handshake.TransportParameters, error) {
  848. params := &handshake.TransportParameters{}
  849. if err := params.Unmarshal(data, s.perspective.Opposite()); err != nil {
  850. return nil, err
  851. }
  852. // check the Retry token
  853. if !params.OriginalConnectionID.Equal(s.origDestConnID) {
  854. return nil, fmt.Errorf("expected original_connection_id to equal %s, is %s", s.origDestConnID, params.OriginalConnectionID)
  855. }
  856. return params, nil
  857. }
  858. func (s *session) processTransportParametersForServer(data []byte) (*handshake.TransportParameters, error) {
  859. params := &handshake.TransportParameters{}
  860. if err := params.Unmarshal(data, s.perspective.Opposite()); err != nil {
  861. return nil, err
  862. }
  863. return params, nil
  864. }
  865. func (s *session) sendPackets() error {
  866. s.pacingDeadline = time.Time{}
  867. sendMode := s.sentPacketHandler.SendMode()
  868. if sendMode == ackhandler.SendNone { // shortcut: return immediately if there's nothing to send
  869. return nil
  870. }
  871. numPackets := s.sentPacketHandler.ShouldSendNumPackets()
  872. var numPacketsSent int
  873. sendLoop:
  874. for {
  875. switch sendMode {
  876. case ackhandler.SendNone:
  877. break sendLoop
  878. case ackhandler.SendAck:
  879. // If we already sent packets, and the send mode switches to SendAck,
  880. // we've just become congestion limited.
  881. // There's no need to try to send an ACK at this moment.
  882. if numPacketsSent > 0 {
  883. return nil
  884. }
  885. // We can at most send a single ACK only packet.
  886. // There will only be a new ACK after receiving new packets.
  887. // SendAck is only returned when we're congestion limited, so we don't need to set the pacingt timer.
  888. return s.maybeSendAckOnlyPacket()
  889. case ackhandler.SendPTO:
  890. if err := s.sendProbePacket(); err != nil {
  891. return err
  892. }
  893. numPacketsSent++
  894. case ackhandler.SendRetransmission:
  895. sentPacket, err := s.maybeSendRetransmission()
  896. if err != nil {
  897. return err
  898. }
  899. if sentPacket {
  900. numPacketsSent++
  901. // This can happen if a retransmission queued, but it wasn't necessary to send it.
  902. // e.g. when an Initial is queued, but we already received a packet from the server.
  903. }
  904. case ackhandler.SendAny:
  905. sentPacket, err := s.sendPacket()
  906. if err != nil {
  907. return err
  908. }
  909. if !sentPacket {
  910. break sendLoop
  911. }
  912. numPacketsSent++
  913. default:
  914. return fmt.Errorf("BUG: invalid send mode %d", sendMode)
  915. }
  916. if numPacketsSent >= numPackets {
  917. break
  918. }
  919. sendMode = s.sentPacketHandler.SendMode()
  920. }
  921. // Only start the pacing timer if we sent as many packets as we were allowed.
  922. // There will probably be more to send when calling sendPacket again.
  923. if numPacketsSent == numPackets {
  924. s.pacingDeadline = s.sentPacketHandler.TimeUntilSend()
  925. }
  926. return nil
  927. }
  928. func (s *session) maybeSendAckOnlyPacket() error {
  929. packet, err := s.packer.MaybePackAckPacket()
  930. if err != nil {
  931. return err
  932. }
  933. if packet == nil {
  934. return nil
  935. }
  936. s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket())
  937. return s.sendPackedPacket(packet)
  938. }
  939. // maybeSendRetransmission sends retransmissions for at most one packet.
  940. // It takes care that Initials aren't retransmitted, if a packet from the server was already received.
  941. func (s *session) maybeSendRetransmission() (bool, error) {
  942. retransmitPacket := s.sentPacketHandler.DequeuePacketForRetransmission()
  943. if retransmitPacket == nil {
  944. return false, nil
  945. }
  946. s.logger.Debugf("Dequeueing retransmission for packet 0x%x (%s)", retransmitPacket.PacketNumber, retransmitPacket.EncryptionLevel)
  947. packets, err := s.packer.PackRetransmission(retransmitPacket)
  948. if err != nil {
  949. return false, err
  950. }
  951. ackhandlerPackets := make([]*ackhandler.Packet, len(packets))
  952. for i, packet := range packets {
  953. ackhandlerPackets[i] = packet.ToAckHandlerPacket()
  954. }
  955. s.sentPacketHandler.SentPacketsAsRetransmission(ackhandlerPackets, retransmitPacket.PacketNumber)
  956. for _, packet := range packets {
  957. if err := s.sendPackedPacket(packet); err != nil {
  958. return false, err
  959. }
  960. }
  961. return true, nil
  962. }
  963. func (s *session) sendProbePacket() error {
  964. p, err := s.sentPacketHandler.DequeueProbePacket()
  965. if err != nil {
  966. return err
  967. }
  968. s.logger.Debugf("Sending a retransmission for %#x as a probe packet.", p.PacketNumber)
  969. packets, err := s.packer.PackRetransmission(p)
  970. if err != nil {
  971. return err
  972. }
  973. ackhandlerPackets := make([]*ackhandler.Packet, len(packets))
  974. for i, packet := range packets {
  975. ackhandlerPackets[i] = packet.ToAckHandlerPacket()
  976. }
  977. s.sentPacketHandler.SentPacketsAsRetransmission(ackhandlerPackets, p.PacketNumber)
  978. for _, packet := range packets {
  979. if err := s.sendPackedPacket(packet); err != nil {
  980. return err
  981. }
  982. }
  983. return nil
  984. }
  985. func (s *session) sendPacket() (bool, error) {
  986. if isBlocked, offset := s.connFlowController.IsNewlyBlocked(); isBlocked {
  987. s.framer.QueueControlFrame(&wire.DataBlockedFrame{DataLimit: offset})
  988. }
  989. s.windowUpdateQueue.QueueAll()
  990. packet, err := s.packer.PackPacket()
  991. if err != nil || packet == nil {
  992. return false, err
  993. }
  994. s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket())
  995. if err := s.sendPackedPacket(packet); err != nil {
  996. return false, err
  997. }
  998. return true, nil
  999. }
  1000. func (s *session) sendPackedPacket(packet *packedPacket) error {
  1001. defer packet.buffer.Release()
  1002. if s.firstAckElicitingPacketAfterIdleSentTime.IsZero() && packet.IsAckEliciting() {
  1003. s.firstAckElicitingPacketAfterIdleSentTime = time.Now()
  1004. }
  1005. s.logPacket(packet)
  1006. return s.conn.Write(packet.raw)
  1007. }
  1008. func (s *session) sendConnectionClose(quicErr *qerr.QuicError) error {
  1009. var reason string
  1010. // don't send details of crypto errors
  1011. if !quicErr.IsCryptoError() {
  1012. reason = quicErr.ErrorMessage
  1013. }
  1014. packet, err := s.packer.PackConnectionClose(&wire.ConnectionCloseFrame{
  1015. ErrorCode: quicErr.ErrorCode,
  1016. ReasonPhrase: reason,
  1017. })
  1018. if err != nil {
  1019. return err
  1020. }
  1021. s.connectionClosePacket = packet
  1022. s.logPacket(packet)
  1023. return s.conn.Write(packet.raw)
  1024. }
  1025. func (s *session) logPacket(packet *packedPacket) {
  1026. if !s.logger.Debug() {
  1027. // We don't need to allocate the slices for calling the format functions
  1028. return
  1029. }
  1030. s.logger.Debugf("-> Sending packet 0x%x (%d bytes) for connection %s, %s", packet.header.PacketNumber, len(packet.raw), s.srcConnID, packet.EncryptionLevel())
  1031. packet.header.Log(s.logger)
  1032. for _, frame := range packet.frames {
  1033. wire.LogFrame(s.logger, frame, true)
  1034. }
  1035. }
  1036. // GetOrOpenStream either returns an existing stream, a newly opened stream, or nil if a stream with the provided ID is already closed.
  1037. // It is *only* needed for gQUIC's H2.
  1038. // It will be removed as soon as gQUIC moves towards the IETF H2/QUIC stream mapping.
  1039. func (s *session) GetOrOpenStream(id protocol.StreamID) (Stream, error) {
  1040. str, err := s.streamsMap.GetOrOpenSendStream(id)
  1041. if str != nil {
  1042. if bstr, ok := str.(Stream); ok {
  1043. return bstr, err
  1044. }
  1045. return nil, fmt.Errorf("Stream %d is not a bidirectional stream", id)
  1046. }
  1047. // make sure to return an actual nil value here, not an Stream with value nil
  1048. return nil, err
  1049. }
  1050. // AcceptStream returns the next stream openend by the peer
  1051. func (s *session) AcceptStream() (Stream, error) {
  1052. return s.streamsMap.AcceptStream()
  1053. }
  1054. func (s *session) AcceptUniStream() (ReceiveStream, error) {
  1055. return s.streamsMap.AcceptUniStream()
  1056. }
  1057. // OpenStream opens a stream
  1058. func (s *session) OpenStream() (Stream, error) {
  1059. return s.streamsMap.OpenStream()
  1060. }
  1061. func (s *session) OpenStreamSync() (Stream, error) {
  1062. return s.streamsMap.OpenStreamSync()
  1063. }
  1064. func (s *session) OpenUniStream() (SendStream, error) {
  1065. return s.streamsMap.OpenUniStream()
  1066. }
  1067. func (s *session) OpenUniStreamSync() (SendStream, error) {
  1068. return s.streamsMap.OpenUniStreamSync()
  1069. }
  1070. func (s *session) newFlowController(id protocol.StreamID) flowcontrol.StreamFlowController {
  1071. var initialSendWindow protocol.ByteCount
  1072. if s.peerParams != nil {
  1073. if id.Type() == protocol.StreamTypeUni {
  1074. initialSendWindow = s.peerParams.InitialMaxStreamDataUni
  1075. } else {
  1076. if id.InitiatedBy() == s.perspective {
  1077. initialSendWindow = s.peerParams.InitialMaxStreamDataBidiRemote
  1078. } else {
  1079. initialSendWindow = s.peerParams.InitialMaxStreamDataBidiLocal
  1080. }
  1081. }
  1082. }
  1083. return flowcontrol.NewStreamFlowController(
  1084. id,
  1085. s.connFlowController,
  1086. protocol.InitialMaxStreamData,
  1087. protocol.ByteCount(s.config.MaxReceiveStreamFlowControlWindow),
  1088. initialSendWindow,
  1089. s.onHasStreamWindowUpdate,
  1090. s.rttStats,
  1091. s.logger,
  1092. )
  1093. }
  1094. // scheduleSending signals that we have data for sending
  1095. func (s *session) scheduleSending() {
  1096. select {
  1097. case s.sendingScheduled <- struct{}{}:
  1098. default:
  1099. }
  1100. }
  1101. func (s *session) tryQueueingUndecryptablePacket(p *receivedPacket) {
  1102. if s.handshakeComplete {
  1103. s.logger.Debugf("Received undecryptable packet from %s after the handshake (%d bytes)", p.remoteAddr.String(), len(p.data))
  1104. return
  1105. }
  1106. if len(s.undecryptablePackets)+1 > protocol.MaxUndecryptablePackets {
  1107. s.logger.Infof("Dropping undecrytable packet (%d bytes). Undecryptable packet queue full.", len(p.data))
  1108. return
  1109. }
  1110. s.logger.Infof("Queueing packet (%d bytes) for later decryption", len(p.data))
  1111. s.undecryptablePackets = append(s.undecryptablePackets, p)
  1112. }
  1113. func (s *session) tryDecryptingQueuedPackets() {
  1114. for _, p := range s.undecryptablePackets {
  1115. s.handlePacket(p)
  1116. }
  1117. s.undecryptablePackets = s.undecryptablePackets[:0]
  1118. }
  1119. func (s *session) queueControlFrame(f wire.Frame) {
  1120. s.framer.QueueControlFrame(f)
  1121. s.scheduleSending()
  1122. }
  1123. func (s *session) onHasStreamWindowUpdate(id protocol.StreamID) {
  1124. s.windowUpdateQueue.AddStream(id)
  1125. s.scheduleSending()
  1126. }
  1127. func (s *session) onHasConnectionWindowUpdate() {
  1128. s.windowUpdateQueue.AddConnection()
  1129. s.scheduleSending()
  1130. }
  1131. func (s *session) onHasStreamData(id protocol.StreamID) {
  1132. s.framer.AddActiveStream(id)
  1133. s.scheduleSending()
  1134. }
  1135. func (s *session) onStreamCompleted(id protocol.StreamID) {
  1136. if err := s.streamsMap.DeleteStream(id); err != nil {
  1137. s.closeLocal(err)
  1138. }
  1139. }
  1140. func (s *session) LocalAddr() net.Addr {
  1141. return s.conn.LocalAddr()
  1142. }
  1143. func (s *session) RemoteAddr() net.Addr {
  1144. return s.conn.RemoteAddr()
  1145. }
  1146. func (s *session) getPerspective() protocol.Perspective {
  1147. return s.perspective
  1148. }
  1149. func (s *session) GetVersion() protocol.VersionNumber {
  1150. return s.version
  1151. }