session.go 40 KB

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