session.go 38 KB

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