Coverage for Lib/asyncio/sslproto.py: 84%

1# Contains code from https://github.com/MagicStack/uvloop/tree/v0.16.0 

2# SPDX-License-Identifier: PSF-2.0 AND (MIT OR Apache-2.0) 

3# SPDX-FileCopyrightText: Copyright (c) 2015-2021 MagicStack Inc. http://magic.io 

4 

5import collections 

6import enum 

7import warnings 

8try: 

9 import ssl 

10except ImportError: # pragma: no cover 

11 ssl = None 

12 

13from . import constants 

14from . import exceptions 

15from . import protocols 

16from . import transports 

17from .log import logger 

18 

19if ssl is not None: 19 ↛ 23line 19 didn't jump to line 23 because the condition on line 19 was always true

20 SSLAgainErrors = (ssl.SSLWantReadError, ssl.SSLSyscallError) 

21 

22 

23class SSLProtocolState(enum.Enum): 

24 UNWRAPPED = "UNWRAPPED" 

25 DO_HANDSHAKE = "DO_HANDSHAKE" 

26 WRAPPED = "WRAPPED" 

27 FLUSHING = "FLUSHING" 

28 SHUTDOWN = "SHUTDOWN" 

29 

30 

31class AppProtocolState(enum.Enum): 

32 # This tracks the state of app protocol (https://git.io/fj59P): 

33 # 

34 # INIT -cm-> CON_MADE [-dr*->] [-er-> EOF?] -cl-> CON_LOST 

35 # 

36 # * cm: connection_made() 

37 # * dr: data_received() 

38 # * er: eof_received() 

39 # * cl: connection_lost() 

40 

41 STATE_INIT = "STATE_INIT" 

42 STATE_CON_MADE = "STATE_CON_MADE" 

43 STATE_EOF = "STATE_EOF" 

44 STATE_CON_LOST = "STATE_CON_LOST" 

45 

46 

47def _create_transport_context(server_side, server_hostname): 

48 if server_side: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 raise ValueError('Server side SSL needs a valid SSLContext') 

50 

51 # Client side may pass ssl=True to use a default 

52 # context; in that case the sslcontext passed is None. 

53 # The default is secure for client connections. 

54 # Python 3.4+: use up-to-date strong settings. 

55 sslcontext = ssl.create_default_context() 

56 if not server_hostname: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 sslcontext.check_hostname = False 

58 return sslcontext 

59 

60 

61def add_flowcontrol_defaults(high, low, kb): 

62 if high is None: 

63 if low is None: 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true

64 hi = kb * 1024 

65 else: 

66 lo = low 

67 hi = 4 * lo 

68 else: 

69 hi = high 

70 if low is None: 70 ↛ 73line 70 didn't jump to line 73 because the condition on line 70 was always true

71 lo = hi // 4 

72 else: 

73 lo = low 

74 

75 if not hi >= lo >= 0: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 raise ValueError('high (%r) must be >= low (%r) must be >= 0' % 

77 (hi, lo)) 

78 

79 return hi, lo 

80 

81 

82class _SSLProtocolTransport(transports._FlowControlMixin, 

83 transports.Transport): 

84 

85 _start_tls_compatible = True 

86 _sendfile_compatible = constants._SendfileMode.FALLBACK 

87 

88 def __init__(self, loop, ssl_protocol): 

89 self._loop = loop 

90 self._ssl_protocol = ssl_protocol 

91 self._closed = False 

92 

93 def get_extra_info(self, name, default=None): 

94 """Get optional transport information.""" 

95 return self._ssl_protocol._get_extra_info(name, default) 

96 

97 def set_protocol(self, protocol): 

98 self._ssl_protocol._set_app_protocol(protocol) 

99 

100 def get_protocol(self): 

101 return self._ssl_protocol._app_protocol 

102 

103 def is_closing(self): 

104 return self._closed or self._ssl_protocol._is_transport_closing() 

105 

106 def close(self): 

107 """Close the transport. 

108 

109 Buffered data will be flushed asynchronously. No more data 

110 will be received. After all buffered data is flushed, the 

111 protocol's connection_lost() method will (eventually) called 

112 with None as its argument. 

113 """ 

114 if not self._closed: 

115 self._closed = True 

116 self._ssl_protocol._start_shutdown() 

117 else: 

118 self._ssl_protocol = None 

119 

120 def __del__(self, _warnings=warnings): 

121 if not self._closed: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 self._closed = True 

123 _warnings.warn( 

124 "unclosed transport <asyncio._SSLProtocolTransport " 

125 "object>", ResourceWarning) 

126 

127 def is_reading(self): 

128 return not self._ssl_protocol._app_reading_paused 

129 

130 def pause_reading(self): 

131 """Pause the receiving end. 

132 

133 No data will be passed to the protocol's data_received() 

134 method until resume_reading() is called. 

135 """ 

136 self._ssl_protocol._pause_reading() 

137 

138 def resume_reading(self): 

139 """Resume the receiving end. 

140 

141 Data received will once again be passed to the protocol's 

142 data_received() method. 

143 """ 

144 self._ssl_protocol._resume_reading() 

145 

146 def set_write_buffer_limits(self, high=None, low=None): 

147 """Set the high- and low-water limits for write flow control. 

148 

149 These two values control when to call the protocol's 

150 pause_writing() and resume_writing() methods. If specified, 

151 the low-water limit must be less than or equal to the 

152 high-water limit. Neither value can be negative. 

153 

154 The defaults are implementation-specific. If only the 

155 high-water limit is given, the low-water limit defaults to an 

156 implementation-specific value less than or equal to the 

157 high-water limit. Setting high to zero forces low to zero as 

158 well, and causes pause_writing() to be called whenever the 

159 buffer becomes non-empty. Setting low to zero causes 

160 resume_writing() to be called only once the buffer is empty. 

161 Use of zero for either limit is generally sub-optimal as it 

162 reduces opportunities for doing I/O and computation 

163 concurrently. 

164 """ 

165 self._ssl_protocol._set_write_buffer_limits(high, low) 

166 self._ssl_protocol._control_app_writing() 

167 

168 def get_write_buffer_limits(self): 

169 return (self._ssl_protocol._outgoing_low_water, 

170 self._ssl_protocol._outgoing_high_water) 

171 

172 def get_write_buffer_size(self): 

173 """Return the current size of the write buffers.""" 

174 return self._ssl_protocol._get_write_buffer_size() 

175 

176 def set_read_buffer_limits(self, high=None, low=None): 

177 """Set the high- and low-water limits for read flow control. 

178 

179 These two values control when to call the upstream transport's 

180 pause_reading() and resume_reading() methods. If specified, 

181 the low-water limit must be less than or equal to the 

182 high-water limit. Neither value can be negative. 

183 

184 The defaults are implementation-specific. If only the 

185 high-water limit is given, the low-water limit defaults to an 

186 implementation-specific value less than or equal to the 

187 high-water limit. Setting high to zero forces low to zero as 

188 well, and causes pause_reading() to be called whenever the 

189 buffer becomes non-empty. Setting low to zero causes 

190 resume_reading() to be called only once the buffer is empty. 

191 Use of zero for either limit is generally sub-optimal as it 

192 reduces opportunities for doing I/O and computation 

193 concurrently. 

194 """ 

195 self._ssl_protocol._set_read_buffer_limits(high, low) 

196 self._ssl_protocol._control_ssl_reading() 

197 

198 def get_read_buffer_limits(self): 

199 return (self._ssl_protocol._incoming_low_water, 

200 self._ssl_protocol._incoming_high_water) 

201 

202 def get_read_buffer_size(self): 

203 """Return the current size of the read buffer.""" 

204 return self._ssl_protocol._get_read_buffer_size() 

205 

206 @property 

207 def _protocol_paused(self): 

208 # Required for sendfile fallback pause_writing/resume_writing logic 

209 return self._ssl_protocol._app_writing_paused 

210 

211 def write(self, data): 

212 """Write some data bytes to the transport. 

213 

214 This does not block; it buffers the data and arranges for it 

215 to be sent out asynchronously. 

216 """ 

217 if not isinstance(data, (bytes, bytearray, memoryview)): 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true

218 raise TypeError(f"data: expecting a bytes-like instance, " 

219 f"got {type(data).__name__}") 

220 if not data: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 return 

222 self._ssl_protocol._write_appdata((data,)) 

223 

224 def writelines(self, list_of_data): 

225 """Write a list (or any iterable) of data bytes to the transport. 

226 

227 The default implementation concatenates the arguments and 

228 calls write() on the result. 

229 """ 

230 self._ssl_protocol._write_appdata(list_of_data) 

231 

232 def write_eof(self): 

233 """Close the write end after flushing buffered data. 

234 

235 This raises :exc:`NotImplementedError` right now. 

236 """ 

237 raise NotImplementedError 

238 

239 def can_write_eof(self): 

240 """Return True if this transport supports write_eof(), False if not.""" 

241 return False 

242 

243 def abort(self): 

244 """Close the transport immediately. 

245 

246 Buffered data will be lost. No more data will be received. 

247 The protocol's connection_lost() method will (eventually) be 

248 called with None as its argument. 

249 """ 

250 self._force_close(None) 

251 

252 def _force_close(self, exc): 

253 self._closed = True 

254 if self._ssl_protocol is not None: 254 ↛ exitline 254 didn't return from function '_force_close' because the condition on line 254 was always true

255 self._ssl_protocol._abort(exc) 

256 

257 def _test__append_write_backlog(self, data): 

258 # for test only 

259 self._ssl_protocol._write_backlog.append(data) 

260 self._ssl_protocol._write_buffer_size += len(data) 

261 

262 

263class SSLProtocol(protocols.BufferedProtocol): 

264 max_size = 256 * 1024 # Buffer size passed to read() 

265 

266 _handshake_start_time = None 

267 _handshake_timeout_handle = None 

268 _shutdown_timeout_handle = None 

269 

270 def __init__(self, loop, app_protocol, sslcontext, waiter, 

271 server_side=False, server_hostname=None, 

272 call_connection_made=True, 

273 ssl_handshake_timeout=None, 

274 ssl_shutdown_timeout=None): 

275 if ssl is None: 

276 raise RuntimeError("stdlib ssl module not available") 

277 

278 self._ssl_buffer = bytearray(self.max_size) 

279 self._ssl_buffer_view = memoryview(self._ssl_buffer) 

280 

281 if ssl_handshake_timeout is None: 

282 ssl_handshake_timeout = constants.SSL_HANDSHAKE_TIMEOUT 

283 elif ssl_handshake_timeout <= 0: 

284 raise ValueError( 

285 f"ssl_handshake_timeout should be a positive number, " 

286 f"got {ssl_handshake_timeout}") 

287 if ssl_shutdown_timeout is None: 287 ↛ 289line 287 didn't jump to line 289 because the condition on line 287 was always true

288 ssl_shutdown_timeout = constants.SSL_SHUTDOWN_TIMEOUT 

289 elif ssl_shutdown_timeout <= 0: 

290 raise ValueError( 

291 f"ssl_shutdown_timeout should be a positive number, " 

292 f"got {ssl_shutdown_timeout}") 

293 

294 if not sslcontext: 

295 sslcontext = _create_transport_context( 

296 server_side, server_hostname) 

297 

298 self._server_side = server_side 

299 if server_hostname and not server_side: 

300 self._server_hostname = server_hostname 

301 else: 

302 self._server_hostname = None 

303 self._sslcontext = sslcontext 

304 # SSL-specific extra info. More info are set when the handshake 

305 # completes. 

306 self._extra = dict(sslcontext=sslcontext) 

307 

308 # App data write buffering 

309 self._write_backlog = collections.deque() 

310 self._write_buffer_size = 0 

311 

312 self._waiter = waiter 

313 self._loop = loop 

314 self._set_app_protocol(app_protocol) 

315 self._app_transport = None 

316 self._app_transport_created = False 

317 # transport, ex: SelectorSocketTransport 

318 self._transport = None 

319 self._ssl_handshake_timeout = ssl_handshake_timeout 

320 self._ssl_shutdown_timeout = ssl_shutdown_timeout 

321 # SSL and state machine 

322 self._incoming = ssl.MemoryBIO() 

323 self._outgoing = ssl.MemoryBIO() 

324 self._state = SSLProtocolState.UNWRAPPED 

325 self._conn_lost = 0 # Set when connection_lost called 

326 if call_connection_made: 

327 self._app_state = AppProtocolState.STATE_INIT 

328 else: 

329 self._app_state = AppProtocolState.STATE_CON_MADE 

330 self._sslobj = self._sslcontext.wrap_bio( 

331 self._incoming, self._outgoing, 

332 server_side=self._server_side, 

333 server_hostname=self._server_hostname) 

334 

335 # Flow Control 

336 

337 self._ssl_writing_paused = False 

338 

339 self._app_reading_paused = False 

340 

341 self._ssl_reading_paused = False 

342 self._incoming_high_water = 0 

343 self._incoming_low_water = 0 

344 self._set_read_buffer_limits() 

345 self._eof_received = False 

346 

347 self._app_writing_paused = False 

348 self._outgoing_high_water = 0 

349 self._outgoing_low_water = 0 

350 self._set_write_buffer_limits() 

351 self._get_app_transport() 

352 

353 def _set_app_protocol(self, app_protocol): 

354 self._app_protocol = app_protocol 

355 # Make fast hasattr check first 

356 if (hasattr(app_protocol, 'get_buffer') and 

357 isinstance(app_protocol, protocols.BufferedProtocol)): 

358 self._app_protocol_get_buffer = app_protocol.get_buffer 

359 self._app_protocol_buffer_updated = app_protocol.buffer_updated 

360 self._app_protocol_is_buffer = True 

361 else: 

362 self._app_protocol_is_buffer = False 

363 

364 def _wakeup_waiter(self, exc=None): 

365 if self._waiter is None: 

366 return 

367 if not self._waiter.cancelled(): 

368 if exc is not None: 

369 self._waiter.set_exception(exc) 

370 else: 

371 self._waiter.set_result(None) 

372 self._waiter = None 

373 

374 def _get_app_transport(self): 

375 if self._app_transport is None: 

376 if self._app_transport_created: 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true

377 raise RuntimeError('Creating _SSLProtocolTransport twice') 

378 self._app_transport = _SSLProtocolTransport(self._loop, self) 

379 self._app_transport_created = True 

380 return self._app_transport 

381 

382 def _is_transport_closing(self): 

383 return self._transport is not None and self._transport.is_closing() 

384 

385 def connection_made(self, transport): 

386 """Called when the low-level connection is made. 

387 

388 Start the SSL handshake. 

389 """ 

390 self._transport = transport 

391 self._start_handshake() 

392 

393 def connection_lost(self, exc): 

394 """Called when the low-level connection is lost or closed. 

395 

396 The argument is an exception object or None (the latter 

397 meaning a regular EOF is received or the connection was 

398 aborted or closed). 

399 """ 

400 self._write_backlog.clear() 

401 self._outgoing.read() 

402 self._conn_lost += 1 

403 

404 # Just mark the app transport as closed so that its __dealloc__ 

405 # doesn't complain. 

406 if self._app_transport is not None: 406 ↛ 409line 406 didn't jump to line 409 because the condition on line 406 was always true

407 self._app_transport._closed = True 

408 

409 if self._state != SSLProtocolState.DO_HANDSHAKE: 

410 if ( 

411 self._app_state == AppProtocolState.STATE_CON_MADE or 

412 self._app_state == AppProtocolState.STATE_EOF 

413 ): 

414 self._app_state = AppProtocolState.STATE_CON_LOST 

415 self._loop.call_soon(self._app_protocol.connection_lost, exc) 

416 self._set_state(SSLProtocolState.UNWRAPPED) 

417 self._transport = None 

418 self._app_transport = None 

419 self._app_protocol = None 

420 self._wakeup_waiter(exc) 

421 

422 if self._shutdown_timeout_handle: 

423 self._shutdown_timeout_handle.cancel() 

424 self._shutdown_timeout_handle = None 

425 if self._handshake_timeout_handle: 

426 self._handshake_timeout_handle.cancel() 

427 self._handshake_timeout_handle = None 

428 

429 def get_buffer(self, n): 

430 want = n 

431 if want <= 0 or want > self.max_size: 431 ↛ 433line 431 didn't jump to line 433 because the condition on line 431 was always true

432 want = self.max_size 

433 if len(self._ssl_buffer) < want: 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true

434 self._ssl_buffer = bytearray(want) 

435 self._ssl_buffer_view = memoryview(self._ssl_buffer) 

436 return self._ssl_buffer_view 

437 

438 def buffer_updated(self, nbytes): 

439 self._incoming.write(self._ssl_buffer_view[:nbytes]) 

440 

441 if self._state == SSLProtocolState.DO_HANDSHAKE: 

442 self._do_handshake() 

443 

444 elif self._state == SSLProtocolState.WRAPPED: 

445 self._do_read() 

446 

447 elif self._state == SSLProtocolState.FLUSHING: 447 ↛ 448line 447 didn't jump to line 448 because the condition on line 447 was never true

448 self._do_flush() 

449 

450 elif self._state == SSLProtocolState.SHUTDOWN: 450 ↛ exitline 450 didn't return from function 'buffer_updated' because the condition on line 450 was always true

451 self._do_shutdown() 

452 

453 def eof_received(self): 

454 """Called when the other end of the low-level stream 

455 is half-closed. 

456 

457 If this returns a false value (including None), the transport 

458 will close itself. If it returns a true value, closing the 

459 transport is up to the protocol. 

460 """ 

461 self._eof_received = True 

462 try: 

463 if self._loop.get_debug(): 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true

464 logger.debug("%r received EOF", self) 

465 

466 if self._state == SSLProtocolState.DO_HANDSHAKE: 

467 self._on_handshake_complete(ConnectionResetError) 

468 

469 elif self._state == SSLProtocolState.WRAPPED: 

470 self._set_state(SSLProtocolState.FLUSHING) 

471 if self._app_reading_paused: 

472 return True 

473 else: 

474 self._do_flush() 

475 

476 elif self._state == SSLProtocolState.FLUSHING: 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true

477 self._do_write() 

478 self._set_state(SSLProtocolState.SHUTDOWN) 

479 self._do_shutdown() 

480 

481 elif self._state == SSLProtocolState.SHUTDOWN: 481 ↛ exitline 481 didn't return from function 'eof_received' because the condition on line 481 was always true

482 self._do_shutdown() 

483 

484 except Exception: 

485 self._transport.close() 

486 raise 

487 

488 def _get_extra_info(self, name, default=None): 

489 if name in self._extra: 

490 return self._extra[name] 

491 elif self._transport is not None: 

492 return self._transport.get_extra_info(name, default) 

493 else: 

494 return default 

495 

496 def _set_state(self, new_state): 

497 allowed = False 

498 

499 if new_state == SSLProtocolState.UNWRAPPED: 

500 allowed = True 

501 

502 elif ( 

503 self._state == SSLProtocolState.UNWRAPPED and 

504 new_state == SSLProtocolState.DO_HANDSHAKE 

505 ): 

506 allowed = True 

507 

508 elif ( 

509 self._state == SSLProtocolState.DO_HANDSHAKE and 

510 new_state == SSLProtocolState.WRAPPED 

511 ): 

512 allowed = True 

513 

514 elif ( 

515 self._state == SSLProtocolState.WRAPPED and 

516 new_state == SSLProtocolState.FLUSHING 

517 ): 

518 allowed = True 

519 

520 elif ( 520 ↛ 526line 520 didn't jump to line 526 because the condition on line 520 was always true

521 self._state == SSLProtocolState.FLUSHING and 

522 new_state == SSLProtocolState.SHUTDOWN 

523 ): 

524 allowed = True 

525 

526 if allowed: 526 ↛ 530line 526 didn't jump to line 530 because the condition on line 526 was always true

527 self._state = new_state 

528 

529 else: 

530 raise RuntimeError( 

531 'cannot switch state from {} to {}'.format( 

532 self._state, new_state)) 

533 

534 # Handshake flow 

535 

536 def _start_handshake(self): 

537 if self._loop.get_debug(): 537 ↛ 538line 537 didn't jump to line 538 because the condition on line 537 was never true

538 logger.debug("%r starts SSL handshake", self) 

539 self._handshake_start_time = self._loop.time() 

540 else: 

541 self._handshake_start_time = None 

542 

543 self._set_state(SSLProtocolState.DO_HANDSHAKE) 

544 

545 # start handshake timeout count down 

546 self._handshake_timeout_handle = \ 

547 self._loop.call_later(self._ssl_handshake_timeout, 

548 self._check_handshake_timeout) 

549 

550 self._do_handshake() 

551 

552 def _check_handshake_timeout(self): 

553 if self._state == SSLProtocolState.DO_HANDSHAKE: 553 ↛ exitline 553 didn't return from function '_check_handshake_timeout' because the condition on line 553 was always true

554 msg = ( 

555 f"SSL handshake is taking longer than " 

556 f"{self._ssl_handshake_timeout} seconds: " 

557 f"aborting the connection" 

558 ) 

559 self._fatal_error(ConnectionAbortedError(msg)) 

560 

561 def _do_handshake(self): 

562 try: 

563 self._sslobj.do_handshake() 

564 except SSLAgainErrors: 

565 self._process_outgoing() 

566 except ssl.SSLError as exc: 

567 self._on_handshake_complete(exc) 

568 else: 

569 self._on_handshake_complete(None) 

570 

571 def _on_handshake_complete(self, handshake_exc): 

572 if self._handshake_timeout_handle is not None: 572 ↛ 576line 572 didn't jump to line 576 because the condition on line 572 was always true

573 self._handshake_timeout_handle.cancel() 

574 self._handshake_timeout_handle = None 

575 

576 sslobj = self._sslobj 

577 try: 

578 if handshake_exc is None: 

579 self._set_state(SSLProtocolState.WRAPPED) 

580 else: 

581 raise handshake_exc 

582 

583 peercert = sslobj.getpeercert() 

584 except Exception as exc: 

585 handshake_exc = None 

586 self._set_state(SSLProtocolState.UNWRAPPED) 

587 if isinstance(exc, ssl.CertificateError): 

588 msg = 'SSL handshake failed on verifying the certificate' 

589 else: 

590 msg = 'SSL handshake failed' 

591 # gh-98078: When the handshake fails, OpenSSL leaves the fatal 

592 # TLS alert (for example "bad certificate" or "protocol 

593 # version") in the outgoing BIO. Send it to the peer before 

594 # closing the transport so that it knows why the handshake 

595 # failed. 

596 self._process_outgoing() 

597 self._fatal_error(exc, msg) 

598 self._wakeup_waiter(exc) 

599 return 

600 

601 if self._loop.get_debug(): 601 ↛ 602line 601 didn't jump to line 602 because the condition on line 601 was never true

602 dt = self._loop.time() - self._handshake_start_time 

603 logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3) 

604 

605 # Add extra info that becomes available after handshake. 

606 self._extra.update(peercert=peercert, 

607 cipher=sslobj.cipher(), 

608 compression=sslobj.compression(), 

609 ssl_object=sslobj) 

610 if self._app_state == AppProtocolState.STATE_INIT: 

611 self._app_state = AppProtocolState.STATE_CON_MADE 

612 self._app_protocol.connection_made(self._get_app_transport()) 

613 self._wakeup_waiter() 

614 self._do_read() 

615 

616 # Shutdown flow 

617 

618 def _start_shutdown(self): 

619 if ( 

620 self._state in ( 

621 SSLProtocolState.FLUSHING, 

622 SSLProtocolState.SHUTDOWN, 

623 SSLProtocolState.UNWRAPPED 

624 ) 

625 ): 

626 return 

627 if self._app_transport is not None: 627 ↛ 629line 627 didn't jump to line 629 because the condition on line 627 was always true

628 self._app_transport._closed = True 

629 if self._state == SSLProtocolState.DO_HANDSHAKE: 

630 self._abort(None) 

631 else: 

632 self._set_state(SSLProtocolState.FLUSHING) 

633 self._shutdown_timeout_handle = self._loop.call_later( 

634 self._ssl_shutdown_timeout, 

635 self._check_shutdown_timeout 

636 ) 

637 self._do_flush() 

638 

639 def _check_shutdown_timeout(self): 

640 if ( 

641 self._state in ( 

642 SSLProtocolState.FLUSHING, 

643 SSLProtocolState.SHUTDOWN 

644 ) 

645 ): 

646 self._transport._force_close( 

647 exceptions.TimeoutError('SSL shutdown timed out')) 

648 

649 def _do_flush(self): 

650 self._do_read() 

651 self._set_state(SSLProtocolState.SHUTDOWN) 

652 self._do_shutdown() 

653 

654 def _do_shutdown(self): 

655 try: 

656 if not self._eof_received: 

657 self._sslobj.unwrap() 

658 except SSLAgainErrors: 

659 self._process_outgoing() 

660 except ssl.SSLError as exc: 

661 # gh-98078: send what OpenSSL left in the outgoing BIO, e.g. 

662 # the close_notify alert, to the peer before closing (see 

663 # _on_handshake_complete()). 

664 self._process_outgoing() 

665 self._on_shutdown_complete(exc) 

666 else: 

667 self._process_outgoing() 

668 self._call_eof_received() 

669 self._on_shutdown_complete(None) 

670 

671 def _on_shutdown_complete(self, shutdown_exc): 

672 if self._shutdown_timeout_handle is not None: 

673 self._shutdown_timeout_handle.cancel() 

674 self._shutdown_timeout_handle = None 

675 

676 if shutdown_exc: 

677 self._fatal_error(shutdown_exc) 

678 else: 

679 self._loop.call_soon(self._transport.close) 

680 

681 def _abort(self, exc): 

682 self._set_state(SSLProtocolState.UNWRAPPED) 

683 if self._transport is not None: 683 ↛ exitline 683 didn't return from function '_abort' because the condition on line 683 was always true

684 self._transport._force_close(exc) 

685 

686 # Outgoing flow 

687 

688 def _write_appdata(self, list_of_data): 

689 if ( 

690 self._state in ( 

691 SSLProtocolState.FLUSHING, 

692 SSLProtocolState.SHUTDOWN, 

693 SSLProtocolState.UNWRAPPED 

694 ) 

695 ): 

696 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 696 ↛ 697line 696 didn't jump to line 697 because the condition on line 696 was never true

697 logger.warning('SSL connection is closed') 

698 self._conn_lost += 1 

699 return 

700 

701 for data in list_of_data: 

702 self._write_backlog.append(data) 

703 self._write_buffer_size += len(data) 

704 

705 try: 

706 if self._state == SSLProtocolState.WRAPPED: 706 ↛ exitline 706 didn't return from function '_write_appdata' because the condition on line 706 was always true

707 self._do_write() 

708 

709 except Exception as ex: 

710 self._fatal_error(ex, 'Fatal error on SSL protocol') 

711 

712 def _do_write(self): 

713 try: 

714 while self._write_backlog: 

715 data = self._write_backlog[0] 

716 count = self._sslobj.write(data) 

717 data_len = len(data) 

718 if count < data_len: 718 ↛ 719line 718 didn't jump to line 719 because the condition on line 718 was never true

719 self._write_backlog[0] = data[count:] 

720 self._write_buffer_size -= count 

721 else: 

722 del self._write_backlog[0] 

723 self._write_buffer_size -= data_len 

724 except SSLAgainErrors: 

725 pass 

726 self._process_outgoing() 

727 

728 def _process_outgoing(self): 

729 if not self._ssl_writing_paused: 

730 data = self._outgoing.read() 

731 if len(data): 

732 self._transport.write(data) 

733 self._control_app_writing() 

734 

735 # Incoming flow 

736 

737 def _do_read(self): 

738 if ( 

739 self._state not in ( 

740 SSLProtocolState.WRAPPED, 

741 SSLProtocolState.FLUSHING, 

742 ) 

743 ): 

744 return 

745 try: 

746 if not self._app_reading_paused: 

747 if self._app_protocol_is_buffer: 

748 self._do_read__buffered() 

749 else: 

750 self._do_read__copied() 

751 if self._write_backlog: 

752 self._do_write() 

753 else: 

754 self._process_outgoing() 

755 self._control_ssl_reading() 

756 except ssl.SSLError as ex: 

757 # gh-98078: send the fatal TLS alert left in the outgoing 

758 # BIO to the peer (see _on_handshake_complete()). 

759 self._process_outgoing() 

760 self._fatal_error(ex, 'Fatal error on SSL protocol') 

761 except Exception as ex: 

762 self._fatal_error(ex, 'Fatal error on SSL protocol') 

763 

764 def _do_read__buffered(self): 

765 offset = 0 

766 count = 1 

767 

768 buf = self._app_protocol_get_buffer(self._get_read_buffer_size()) 

769 wants = len(buf) 

770 

771 try: 

772 count = self._sslobj.read(wants, buf) 

773 

774 if count > 0: 774 ↛ 786line 774 didn't jump to line 786 because the condition on line 774 was always true

775 offset = count 

776 while offset < wants: 776 ↛ 777line 776 didn't jump to line 777 because the condition on line 776 was never true

777 count = self._sslobj.read(wants - offset, buf[offset:]) 

778 if count > 0: 

779 offset += count 

780 else: 

781 break 

782 else: 

783 self._loop.call_soon(self._do_read) 

784 except SSLAgainErrors: 

785 pass 

786 if offset > 0: 

787 self._app_protocol_buffer_updated(offset) 

788 if not count: 788 ↛ 790line 788 didn't jump to line 790 because the condition on line 788 was never true

789 # close_notify 

790 self._call_eof_received() 

791 self._start_shutdown() 

792 

793 def _do_read__copied(self): 

794 chunk = b'1' 

795 zero = True 

796 one = False 

797 

798 try: 

799 while True: 

800 chunk = self._sslobj.read(self.max_size) 

801 if not chunk: 

802 break 

803 if zero: 

804 zero = False 

805 one = True 

806 first = chunk 

807 elif one: 

808 one = False 

809 data = [first, chunk] 

810 else: 

811 data.append(chunk) 

812 except SSLAgainErrors: 

813 pass 

814 if one: 

815 self._app_protocol.data_received(first) 

816 elif not zero: 

817 self._app_protocol.data_received(b''.join(data)) 

818 if not chunk: 

819 # close_notify 

820 self._call_eof_received() 

821 self._start_shutdown() 

822 

823 def _call_eof_received(self): 

824 try: 

825 if self._app_state == AppProtocolState.STATE_CON_MADE: 

826 self._app_state = AppProtocolState.STATE_EOF 

827 keep_open = self._app_protocol.eof_received() 

828 if keep_open: 828 ↛ 829line 828 didn't jump to line 829 because the condition on line 828 was never true

829 logger.warning('returning true from eof_received() ' 

830 'has no effect when using ssl') 

831 except (KeyboardInterrupt, SystemExit): 

832 raise 

833 except BaseException as ex: 

834 self._fatal_error(ex, 'Error calling eof_received()') 

835 

836 # Flow control for writes from APP socket 

837 

838 def _control_app_writing(self): 

839 size = self._get_write_buffer_size() 

840 if size >= self._outgoing_high_water and not self._app_writing_paused: 

841 self._app_writing_paused = True 

842 try: 

843 self._app_protocol.pause_writing() 

844 except (KeyboardInterrupt, SystemExit): 

845 raise 

846 except BaseException as exc: 

847 self._loop.call_exception_handler({ 

848 'message': 'protocol.pause_writing() failed', 

849 'exception': exc, 

850 'transport': self._app_transport, 

851 'protocol': self, 

852 }) 

853 elif size <= self._outgoing_low_water and self._app_writing_paused: 

854 self._app_writing_paused = False 

855 try: 

856 self._app_protocol.resume_writing() 

857 except (KeyboardInterrupt, SystemExit): 

858 raise 

859 except BaseException as exc: 

860 self._loop.call_exception_handler({ 

861 'message': 'protocol.resume_writing() failed', 

862 'exception': exc, 

863 'transport': self._app_transport, 

864 'protocol': self, 

865 }) 

866 

867 def _get_write_buffer_size(self): 

868 return self._outgoing.pending + self._write_buffer_size 

869 

870 def _set_write_buffer_limits(self, high=None, low=None): 

871 high, low = add_flowcontrol_defaults( 

872 high, low, constants.FLOW_CONTROL_HIGH_WATER_SSL_WRITE) 

873 self._outgoing_high_water = high 

874 self._outgoing_low_water = low 

875 

876 # Flow control for reads to APP socket 

877 

878 def _pause_reading(self):