Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/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""" 

14 

15from __future__ import annotations 

16 

17import errno 

18import io 

19import os 

20import selectors 

21import socket 

22import socketserver 

23import sys 

24import typing as t 

25from datetime import datetime as dt 

26from datetime import timedelta 

27from datetime import timezone 

28from http.server import BaseHTTPRequestHandler 

29from http.server import HTTPServer 

30from urllib.parse import unquote 

31from urllib.parse import urlsplit 

32 

33from ._internal import _log 

34from ._internal import _wsgi_encoding_dance 

35from .exceptions import InternalServerError 

36from .http import parse_set_header 

37from .urls import uri_to_iri 

38 

39try: 

40 import ssl 

41 

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

43 ConnectionError, 

44 socket.timeout, 

45 ssl.SSLEOFError, 

46 ) 

47except ImportError: 

48 

49 class _SslDummy: 

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

51 raise RuntimeError( # noqa: B904 

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

53 " compiled with SSL/TLS support." 

54 ) 

55 

56 ssl = _SslDummy() # type: ignore 

57 connection_dropped_errors = (ConnectionError, socket.timeout) 

58 

59_log_add_style = True 

60 

61if os.name == "nt": 

62 try: 

63 __import__("colorama") 

64 except ImportError: 

65 _log_add_style = False 

66 

67can_fork = hasattr(os, "fork") 

68 

69if can_fork: 

70 ForkingMixIn = socketserver.ForkingMixIn 

71else: 

72 

73 class ForkingMixIn: # type: ignore 

74 pass 

75 

76 

77try: 

78 af_unix = socket.AF_UNIX 

79except AttributeError: 

80 af_unix = None # type: ignore 

81 

82LISTEN_QUEUE = 128 

83 

84_TSSLContextArg = t.Optional[ 

85 t.Union["ssl.SSLContext", tuple[str, t.Optional[str]], t.Literal["adhoc"]] 

86] 

87 

88if t.TYPE_CHECKING: 

89 from _typeshed.wsgi import WSGIApplication 

90 from _typeshed.wsgi import WSGIEnvironment 

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

92 RSAPrivateKeyWithSerialization, 

93 ) 

94 from cryptography.x509 import Certificate 

95 

96 

97class DechunkedInput(io.RawIOBase): 

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

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

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

101 """ 

102 

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

104 self._rfile = rfile 

105 self._done = False 

106 self._len = 0 

107 

108 def readable(self) -> bool: 

109 return True 

110 

111 def read_chunk_len(self) -> int: 

112 try: 

113 line = self._rfile.readline(100).decode("latin1") 

114 _len = int(line.strip(" \t\r\n"), 16) 

115 except ValueError as e: 

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

117 if _len < 0: 

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

119 return _len 

120 

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

122 read = 0 

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

124 if self._len == 0: 

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

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

127 self._len = self.read_chunk_len() 

128 

129 if self._len == 0: 

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

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

132 self._done = True 

133 

134 if self._len > 0: 

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

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

137 # reset self._len to 0. 

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

139 

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

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

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

143 if read + n > len(buf): 

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

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

146 read = len(buf) 

147 else: 

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

149 self._len -= n 

150 read += n 

151 

152 if self._len == 0: 

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

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

155 terminator = self._rfile.readline(2) 

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

157 raise OSError("Missing chunk terminating newline") 

158 

159 return read 

160 

161 

162class WSGIRequestHandler(BaseHTTPRequestHandler): 

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

164 

165 server: BaseWSGIServer 

166 

167 @property 

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

169 return self.server._server_version 

170 

171 def make_environ(self) -> WSGIEnvironment: 

172 request_url = urlsplit(self.path) 

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

174 

175 if not self.client_address: 

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

177 elif isinstance(self.client_address, str): 

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

179 

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

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

182 # netloc, prepend it to the path again. 

183 if not request_url.scheme and request_url.netloc: 

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

185 else: 

186 path_info = request_url.path 

187 

188 path_info = unquote(path_info) 

189 

190 environ: WSGIEnvironment = { 

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

192 "wsgi.url_scheme": url_scheme, 

193 "wsgi.input": self.rfile, 

194 "wsgi.errors": sys.stderr, 

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

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

197 "wsgi.run_once": False, 

198 "werkzeug.socket": self.connection, 

199 "SERVER_SOFTWARE": self.server_version, 

200 "REQUEST_METHOD": self.command, 

201 "SCRIPT_NAME": "", 

202 "PATH_INFO": _wsgi_encoding_dance(path_info), 

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

204 # Non-standard, added by mod_wsgi, uWSGI 

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

206 # Non-standard, added by gunicorn 

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

208 "REMOTE_ADDR": self.address_string(), 

209 "REMOTE_PORT": self.port_integer(), 

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

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

212 "SERVER_PROTOCOL": self.request_version, 

213 } 

214 

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

216 if "_" in key: 

217 continue 

218 

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

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

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

222 key = f"HTTP_{key}" 

223 if key in environ: 

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

225 environ[key] = value 

226 

227 if "chunked" in parse_set_header(environ.get("HTTP_TRANSFER_ENCODING")): 

228 environ["wsgi.input_terminated"] = True 

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

230 

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

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

233 if request_url.scheme and request_url.netloc: 

234 environ["HTTP_HOST"] = request_url.netloc 

235 

236 try: 

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

238 # what Nginx or Apache could return. 

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

240 if peer_cert is not None: 

241 # Nginx and Apache use PEM format. 

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

243 except ValueError: 

244 # SSL handshake hasn't finished. 

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

246 except AttributeError: 

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

248 pass 

249 

250 return environ 

251 

252 def run_wsgi(self) -> None: 

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

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

255 

256 self.environ = environ = self.make_environ() 

257 status_set: str | None = None 

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

259 status_sent: str | None = None 

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

261 chunk_response: bool = False 

262 

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

264 nonlocal status_sent, headers_sent, chunk_response 

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

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

267 if status_sent is None: 

268 status_sent = status_set 

269 headers_sent = headers_set 

270 try: 

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

272 except ValueError: 

273 code_str, msg = status_sent, "" 

274 code = int(code_str) 

275 self.send_response(code, msg) 

276 header_keys = set() 

277 for key, value in headers_sent: 

278 self.send_header(key, value) 

279 header_keys.add(key.lower()) 

280 

281 # Use chunked transfer encoding if there is no content 

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

283 # responses and HEAD requests are also excluded, which 

284 # is the more conservative behavior and matches other 

285 # parts of the code. 

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

287 if ( 

288 not ( 

289 "content-length" in header_keys 

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

291 or (100 <= code < 200) 

292 or code in {204, 304} 

293 ) 

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

295 ): 

296 chunk_response = True 

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

298 

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

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

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

302 # drain the stream before the next request line. 

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

304 self.end_headers() 

305 

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

307 

308 if data: 

309 if chunk_response: 

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

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

312 

313 self.wfile.write(data) 

314 

315 if chunk_response: 

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

317 

318 self.wfile.flush() 

319 

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

321 nonlocal status_set, headers_set 

322 if exc_info: 

323 try: 

324 if headers_sent: 

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

326 finally: 

327 exc_info = None 

328 elif headers_set: 

329 raise AssertionError("Headers already set") 

330 status_set = status 

331 headers_set = headers 

332 return write 

333 

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

335 application_iter = app(environ, start_response) 

336 try: 

337 for data in application_iter: 

338 write(data) 

339 if not headers_sent: 

340 write(b"") 

341 if chunk_response: 

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

343 finally: 

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

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

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

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

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

349 # connection we can read everything. 

350 selector = selectors.DefaultSelector() 

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

352 total_size = 0 

353 total_reads = 0 

354 

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

356 # time to continue sending its data. 

357 while selector.select(timeout=0.01): 

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

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

360 total_size += len(data) 

361 total_reads += 1 

362 

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

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

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

366 break 

367 

368 selector.close() 

369 

370 if hasattr(application_iter, "close"): 

371 application_iter.close() 

372 

373 try: 

374 execute(self.server.app) 

375 except connection_dropped_errors as e: 

376 self.connection_dropped(e, environ) 

377 except Exception as e: 

378 if self.server.passthrough_errors: 

379 raise 

380 

381 if status_sent is not None and chunk_response: 

382 self.close_connection = True 

383 

384 try: 

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

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

387 if status_sent is None: 

388 status_set = None 

389 headers_set = None 

390 execute(InternalServerError()) 

391 except Exception: 

392 pass 

393 

394 from .debug.tbtools import DebugTraceback 

395 

396 msg = DebugTraceback(e).render_traceback_text() 

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

398 

399 def handle(self) -> None: 

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

401 try: 

402 super().handle() 

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

404 self.connection_dropped(e) 

405 except Exception as e: 

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

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

408 else: 

409 raise 

410 

411 def connection_dropped( 

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

413 ) -> None: 

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

415 nothing happens. 

416 """ 

417 

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

419 # All HTTP methods are handled by run_wsgi. 

420 if name.startswith("do_"): 

421 return self.run_wsgi 

422 

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

424 return getattr(super(), name) 

425 

426 def address_string(self) -> str: 

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

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

429 

430 if not self.client_address: 

431 return "<local>" 

432 

433 return self.client_address[0] 

434 

435 def port_integer(self) -> int: 

436 return self.client_address[1] 

437 

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

439 _control_char_table = str.maketrans( 

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

441 ) 

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

443 

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

445 try: 

446 path = uri_to_iri(self.path) 

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

448 except AttributeError: 

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

450 msg = self.requestline 

451 

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

453 msg = msg.translate(self._control_char_table) 

454 code = str(code) 

455 

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

457 msg = _ansi_style(msg, "bold") 

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

459 pass 

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

461 msg = _ansi_style(msg, "cyan") 

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

463 msg = _ansi_style(msg, "green") 

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

465 msg = _ansi_style(msg, "yellow") 

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

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

468 else: # 5xx, or any other response 

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

470 

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

472 

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

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

475 

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

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

478 

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

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

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

482 _log( 

483 type, 

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

485 *args, 

486 ) 

487 

488 

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

490 if not _log_add_style: 

491 return value 

492 

493 codes = { 

494 "bold": 1, 

495 "red": 31, 

496 "green": 32, 

497 "yellow": 33, 

498 "magenta": 35, 

499 "cyan": 36, 

500 } 

501 

502 for style in styles: 

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

504 

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

506 

507 

508def generate_adhoc_ssl_pair( 

509 cn: str | None = None, 

510) -> tuple[Certificate, RSAPrivateKeyWithSerialization]: 

511 try: 

512 from cryptography import x509 

513 from cryptography.hazmat.backends import default_backend 

514 from cryptography.hazmat.primitives import hashes 

515 from cryptography.hazmat.primitives.asymmetric import rsa 

516 from cryptography.x509.oid import NameOID 

517 except ImportError: 

518 raise TypeError( 

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

520 ) from None 

521 

522 backend = default_backend() 

523 pkey = rsa.generate_private_key( 

524 public_exponent=65537, key_size=2048, backend=backend 

525 ) 

526 

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

528 if cn is None: 

529 cn = "*" 

530 

531 subject = x509.Name( 

532 [ 

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

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

535 ] 

536 ) 

537 

538 backend = default_backend() 

539 cert = ( 

540 x509.CertificateBuilder() 

541 .subject_name(subject) 

542 .issuer_name(subject) 

543 .public_key(pkey.public_key()) 

544 .serial_number(x509.random_serial_number()) 

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

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

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

548 .add_extension( 

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

550 critical=False, 

551 ) 

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

553 ) 

554 return cert, pkey 

555 

556 

557def make_ssl_devcert( 

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

559) -> tuple[str, str]: 

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

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

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

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

564 ``*.host/CN=host``. 

565 

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

567 

568 .. versionadded:: 0.9 

569 

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

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

572 added for the key. 

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

574 for the `cn`. 

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

576 """ 

577 

578 if host is not None: 

579 cn = host 

580 cert, pkey = generate_adhoc_ssl_pair(cn=cn) 

581 

582 from cryptography.hazmat.primitives import serialization 

583 

584 cert_file = f"{base_path}.crt" 

585 pkey_file = f"{base_path}.key" 

586 

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

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

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

590 f.write( 

591 pkey.private_bytes( 

592 encoding=serialization.Encoding.PEM, 

593 format=serialization.PrivateFormat.TraditionalOpenSSL, 

594 encryption_algorithm=serialization.NoEncryption(), 

595 ) 

596 ) 

597 

598 return cert_file, pkey_file 

599 

600 

601def generate_adhoc_ssl_context() -> ssl.SSLContext: 

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

603 import atexit 

604 import tempfile 

605 

606 cert, pkey = generate_adhoc_ssl_pair() 

607 

608 from cryptography.hazmat.primitives import serialization 

609 

610 cert_handle, cert_file = tempfile.mkstemp() 

611 pkey_handle, pkey_file = tempfile.mkstemp() 

612 atexit.register(os.remove, pkey_file) 

613 atexit.register(os.remove, cert_file) 

614 

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

616 os.write( 

617 pkey_handle, 

618 pkey.private_bytes( 

619 encoding=serialization.Encoding.PEM, 

620 format=serialization.PrivateFormat.TraditionalOpenSSL, 

621 encryption_algorithm=serialization.NoEncryption(), 

622 ), 

623 ) 

624 

625 os.close(cert_handle) 

626 os.close(pkey_handle) 

627 ctx = load_ssl_context(cert_file, pkey_file) 

628 return ctx 

629 

630 

631def load_ssl_context( 

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

633) -> ssl.SSLContext: 

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

635 Many parameters are directly taken from the API of 

636 :py:class:`ssl.SSLContext`. 

637 

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

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

640 will be obtained from the certificate file. 

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

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

643 """ 

644 if protocol is None: 

645 protocol = ssl.PROTOCOL_TLS_SERVER 

646 

647 ctx = ssl.SSLContext(protocol) 

648 ctx.load_cert_chain(cert_file, pkey_file) 

649 return ctx 

650 

651 

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

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

654 if error is None: 

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

656 return isinstance(error, ssl.SSLError) 

657 

658 

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

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

661 the host and port.""" 

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

663 return socket.AF_UNIX 

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

665 return socket.AF_INET6 

666 return socket.AF_INET 

667 

668 

669def get_sockaddr( 

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

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

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

673 :func:`socket.bind`.""" 

674 if family == af_unix: 

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

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

677 try: 

678 res = socket.getaddrinfo( 

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

680 ) 

681 except socket.gaierror: 

682 return host, port 

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

684 

685 

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

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

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

689 

690 :meta private: 

691 """ 

692 # arbitrary private address 

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

694 

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

696 try: 

697 s.connect((host, 58162)) 

698 except OSError: 

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

700 

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

702 

703 

704class BaseWSGIServer(HTTPServer): 

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

706 

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

708 """ 

709 

710 multithread = False 

711 multiprocess = False 

712 request_queue_size = LISTEN_QUEUE 

713 allow_reuse_address = True 

714 

715 def __init__( 

716 self, 

717 host: str, 

718 port: int, 

719 app: WSGIApplication, 

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

721 passthrough_errors: bool = False, 

722 ssl_context: _TSSLContextArg | None = None, 

723 fd: int | None = None, 

724 ) -> None: 

725 if handler is None: 

726 handler = WSGIRequestHandler 

727 

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

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

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

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

732 self.multithread or self.multiprocess 

733 ): 

734 handler.protocol_version = "HTTP/1.1" 

735 

736 self.host = host 

737 self.port = port 

738 self.app = app 

739 self.passthrough_errors = passthrough_errors 

740 

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

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

743 

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

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

746 if address_family == af_unix and fd is None: 

747 server_address = t.cast(str, server_address) 

748 

749 if os.path.exists(server_address): 

750 os.unlink(server_address) 

751 

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

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

754 super().__init__( 

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

756 handler, 

757 bind_and_activate=False, 

758 ) 

759 

760 if fd is None: 

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

762 try: 

763 self.server_bind() 

764 self.server_activate() 

765 except OSError as e: 

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

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

768 self.server_close() 

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

770 

771 if e.errno == errno.EADDRINUSE: 

772 print( 

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

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

775 " port.", 

776 file=sys.stderr, 

777 ) 

778 

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

780 print( 

781 "On macOS, try searching for and disabling" 

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

783 file=sys.stderr, 

784 ) 

785 

786 sys.exit(1) 

787 except BaseException: 

788 self.server_close() 

789 raise 

790 else: 

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

792 # Close it to silence a ResourceWarning. 

793 self.server_close() 

794 

795 # Use the passed in socket directly. 

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

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

798 

799 if address_family != af_unix: 

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

801 self.port = self.server_address[1] 

802 

803 if ssl_context is not None: 

804 if isinstance(ssl_context, tuple): 

805 ssl_context = load_ssl_context(*ssl_context) 

806 elif ssl_context == "adhoc": 

807 ssl_context = generate_adhoc_ssl_context() 

808 

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

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

811 else: 

812 self.ssl_context = None 

813 

814 import importlib.metadata 

815 

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

817 

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

819 _log(type, message, *args) 

820 

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

822 try: 

823 super().serve_forever(poll_interval=poll_interval) 

824 except KeyboardInterrupt: 

825 pass 

826 finally: 

827 self.server_close() 

828 

829 def handle_error( 

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

831 ) -> None: 

832 if self.passthrough_errors: 

833 raise 

834 

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

836 

837 def log_startup(self) -> None: 

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

839 dev_warning = ( 

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

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

842 ) 

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

844 messages = [dev_warning] 

845 

846 if self.address_family == af_unix: 

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