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

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""" 

14 

15from __future__ import annotations 

16 

17import contextlib 

18import datetime as dt 

19import errno 

20import io 

21import os 

22import selectors 

23import socket 

24import socketserver 

25import sys 

26import typing as t 

27import wsgiref.types as wt 

28from http.server import BaseHTTPRequestHandler 

29from http.server import HTTPServer 

30from types import TracebackType 

31from urllib.parse import unquote 

32from urllib.parse import urlsplit 

33 

34from ._internal import _log 

35from ._internal import _plain_int 

36from ._internal import _wsgi_encoding_dance 

37from .datastructures.set import HeaderSet 

38from .exceptions import InternalServerError 

39from .urls import uri_to_iri 

40 

41try: 

42 import ssl 

43 

44 connection_dropped_errors: tuple[type[Exception], ...] = ( 

45 ConnectionError, 

46 socket.timeout, 

47 ssl.SSLEOFError, 

48 ) 

49except ImportError: 

50 

51 class _SslDummy: 

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

53 raise RuntimeError( # noqa: B904 

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

55 " compiled with SSL/TLS support." 

56 ) 

57 

58 ssl = _SslDummy() # type: ignore[assignment] 

59 connection_dropped_errors = (ConnectionError, socket.timeout) 

60 

61can_fork = hasattr(os, "fork") 

62 

63if can_fork: 

64 ForkingMixIn = socketserver.ForkingMixIn 

65else: 

66 

67 class ForkingMixIn: # type: ignore[no-redef] 

68 pass 

69 

70 

71try: 

72 af_unix: int | None = socket.AF_UNIX 

73except AttributeError: 

74 af_unix = None 

75 

76LISTEN_QUEUE = 128 

77 

78_TSSLContextArg = t.Union[ 

79 "ssl.SSLContext", tuple[str, str | None], t.Literal["adhoc"], None 

80] 

81 

82if t.TYPE_CHECKING: 

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

84 RSAPrivateKeyWithSerialization, 

85 ) 

86 from cryptography.x509 import Certificate 

87 

88 

89class DechunkedInput(io.RawIOBase): 

90 """An input stream that handles ``Transfer-Encoding: chunked``. Only 

91 used by the dev server, which must not be used in production. A production 

92 WSGI server will have its own robust, secure chunk handler. 

93 """ 

94 

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

96 self._rfile = rfile 

97 self._done = False 

98 self._len = 0 

99 

100 def readable(self) -> bool: 

101 return True 

102 

103 def read_chunk_len(self) -> int: 

104 try: 

105 line = self._rfile.readline(100).decode("latin1").removesuffix("\r\n") 

106 _len = _plain_int(line, 16) 

107 except ValueError as e: 

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

109 if _len < 0: 

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

111 return _len 

112 

113 def readinto(self, buf: bytearray) -> int: # type: ignore[override] 

114 read = 0 

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

116 if self._len == 0: 

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

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

119 self._len = self.read_chunk_len() 

120 

121 if self._len == 0: 

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

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

124 self._done = True 

125 

126 if self._len > 0: 

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

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

129 # reset self._len to 0. 

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

131 

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

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

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

135 if read + n > len(buf): 

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

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

138 read = len(buf) 

139 else: 

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

141 self._len -= n 

142 read += n 

143 

144 if self._len == 0: 

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

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

147 terminator = self._rfile.readline(2) 

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

149 raise OSError("Missing chunk terminating newline") 

150 

151 return read 

152 

153 

154class WSGIRequestHandler(BaseHTTPRequestHandler): 

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

156 

157 server: BaseWSGIServer 

158 

159 @property 

160 def server_version(self) -> str: # type: ignore[override] 

161 return self.server._server_version 

162 

163 def make_environ(self) -> wt.WSGIEnvironment: 

164 request_url = urlsplit(self.path) 

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

166 

167 if not self.client_address: 

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

169 elif isinstance(self.client_address, str): 

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

171 

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

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

174 # netloc, prepend it to the path again. 

175 if not request_url.scheme and request_url.netloc: 

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

177 else: 

178 path_info = request_url.path 

179 

180 path_info = unquote(path_info) 

181 

182 environ: wt.WSGIEnvironment = { 

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

184 "wsgi.url_scheme": url_scheme, 

185 "wsgi.input": self.rfile, 

186 "wsgi.errors": sys.stderr, 

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

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

189 "wsgi.run_once": False, 

190 "werkzeug.socket": self.connection, 

191 "SERVER_SOFTWARE": self.server_version, 

192 "REQUEST_METHOD": self.command, 

193 "SCRIPT_NAME": "", 

194 "PATH_INFO": _wsgi_encoding_dance(path_info), 

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

196 # Non-standard, added by mod_wsgi, uWSGI 

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

198 # Non-standard, added by gunicorn 

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

200 "REMOTE_ADDR": self.address_string(), 

201 "REMOTE_PORT": self.port_integer(), 

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

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

204 "SERVER_PROTOCOL": self.request_version, 

205 } 

206 

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

208 if "_" in key: 

209 continue 

210 

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

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

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

214 key = f"HTTP_{key}" 

215 if key in environ: 

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

217 environ[key] = value 

218 

219 if "chunked" in HeaderSet.from_header(environ.get("HTTP_TRANSFER_ENCODING")): 

220 environ["wsgi.input_terminated"] = True 

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

222 

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

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

225 if request_url.scheme and request_url.netloc: 

226 environ["HTTP_HOST"] = request_url.netloc 

227 

228 try: 

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

230 # what Nginx or Apache could return. 

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

232 if peer_cert is not None: 

233 # Nginx and Apache use PEM format. 

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

235 except ValueError: 

236 # SSL handshake hasn't finished. 

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

238 except AttributeError: 

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

240 pass 

241 

242 return environ 

243 

244 def run_wsgi(self) -> None: 

245 self.environ = environ = self.make_environ() 

246 status_set: str | None = None 

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

248 status_sent: str | None = None 

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

250 chunk_response: bool = False 

251 

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

253 nonlocal status_sent, headers_sent, chunk_response 

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

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

256 if status_sent is None: 

257 status_sent = status_set 

258 headers_sent = headers_set 

259 try: 

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

261 except ValueError: 

262 code_str, msg = status_sent, "" 

263 code = int(code_str) 

264 self.send_response(code, msg) 

265 header_keys = set() 

266 for key, value in headers_sent: 

267 self.send_header(key, value) 

268 header_keys.add(key.lower()) 

269 

270 # Use chunked transfer encoding if there is no content 

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

272 # responses and HEAD requests are also excluded, which 

273 # is the more conservative behavior and matches other 

274 # parts of the code. 

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

276 if ( 

277 not ( 

278 "content-length" in header_keys 

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

280 or (100 <= code < 200) 

281 or code in {204, 304} 

282 ) 

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

284 ): 

285 chunk_response = True 

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

287 

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

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

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

291 # drain the stream before the next request line. 

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

293 self.end_headers() 

294 

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

296 

297 if data: 

298 if chunk_response: 

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

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

301 

302 self.wfile.write(data) 

303 

304 if chunk_response: 

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

306 

307 self.wfile.flush() 

308 

309 def start_response( 

310 status: str, 

311 headers: list[tuple[str, str]], 

312 exc_info: tuple[type[BaseException], BaseException, TracebackType] 

313 | tuple[None, None, None] 

314 | None = None, 

315 ) -> t.Callable[[bytes], object]: 

316 nonlocal status_set, headers_set 

317 if exc_info: 

318 try: 

319 if headers_sent: 

320 raise exc_info[1].with_traceback(exc_info[2]) # type: ignore[union-attr] 

321 finally: 

322 exc_info = None 

323 elif headers_set: 

324 raise AssertionError("Headers already set") 

325 status_set = status 

326 headers_set = headers 

327 return write 

328 

329 def execute(app: wt.WSGIApplication) -> None: 

330 application_iter = app(environ, start_response) 

331 try: 

332 for data in application_iter: 

333 write(data) 

334 if not headers_sent: 

335 write(b"") 

336 if chunk_response: 

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

338 finally: 

339 if hasattr(application_iter, "close"): 

340 close_iter: contextlib.AbstractContextManager[t.Any] = ( 

341 contextlib.closing(application_iter) 

342 ) 

343 else: 

344 close_iter = contextlib.nullcontext() 

345 

346 with close_iter, selectors.DefaultSelector() as selector: 

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

348 # This will read past request.max_content_length, but lets the 

349 # client see a 413 response instead of a connection reset failure. 

350 # If we supported keep-alive connections, this naive approach would 

351 # break by reading the next request line. Since we know that write 

352 # (above) closes every connection we can read everything. 

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

354 total_size = 0 

355 total_reads = 0 

356 

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

358 # of time to continue sending its data. 

359 while selector.select(timeout=0.01): 

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

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

362 total_size += len(data) 

363 total_reads += 1 

364 

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

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

367 if ( 

368 not data 

369 or total_size >= 10_000_000_000 

370 or total_reads > 1_000 

371 ): 

372 break 

373 

374 try: 

375 execute(self.server.app) 

376 except connection_dropped_errors as e: 

377 self.connection_dropped(e, environ) 

378 except Exception as e: 

379 if self.server.passthrough_errors: 

380 raise 

381 

382 if status_sent is not None and chunk_response: 

383 self.close_connection = True 

384 

385 try: 

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

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

388 if status_sent is None: 

389 status_set = None 

390 headers_set = None 

391 execute(InternalServerError()) 

392 except Exception: 

393 pass 

394 

395 from .debug.tbtools import DebugTraceback 

396 

397 msg = DebugTraceback(e).render_traceback_text() 

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

399 

400 def handle(self) -> None: 

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

402 try: 

403 super().handle() 

404 except (ConnectionError, TimeoutError) as e: 

405 self.connection_dropped(e) 

406 except Exception as e: 

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

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

409 else: 

410 raise 

411 

412 def connection_dropped( 

413 self, error: BaseException, environ: wt.WSGIEnvironment | None = None 

414 ) -> None: 

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

416 nothing happens. 

417 """ 

418 

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

420 # All HTTP methods are handled by run_wsgi. 

421 if name.startswith("do_"): 

422 return self.run_wsgi 

423 

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

425 return getattr(super(), name) 

426 

427 def address_string(self) -> str: 

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

429 return self.environ["REMOTE_ADDR"] # type: ignore[no-any-return] 

430 

431 if not self.client_address: 

432 return "<local>" 

433 

434 return self.client_address[0] 

435 

436 def port_integer(self) -> int: 

437 return self.client_address[1] 

438 

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

440 _control_char_table = str.maketrans( 

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

442 ) 

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

444 

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

446 try: 

447 path = uri_to_iri(self.path) 

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

449 except AttributeError: 

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

451 msg = self.requestline 

452 

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

454 msg = msg.translate(self._control_char_table) 

455 code = str(code) 

456 

457 if code.startswith("1"): # 1xx - Informational 

458 msg = _ansi_style(msg, "bold") 

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

460 pass 

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

462 msg = _ansi_style(msg, "cyan") 

463 elif code.startswith("3"): # 3xx - Redirection 

464 msg = _ansi_style(msg, "green") 

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

466 msg = _ansi_style(msg, "yellow") 

467 elif code.startswith("4"): # 4xx - Client Error 

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

469 else: # 5xx, or any other response 

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

471 

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

473 

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

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

476 

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

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

479 

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

481 # an IPv6 scoped address contains "%" which breaks logging 

482 address_string = self.address_string().replace("%", "%%") 

483 _log( 

484 type, 

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

486 *args, 

487 ) 

488 

489 

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

491 codes = { 

492 "bold": 1, 

493 "red": 31, 

494 "green": 32, 

495 "yellow": 33, 

496 "magenta": 35, 

497 "cyan": 36, 

498 } 

499 

500 for style in styles: 

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

502 

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

504 

505 

506def generate_adhoc_ssl_pair( 

507 cn: str | None = None, 

508) -> tuple[Certificate, RSAPrivateKeyWithSerialization]: 

509 try: 

510 from cryptography import x509 

511 from cryptography.hazmat.backends import default_backend 

512 from cryptography.hazmat.primitives import hashes 

513 from cryptography.hazmat.primitives.asymmetric import rsa 

514 from cryptography.x509.oid import NameOID 

515 except ImportError: 

516 raise TypeError( 

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

518 ) from None 

519 

520 backend = default_backend() 

521 pkey = rsa.generate_private_key( 

522 public_exponent=65537, key_size=2048, backend=backend 

523 ) 

524 

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

526 if cn is None: 

527 cn = "*" 

528 

529 subject = x509.Name( 

530 [ 

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

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

533 ] 

534 ) 

535 

536 backend = default_backend() 

537 cert = ( 

538 x509.CertificateBuilder() 

539 .subject_name(subject) 

540 .issuer_name(subject) 

541 .public_key(pkey.public_key()) 

542 .serial_number(x509.random_serial_number()) 

543 .not_valid_before(dt.datetime.now(dt.UTC)) 

544 .not_valid_after(dt.datetime.now(dt.UTC) + dt.timedelta(days=365)) 

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

546 .add_extension( 

547 x509.SubjectAlternativeName([x509.DNSName(cn), x509.DNSName(f"*.{cn}")]), 

548 critical=False, 

549 ) 

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

551 ) 

552 return cert, pkey 

553 

554 

555def make_ssl_devcert( 

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

557) -> tuple[str, str]: 

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

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

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

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

562 ``*.host/CN=host``. 

563 

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

565 

566 .. versionadded:: 0.9 

567 

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

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

570 added for the key. 

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

572 for the `cn`. 

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

574 """ 

575 

576 if host is not None: 

577 cn = host 

578 cert, pkey = generate_adhoc_ssl_pair(cn=cn) 

579 

580 from cryptography.hazmat.primitives import serialization 

581 

582 cert_file = f"{base_path}.crt" 

583 pkey_file = f"{base_path}.key" 

584 

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

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

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

588 f.write( 

589 pkey.private_bytes( 

590 encoding=serialization.Encoding.PEM, 

591 format=serialization.PrivateFormat.TraditionalOpenSSL, 

592 encryption_algorithm=serialization.NoEncryption(), 

593 ) 

594 ) 

595 

596 return cert_file, pkey_file 

597 

598 

599def generate_adhoc_ssl_context() -> ssl.SSLContext: 

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

601 import atexit 

602 import tempfile 

603 

604 cert, pkey = generate_adhoc_ssl_pair() 

605 

606 from cryptography.hazmat.primitives import serialization 

607 

608 with tempfile.NamedTemporaryFile("wb", delete=False) as cert_file: 

609 cert_file.write(cert.public_bytes(serialization.Encoding.PEM)) 

610 

611 with tempfile.NamedTemporaryFile("wb", delete=False) as pkey_file: 

612 pkey_file.write( 

613 pkey.private_bytes( 

614 encoding=serialization.Encoding.PEM, 

615 format=serialization.PrivateFormat.TraditionalOpenSSL, 

616 encryption_algorithm=serialization.NoEncryption(), 

617 ) 

618 ) 

619 

620 atexit.register(os.remove, cert_file.name) 

621 atexit.register(os.remove, pkey_file.name) 

622 ctx = load_ssl_context(cert_file.name, pkey_file.name) 

623 return ctx 

624 

625 

626def load_ssl_context( 

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

628) -> ssl.SSLContext: 

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

630 Many parameters are directly taken from the API of 

631 :py:class:`ssl.SSLContext`. 

632 

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

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

635 will be obtained from the certificate file. 

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

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

638 """ 

639 if protocol is None: 

640 protocol = ssl.PROTOCOL_TLS_SERVER 

641 

642 ctx = ssl.SSLContext(protocol) 

643 ctx.load_cert_chain(cert_file, pkey_file) 

644 return ctx 

645 

646 

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

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

649 if error is None: 

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

651 return isinstance(error, ssl.SSLError) 

652 

653 

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

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

656 the host and port.""" 

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

658 return socket.AF_UNIX 

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

660 return socket.AF_INET6 

661 return socket.AF_INET 

662 

663 

664def get_sockaddr( 

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

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

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

668 :func:`socket.bind`.""" 

669 if family == af_unix: 

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

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

672 try: 

673 res = socket.getaddrinfo( 

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

675 ) 

676 except socket.gaierror: 

677 return host, port 

678 return res[0][4] # type: ignore[return-value] 

679 

680 

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

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

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

684 

685 :meta private: 

686 """ 

687 # arbitrary private address 

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

689 

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

691 try: 

692 s.connect((host, 58162)) 

693 except OSError: 

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

695 

696 return s.getsockname()[0] # type: ignore[no-any-return] 

697 

698 

699class BaseWSGIServer(HTTPServer): 

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

701 

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

703 """ 

704 

705 multithread = False 

706 multiprocess = False 

707 request_queue_size = LISTEN_QUEUE 

708 allow_reuse_address = True 

709 

710 def __init__( 

711 self, 

712 host: str, 

713 port: int, 

714 app: wt.WSGIApplication, 

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

716 passthrough_errors: bool = False, 

717 ssl_context: _TSSLContextArg = None, 

718 fd: int | None = None, 

719 ) -> None: 

720 if handler is None: 

721 handler = WSGIRequestHandler 

722 

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

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

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

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

727 self.multithread or self.multiprocess 

728 ): 

729 handler.protocol_version = "HTTP/1.1" 

730 

731 self.host = host 

732 self.port = port 

733 self.app = app 

734 self.passthrough_errors = passthrough_errors 

735 

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

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

738 

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

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

741 if address_family == af_unix and fd is None: 

742 server_address = t.cast(str, server_address) 

743 

744 if os.path.exists(server_address): 

745 os.unlink(server_address) 

746 

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

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

749 super().__init__( 

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

751 handler, 

752 bind_and_activate=False, 

753 ) 

754 

755 if fd is None: 

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

757 try: 

758 self.server_bind() 

759 self.server_activate() 

760 except OSError as e: 

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

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

763 self.server_close() 

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

765 

766 if e.errno == errno.EADDRINUSE: 

767 print( 

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

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

770 " port.", 

771 file=sys.stderr, 

772 ) 

773 

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

775 print( 

776 "On macOS, try searching for and disabling" 

777 " 'AirPlay Receiver' in System Settings.", 

778 file=sys.stderr, 

779 ) 

780 

781 sys.exit(1) 

782 except BaseException: 

783 self.server_close() 

784 raise 

785 else: 

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

787 # Close it to silence a ResourceWarning. 

788 self.server_close() 

789 

790 # Use the passed in socket directly. 

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

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

793 

794 if address_family != af_unix: 

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

796 self.port = self.server_address[1] 

797 

798 if ssl_context is not None: 

799 if isinstance(ssl_context, tuple): 

800 ssl_context = load_ssl_context(*ssl_context) 

801 elif ssl_context == "adhoc": 

802 ssl_context = generate_adhoc_ssl_context() 

803 

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

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

806 else: 

807 self.ssl_context = None 

808 

809 import importlib.metadata 

810 

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

812 

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

814 _log(type, message, *args) 

815 

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

817 try: 

818 super().serve_forever(poll_interval=poll_interval) 

819 except KeyboardInterrupt: 

820 pass 

821 finally: 

822 self.server_close() 

823 

824 def handle_error( 

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

826 ) -> None: 

827 if self.passthrough_errors: 

828 raise 

829 

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

831 

832 def log_startup(self) -> None: 

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

834 dev_warning = ( 

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

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

837 ) 

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

839 messages = [dev_warning] 

840 

841 if self.address_family == af_unix: 

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