Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/werkzeug/serving.py: 17%

1"""A WSGI and HTTP server for use **during development only**. This 

2server is convenient to use, but is not designed to be particularly 

3stable, secure, or efficient. Use a dedicate WSGI server and HTTP 

4server when deploying to production. 

5 

6It provides features like interactive debugging and code reloading. Use 

7``run_simple`` to start the server. Put this in a ``run.py`` script: 

8 

9.. code-block:: python 

10 

11 from myapp import create_app 

12 from werkzeug import run_simple 

13""" 

14from __future__ import annotations 

15 

16import errno 

17import io 

18import os 

19import selectors 

20import socket 

21import socketserver 

22import sys 

23import typing as t 

24from datetime import datetime as dt 

25from datetime import timedelta 

26from datetime import timezone 

27from http.server import BaseHTTPRequestHandler 

28from http.server import HTTPServer 

29from urllib.parse import unquote 

30from urllib.parse import urlsplit 

31 

32from ._internal import _log 

33from ._internal import _wsgi_encoding_dance 

34from .exceptions import InternalServerError 

35from .urls import uri_to_iri 

36 

37try: 

38 import ssl 

39except ImportError: 

40 

41 class _SslDummy: 

42 def __getattr__(self, name: str) -> t.Any: 

43 raise RuntimeError( # noqa: B904 

44 "SSL is unavailable because this Python runtime was not" 

45 " compiled with SSL/TLS support." 

46 ) 

47 

48 ssl = _SslDummy() # type: ignore 

49 

50_log_add_style = True 

51 

52if os.name == "nt": 

53 try: 

54 __import__("colorama") 

55 except ImportError: 

56 _log_add_style = False 

57 

58can_fork = hasattr(os, "fork") 

59 

60if can_fork: 

61 ForkingMixIn = socketserver.ForkingMixIn 

62else: 

63 

64 class ForkingMixIn: # type: ignore 

65 pass 

66 

67 

68try: 

69 af_unix = socket.AF_UNIX 

70except AttributeError: 

71 af_unix = None # type: ignore 

72 

73LISTEN_QUEUE = 128 

74 

75_TSSLContextArg = t.Optional[ 

76 t.Union["ssl.SSLContext", t.Tuple[str, t.Optional[str]], t.Literal["adhoc"]] 

77] 

78 

79if t.TYPE_CHECKING: 

80 from _typeshed.wsgi import WSGIApplication 

81 from _typeshed.wsgi import WSGIEnvironment 

82 from cryptography.hazmat.primitives.asymmetric.rsa import ( 

83 RSAPrivateKeyWithSerialization, 

84 ) 

85 from cryptography.x509 import Certificate 

86 

87 

88class DechunkedInput(io.RawIOBase): 

89 """An input stream that handles Transfer-Encoding 'chunked'""" 

90 

91 def __init__(self, rfile: t.IO[bytes]) -> None: 

92 self._rfile = rfile 

93 self._done = False 

94 self._len = 0 

95 

96 def readable(self) -> bool: 

97 return True 

98 

99 def read_chunk_len(self) -> int: 

100 try: 

101 line = self._rfile.readline().decode("latin1") 

102 _len = int(line.strip(), 16) 

103 except ValueError as e: 

104 raise OSError("Invalid chunk header") from e 

105 if _len < 0: 

106 raise OSError("Negative chunk length not allowed") 

107 return _len 

108 

109 def readinto(self, buf: bytearray) -> int: # type: ignore 

110 read = 0 

111 while not self._done and read < len(buf): 

112 if self._len == 0: 

113 # This is the first chunk or we fully consumed the previous 

114 # one. Read the next length of the next chunk 

115 self._len = self.read_chunk_len() 

116 

117 if self._len == 0: 

118 # Found the final chunk of size 0. The stream is now exhausted, 

119 # but there is still a final newline that should be consumed 

120 self._done = True 

121 

122 if self._len > 0: 

123 # There is data (left) in this chunk, so append it to the 

124 # buffer. If this operation fully consumes the chunk, this will 

125 # reset self._len to 0. 

126 n = min(len(buf), self._len) 

127 

128 # If (read + chunk size) becomes more than len(buf), buf will 

129 # grow beyond the original size and read more data than 

130 # required. So only read as much data as can fit in buf. 

131 if read + n > len(buf): 

132 buf[read:] = self._rfile.read(len(buf) - read) 

133 self._len -= len(buf) - read 

134 read = len(buf) 

135 else: 

136 buf[read : read + n] = self._rfile.read(n) 

137 self._len -= n 

138 read += n 

139 

140 if self._len == 0: 

141 # Skip the terminating newline of a chunk that has been fully 

142 # consumed. This also applies to the 0-sized final chunk 

143 terminator = self._rfile.readline() 

144 if terminator not in (b"\n", b"\r\n", b"\r"): 

145 raise OSError("Missing chunk terminating newline") 

146 

147 return read 

148 

149 

150class WSGIRequestHandler(BaseHTTPRequestHandler): 

151 """A request handler that implements WSGI dispatching.""" 

152 

153 server: BaseWSGIServer 

154 

155 @property 

156 def server_version(self) -> str: # type: ignore 

157 return self.server._server_version 

158 

159 def make_environ(self) -> WSGIEnvironment: 

160 request_url = urlsplit(self.path) 

161 url_scheme = "http" if self.server.ssl_context is None else "https" 

162 

163 if not self.client_address: 

164 self.client_address = ("<local>", 0) 

165 elif isinstance(self.client_address, str): 

166 self.client_address = (self.client_address, 0) 

167 

168 # If there was no scheme but the path started with two slashes, 

169 # the first segment may have been incorrectly parsed as the 

170 # netloc, prepend it to the path again. 

171 if not request_url.scheme and request_url.netloc: 

172 path_info = f"/{request_url.netloc}{request_url.path}" 

173 else: 

174 path_info = request_url.path 

175 

176 path_info = unquote(path_info) 

177 

178 environ: WSGIEnvironment = { 

179 "wsgi.version": (1, 0), 

180 "wsgi.url_scheme": url_scheme, 

181 "wsgi.input": self.rfile, 

182 "wsgi.errors": sys.stderr, 

183 "wsgi.multithread": self.server.multithread, 

184 "wsgi.multiprocess": self.server.multiprocess, 

185 "wsgi.run_once": False, 

186 "werkzeug.socket": self.connection, 

187 "SERVER_SOFTWARE": self.server_version, 

188 "REQUEST_METHOD": self.command, 

189 "SCRIPT_NAME": "", 

190 "PATH_INFO": _wsgi_encoding_dance(path_info), 

191 "QUERY_STRING": _wsgi_encoding_dance(request_url.query), 

192 # Non-standard, added by mod_wsgi, uWSGI 

193 "REQUEST_URI": _wsgi_encoding_dance(self.path), 

194 # Non-standard, added by gunicorn 

195 "RAW_URI": _wsgi_encoding_dance(self.path), 

196 "REMOTE_ADDR": self.address_string(), 

197 "REMOTE_PORT": self.port_integer(), 

198 "SERVER_NAME": self.server.server_address[0], 

199 "SERVER_PORT": str(self.server.server_address[1]), 

200 "SERVER_PROTOCOL": self.request_version, 

201 } 

202 

203 for key, value in self.headers.items(): 

204 if "_" in key: 

205 continue 

206 

207 key = key.upper().replace("-", "_") 

208 value = value.replace("\r\n", "") 

209 if key not in ("CONTENT_TYPE", "CONTENT_LENGTH"): 

210 key = f"HTTP_{key}" 

211 if key in environ: 

212 value = f"{environ[key]},{value}" 

213 environ[key] = value 

214 

215 if environ.get("HTTP_TRANSFER_ENCODING", "").strip().lower() == "chunked": 

216 environ["wsgi.input_terminated"] = True 

217 environ["wsgi.input"] = DechunkedInput(environ["wsgi.input"]) 

218 

219 # Per RFC 2616, if the URL is absolute, use that as the host. 

220 # We're using "has a scheme" to indicate an absolute URL. 

221 if request_url.scheme and request_url.netloc: 

222 environ["HTTP_HOST"] = request_url.netloc 

223 

224 try: 

225 # binary_form=False gives nicer information, but wouldn't be compatible with 

226 # what Nginx or Apache could return. 

227 peer_cert = self.connection.getpeercert(binary_form=True) 

228 if peer_cert is not None: 

229 # Nginx and Apache use PEM format. 

230 environ["SSL_CLIENT_CERT"] = ssl.DER_cert_to_PEM_cert(peer_cert) 

231 except ValueError: 

232 # SSL handshake hasn't finished. 

233 self.server.log("error", "Cannot fetch SSL peer certificate info") 

234 except AttributeError: 

235 # Not using TLS, the socket will not have getpeercert(). 

236 pass 

237 

238 return environ 

239 

240 def run_wsgi(self) -> None: 

241 if self.headers.get("Expect", "").lower().strip() == "100-continue": 

242 self.wfile.write(b"HTTP/1.1 100 Continue\r\n\r\n") 

243 

244 self.environ = environ = self.make_environ() 

245 status_set: str | None = None 

246 headers_set: list[tuple[str, str]] | None = None 

247 status_sent: str | None = None 

248 headers_sent: list[tuple[str, str]] | None = None 

249 chunk_response: bool = False 

250 

251 def write(data: bytes) -> None: 

252 nonlocal status_sent, headers_sent, chunk_response 

253 assert status_set is not None, "write() before start_response" 

254 assert headers_set is not None, "write() before start_response" 

255 if status_sent is None: 

256 status_sent = status_set 

257 headers_sent = headers_set 

258 try: 

259 code_str, msg = status_sent.split(None, 1) 

260 except ValueError: 

261 code_str, msg = status_sent, "" 

262 code = int(code_str) 

263 self.send_response(code, msg) 

264 header_keys = set() 

265 for key, value in headers_sent: 

266 self.send_header(key, value) 

267 header_keys.add(key.lower()) 

268 

269 # Use chunked transfer encoding if there is no content 

270 # length. Do not use for 1xx and 204 responses. 304 

271 # responses and HEAD requests are also excluded, which 

272 # is the more conservative behavior and matches other 

273 # parts of the code. 

274 # https://httpwg.org/specs/rfc7230.html#rfc.section.3.3.1 

275 if ( 

276 not ( 

277 "content-length" in header_keys 

278 or environ["REQUEST_METHOD"] == "HEAD" 

279 or (100 <= code < 200) 

280 or code in {204, 304} 

281 ) 

282 and self.protocol_version >= "HTTP/1.1" 

283 ): 

284 chunk_response = True 

285 self.send_header("Transfer-Encoding", "chunked") 

286 

287 # Always close the connection. This disables HTTP/1.1 

288 # keep-alive connections. They aren't handled well by 

289 # Python's http.server because it doesn't know how to 

290 # drain the stream before the next request line. 

291 self.send_header("Connection", "close") 

292 self.end_headers() 

293 

294 assert isinstance(data, bytes), "applications must write bytes" 

295 

296 if data: 

297 if chunk_response: 

298 self.wfile.write(hex(len(data))[2:].encode()) 

299 self.wfile.write(b"\r\n") 

300 

301 self.wfile.write(data) 

302 

303 if chunk_response: 

304 self.wfile.write(b"\r\n") 

305 

306 self.wfile.flush() 

307 

308 def start_response(status, headers, exc_info=None): # type: ignore 

309 nonlocal status_set, headers_set 

310 if exc_info: 

311 try: 

312 if headers_sent: 

313 raise exc_info[1].with_traceback(exc_info[2]) 

314 finally: 

315 exc_info = None 

316 elif headers_set: 

317 raise AssertionError("Headers already set") 

318 status_set = status 

319 headers_set = headers 

320 return write 

321 

322 def execute(app: WSGIApplication) -> None: 

323 application_iter = app(environ, start_response) 

324 try: 

325 for data in application_iter: 

326 write(data) 

327 if not headers_sent: 

328 write(b"") 

329 if chunk_response: 

330 self.wfile.write(b"0\r\n\r\n") 

331 finally: 

332 # Check for any remaining data in the read socket, and discard it. This 

333 # will read past request.max_content_length, but lets the client see a 

334 # 413 response instead of a connection reset failure. If we supported 

335 # keep-alive connections, this naive approach would break by reading the 

336 # next request line. Since we know that write (above) closes every 

337 # connection we can read everything. 

338 selector = selectors.DefaultSelector() 

339 selector.register(self.connection, selectors.EVENT_READ) 

340 total_size = 0 

341 total_reads = 0 

342 

343 # A timeout of 0 tends to fail because a client needs a small amount of 

344 # time to continue sending its data. 

345 while selector.select(timeout=0.01): 

346 # Only read 10MB into memory at a time. 

347 data = self.rfile.read(10_000_000) 

348 total_size += len(data) 

349 total_reads += 1 

350 

351 # Stop reading on no data, >=10GB, or 1000 reads. If a client sends 

352 # more than that, they'll get a connection reset failure. 

353 if not data or total_size >= 10_000_000_000 or total_reads > 1000: 

354 break 

355 

356 selector.close() 

357 

358 if hasattr(application_iter, "close"): 

359 application_iter.close() 

360 

361 try: 

362 execute(self.server.app) 

363 except (ConnectionError, socket.timeout) as e: 

364 self.connection_dropped(e, environ) 

365 except Exception as e: 

366 if self.server.passthrough_errors: 

367 raise 

368 

369 if status_sent is not None and chunk_response: 

370 self.close_connection = True 

371 

372 try: 

373 # if we haven't yet sent the headers but they are set 

374 # we roll back to be able to set them again. 

375 if status_sent is None: 

376 status_set = None 

377 headers_set = None 

378 execute(InternalServerError()) 

379 except Exception: 

380 pass 

381 

382 from .debug.tbtools import DebugTraceback 

383 

384 msg = DebugTraceback(e).render_traceback_text() 

385 self.server.log("error", f"Error on request:\n{msg}") 

386 

387 def handle(self) -> None: 

388 """Handles a request ignoring dropped connections.""" 

389 try: 

390 super().handle() 

391 except (ConnectionError, socket.timeout) as e: 

392 self.connection_dropped(e) 

393 except Exception as e: 

394 if self.server.ssl_context is not None and is_ssl_error(e): 

395 self.log_error("SSL error occurred: %s", e) 

396 else: 

397 raise 

398 

399 def connection_dropped( 

400 self, error: BaseException, environ: WSGIEnvironment | None = None 

401 ) -> None: 

402 """Called if the connection was closed by the client. By default 

403 nothing happens. 

404 """ 

405 

406 def __getattr__(self, name: str) -> t.Any: 

407 # All HTTP methods are handled by run_wsgi. 

408 if name.startswith("do_"): 

409 return self.run_wsgi 

410 

411 # All other attributes are forwarded to the base class. 

412 return getattr(super(), name) 

413 

414 def address_string(self) -> str: 

415 if getattr(self, "environ", None): 

416 return self.environ["REMOTE_ADDR"] # type: ignore 

417 

418 if not self.client_address: 

419 return "<local>" 

420 

421 return self.client_address[0] 

422 

423 def port_integer(self) -> int: 

424 return self.client_address[1] 

425 

426 # Escape control characters. This is defined (but private) in Python 3.12. 

427 _control_char_table = str.maketrans( 

428 {c: rf"\x{c:02x}" for c in [*range(0x20), *range(0x7F, 0xA0)]} 

429 ) 

430 _control_char_table[ord("\\")] = r"\\" 

431 

432 def log_request(self, code: int | str = "-", size: int | str = "-") -> None: 

433 try: 

434 path = uri_to_iri(self.path) 

435 msg = f"{self.command} {path} {self.request_version}" 

436 except AttributeError: 

437 # path isn't set if the requestline was bad 

438 msg = self.requestline 

439 

440 # Escape control characters that may be in the decoded path. 

441 msg = msg.translate(self._control_char_table) 

442 code = str(code) 

443 

444 if code[0] == "1": # 1xx - Informational 

445 msg = _ansi_style(msg, "bold") 

446 elif code == "200": # 2xx - Success 

447 pass 

448 elif code == "304": # 304 - Resource Not Modified 

449 msg = _ansi_style(msg, "cyan") 

450 elif code[0] == "3": # 3xx - Redirection 

451 msg = _ansi_style(msg, "green") 

452 elif code == "404": # 404 - Resource Not Found 

453 msg = _ansi_style(msg, "yellow") 

454 elif code[0] == "4": # 4xx - Client Error 

455 msg = _ansi_style(msg, "bold", "red") 

456 else: # 5xx, or any other response 

457 msg = _ansi_style(msg, "bold", "magenta") 

458 

459 self.log("info", '"%s" %s %s', msg, code, size) 

460 

461 def log_error(self, format: str, *args: t.Any) -> None: 

462 self.log("error", format, *args) 

463 

464 def log_message(self, format: str, *args: t.Any) -> None: 

465 self.log("info", format, *args) 

466 

467 def log(self, type: str, message: str, *args: t.Any) -> None: 

468 _log( 

469 type, 

470 f"{self.address_string()} - - [{self.log_date_time_string()}] {message}\n", 

471 *args, 

472 ) 

473 

474 

475def _ansi_style(value: str, *styles: str) -> str: 

476 if not _log_add_style: 

477 return value 

478 

479 codes = { 

480 "bold": 1, 

481 "red": 31, 

482 "green": 32, 

483 "yellow": 33, 

484 "magenta": 35, 

485 "cyan": 36, 

486 } 

487 

488 for style in styles: 

489 value = f"\x1b[{codes[style]}m{value}" 

490 

491 return f"{value}\x1b[0m" 

492 

493 

494def generate_adhoc_ssl_pair( 

495 cn: str | None = None, 

496) -> tuple[Certificate, RSAPrivateKeyWithSerialization]: 

497 try: 

498 from cryptography import x509 

499 from cryptography.x509.oid import NameOID 

500 from cryptography.hazmat.backends import default_backend 

501 from cryptography.hazmat.primitives import hashes 

502 from cryptography.hazmat.primitives.asymmetric import rsa 

503 except ImportError: 

504 raise TypeError( 

505 "Using ad-hoc certificates requires the cryptography library." 

506 ) from None 

507 

508 backend = default_backend() 

509 pkey = rsa.generate_private_key( 

510 public_exponent=65537, key_size=2048, backend=backend 

511 ) 

512 

513 # pretty damn sure that this is not actually accepted by anyone 

514 if cn is None: 

515 cn = "*" 

516 

517 subject = x509.Name( 

518 [ 

519 x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Dummy Certificate"), 

520 x509.NameAttribute(NameOID.COMMON_NAME, cn), 

521 ] 

522 ) 

523 

524 backend = default_backend() 

525 cert = ( 

526 x509.CertificateBuilder() 

527 .subject_name(subject) 

528 .issuer_name(subject) 

529 .public_key(pkey.public_key()) 

530 .serial_number(x509.random_serial_number()) 

531 .not_valid_before(dt.now(timezone.utc)) 

532 .not_valid_after(dt.now(timezone.utc) + timedelta(days=365)) 

533 .add_extension(x509.ExtendedKeyUsage([x509.OID_SERVER_AUTH]), critical=False) 

534 .add_extension(x509.SubjectAlternativeName([x509.DNSName(cn)]), critical=False) 

535 .sign(pkey, hashes.SHA256(), backend) 

536 ) 

537 return cert, pkey 

538 

539 

540def make_ssl_devcert( 

541 base_path: str, host: str | None = None, cn: str | None = None 

542) -> tuple[str, str]: 

543 """Creates an SSL key for development. This should be used instead of 

544 the ``'adhoc'`` key which generates a new cert on each server start. 

545 It accepts a path for where it should store the key and cert and 

546 either a host or CN. If a host is given it will use the CN 

547 ``*.host/CN=host``. 

548 

549 For more information see :func:`run_simple`. 

550 

551 .. versionadded:: 0.9 

552 

553 :param base_path: the path to the certificate and key. The extension 

554 ``.crt`` is added for the certificate, ``.key`` is 

555 added for the key. 

556 :param host: the name of the host. This can be used as an alternative 

557 for the `cn`. 

558 :param cn: the `CN` to use. 

559 """ 

560 

561 if host is not None: 

562 cn = f"*.{host}/CN={host}" 

563 cert, pkey = generate_adhoc_ssl_pair(cn=cn) 

564 

565 from cryptography.hazmat.primitives import serialization 

566 

567 cert_file = f"{base_path}.crt" 

568 pkey_file = f"{base_path}.key" 

569 

570 with open(cert_file, "wb") as f: 

571 f.write(cert.public_bytes(serialization.Encoding.PEM)) 

572 with open(pkey_file, "wb") as f: 

573 f.write( 

574 pkey.private_bytes( 

575 encoding=serialization.Encoding.PEM, 

576 format=serialization.PrivateFormat.TraditionalOpenSSL, 

577 encryption_algorithm=serialization.NoEncryption(), 

578 ) 

579 ) 

580 

581 return cert_file, pkey_file 

582 

583 

584def generate_adhoc_ssl_context() -> ssl.SSLContext: 

585 """Generates an adhoc SSL context for the development server.""" 

586 import tempfile 

587 import atexit 

588 

589 cert, pkey = generate_adhoc_ssl_pair() 

590 

591 from cryptography.hazmat.primitives import serialization 

592 

593 cert_handle, cert_file = tempfile.mkstemp() 

594 pkey_handle, pkey_file = tempfile.mkstemp() 

595 atexit.register(os.remove, pkey_file) 

596 atexit.register(os.remove, cert_file) 

597 

598 os.write(cert_handle, cert.public_bytes(serialization.Encoding.PEM)) 

599 os.write( 

600 pkey_handle, 

601 pkey.private_bytes( 

602 encoding=serialization.Encoding.PEM, 

603 format=serialization.PrivateFormat.TraditionalOpenSSL, 

604 encryption_algorithm=serialization.NoEncryption(), 

605 ), 

606 ) 

607 

608 os.close(cert_handle) 

609 os.close(pkey_handle) 

610 ctx = load_ssl_context(cert_file, pkey_file) 

611 return ctx 

612 

613 

614def load_ssl_context( 

615 cert_file: str, pkey_file: str | None = None, protocol: int | None = None 

616) -> ssl.SSLContext: 

617 """Loads SSL context from cert/private key files and optional protocol. 

618 Many parameters are directly taken from the API of 

619 :py:class:`ssl.SSLContext`. 

620 

621 :param cert_file: Path of the certificate to use. 

622 :param pkey_file: Path of the private key to use. If not given, the key 

623 will be obtained from the certificate file. 

624 :param protocol: A ``PROTOCOL`` constant from the :mod:`ssl` module. 

625 Defaults to :data:`ssl.PROTOCOL_TLS_SERVER`. 

626 """ 

627 if protocol is None: 

628 protocol = ssl.PROTOCOL_TLS_SERVER 

629 

630 ctx = ssl.SSLContext(protocol) 

631 ctx.load_cert_chain(cert_file, pkey_file) 

632 return ctx 

633 

634 

635def is_ssl_error(error: Exception | None = None) -> bool: 

636 """Checks if the given error (or the current one) is an SSL error.""" 

637 if error is None: 

638 error = t.cast(Exception, sys.exc_info()[1]) 

639 return isinstance(error, ssl.SSLError) 

640 

641 

642def select_address_family(host: str, port: int) -> socket.AddressFamily: 

643 """Return ``AF_INET4``, ``AF_INET6``, or ``AF_UNIX`` depending on 

644 the host and port.""" 

645 if host.startswith("unix://"): 

646 return socket.AF_UNIX 

647 elif ":" in host and hasattr(socket, "AF_INET6"): 

648 return socket.AF_INET6 

649 return socket.AF_INET 

650 

651 

652def get_sockaddr( 

653 host: str, port: int, family: socket.AddressFamily 

654) -> tuple[str, int] | str: 

655 """Return a fully qualified socket address that can be passed to 

656 :func:`socket.bind`.""" 

657 if family == af_unix: 

658 # Absolute path avoids IDNA encoding error when path starts with dot. 

659 return os.path.abspath(host.partition("://")[2]) 

660 try: 

661 res = socket.getaddrinfo( 

662 host, port, family, socket.SOCK_STREAM, socket.IPPROTO_TCP 

663 ) 

664 except socket.gaierror: 

665 return host, port 

666 return res[0][4] # type: ignore 

667 

668 

669def get_interface_ip(family: socket.AddressFamily) -> str: 

670 """Get the IP address of an external interface. Used when binding to 

671 0.0.0.0 or ::1 to show a more useful URL. 

672 

673 :meta private: 

674 """ 

675 # arbitrary private address 

676 host = "fd31:f903:5ab5:1::1" if family == socket.AF_INET6 else "10.253.155.219" 

677 

678 with socket.socket(family, socket.SOCK_DGRAM) as s: 

679 try: 

680 s.connect((host, 58162)) 

681 except OSError: 

682 return "::1" if family == socket.AF_INET6 else "127.0.0.1" 

683 

684 return s.getsockname()[0] # type: ignore 

685 

686 

687class BaseWSGIServer(HTTPServer): 

688 """A WSGI server that that handles one request at a time. 

689 

690 Use :func:`make_server` to create a server instance. 

691 """ 

692 

693 multithread = False 

694 multiprocess = False 

695 request_queue_size = LISTEN_QUEUE 

696 allow_reuse_address = True 

697 

698 def __init__( 

699 self, 

700 host: str, 

701 port: int, 

702 app: WSGIApplication, 

703 handler: type[WSGIRequestHandler] | None = None, 

704 passthrough_errors: bool = False, 

705 ssl_context: _TSSLContextArg | None = None, 

706 fd: int | None = None, 

707 ) -> None: 

708 if handler is None: 

709 handler = WSGIRequestHandler 

710 

711 # If the handler doesn't directly set a protocol version and 

712 # thread or process workers are used, then allow chunked 

713 # responses and keep-alive connections by enabling HTTP/1.1. 

714 if "protocol_version" not in vars(handler) and ( 

715 self.multithread or self.multiprocess 

716 ): 

717 handler.protocol_version = "HTTP/1.1" 

718 

719 self.host = host 

720 self.port = port 

721 self.app = app 

722 self.passthrough_errors = passthrough_errors 

723 

724 self.address_family = address_family = select_address_family(host, port) 

725 server_address = get_sockaddr(host, int(port), address_family) 

726 

727 # Remove a leftover Unix socket file from a previous run. Don't 

728 # remove a file that was set up by run_simple. 

729 if address_family == af_unix and fd is None: 

730 server_address = t.cast(str, server_address) 

731 

732 if os.path.exists(server_address): 

733 os.unlink(server_address) 

734 

735 # Bind and activate will be handled manually, it should only 

736 # happen if we're not using a socket that was already set up. 

737 super().__init__( 

738 server_address, # type: ignore[arg-type] 

739 handler, 

740 bind_and_activate=False, 

741 ) 

742 

743 if fd is None: 

744 # No existing socket descriptor, do bind_and_activate=True. 

745 try: 

746 self.server_bind() 

747 self.server_activate() 

748 except OSError as e: 

749 # Catch connection issues and show them without the traceback. Show 

750 # extra instructions for address not found, and for macOS. 

751 self.server_close() 

752 print(e.strerror, file=sys.stderr) 

753 

754 if e.errno == errno.EADDRINUSE: 

755 print( 

756 f"Port {port} is in use by another program. Either identify and" 

757 " stop that program, or start the server with a different" 

758 " port.", 

759 file=sys.stderr, 

760 ) 

761 

762 if sys.platform == "darwin" and port == 5000: 

763 print( 

764 "On macOS, try disabling the 'AirPlay Receiver' service" 

765 " from System Preferences -> General -> AirDrop & Handoff.", 

766 file=sys.stderr, 

767 ) 

768 

769 sys.exit(1) 

770 except BaseException: 

771 self.server_close() 

772 raise 

773 else: 

774 # TCPServer automatically opens a socket even if bind_and_activate is False. 

775 # Close it to silence a ResourceWarning. 

776 self.server_close() 

777 

778 # Use the passed in socket directly. 

779 self.socket = socket.fromfd(fd, address_family, socket.SOCK_STREAM) 

780 self.server_address = self.socket.getsockname() 

781 

782 if address_family != af_unix: 

783 # If port was 0, this will record the bound port. 

784 self.port = self.server_address[1] 

785 

786 if ssl_context is not None: 

787 if isinstance(ssl_context, tuple): 

788 ssl_context = load_ssl_context(*ssl_context) 

789 elif ssl_context == "adhoc": 

790 ssl_context = generate_adhoc_ssl_context() 

791 

792 self.socket = ssl_context.wrap_socket(self.socket, server_side=True) 

793 self.ssl_context: ssl.SSLContext | None = ssl_context 

794 else: 

795 self.ssl_context = None 

796 

797 import importlib.metadata 

798 

799 self._server_version = f"Werkzeug/{importlib.metadata.version('werkzeug')}" 

800 

801 def log(self, type: str, message: str, *args: t.Any) -> None: 

802 _log(type, message, *args) 

803 

804 def serve_forever(self, poll_interval: float = 0.5) -> None: 

805 try: 

806 super().serve_forever(poll_interval=poll_interval) 

807 except KeyboardInterrupt: 

808 pass 

809 finally: 

810 self.server_close() 

811 

812 def handle_error( 

813 self, request: t.Any, client_address: tuple[str, int] | str 

814 ) -> None: 

815 if self.passthrough_errors: 

816 raise 

817 

818 return super().handle_error(request, client_address) 

819 

820 def log_startup(self) -> None: 

821 """Show information about the address when starting the server.""" 

822 dev_warning = ( 

823 "WARNING: This is a development server. Do not use it in a production" 

824 " deployment. Use a production WSGI server instead." 

825 ) 

826 dev_warning = _ansi_style(dev_warning, "bold", "red") 

827 messages = [dev_warning] 

828 

829 if self.address_family == af_unix: 

830 messages.append(f" * Running on {self.host}") 

831 else: 

832 scheme = "http" if self.ssl_context is None else "https" 

833 display_hostname = self.host 

834 

835 if self.host in {"0.0.0.0", "::"}: 

836 messages.append(f" * Running on all addresses ({self.host})") 

837 

838 if self.host == "0.0.0.0": 

839 localhost = "127.0.0.1" 

840 display_hostname = get_interface_ip(socket.AF_INET) 

841 else: 

842 localhost = "[::1]" 

843 display_hostname = get_interface_ip(socket.AF_INET6)