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

1from __future__ import annotations

3import io

4import typing as t

5from functools import partial

6from functools import update_wrapper

8from .exceptions import ClientDisconnected

9from .exceptions import RequestEntityTooLarge

10from .sansio import utils as _sansio_utils

11from .sansio.utils import host_is_trusted # noqa: F401 # Imported as part of API

13if t.TYPE_CHECKING:

14 from _typeshed.wsgi import WSGIApplication

15 from _typeshed.wsgi import WSGIEnvironment

18def responder(f: t.Callable[..., WSGIApplication]) -> WSGIApplication:

19 """Marks a function as responder. Decorate a function with it and it

20 will automatically call the return value as WSGI application.

22 Example::

24 @responder

25 def application(environ, start_response):

26 return Response('Hello World!')

27 """

28 return update_wrapper(lambda *a: f(*a)(*a[-2:]), f)

31def get_current_url(

32 environ: WSGIEnvironment,

33 root_only: bool = False,

34 strip_querystring: bool = False,

35 host_only: bool = False,

36 trusted_hosts: t.Collection[str] | None = None,

37) -> str:

38 """Recreate the URL for a request from the parts in a WSGI

39 environment.

41 The URL is an IRI, not a URI, so it may contain Unicode characters.

42 Use :func:`~werkzeug.urls.iri_to_uri` to convert it to ASCII.

44 :param environ: The WSGI environment to get the URL parts from.

45 :param root_only: Only build the root path, don't include the

46 remaining path or query string.

47 :param strip_querystring: Don't include the query string.

48 :param host_only: Only build the scheme and host.

49 :param trusted_hosts: A list of trusted host names to validate the

50 host against.

51 """

52 parts = {

53 "scheme": environ["wsgi.url_scheme"],

54 "host": get_host(environ, trusted_hosts),

55 }

57 if not host_only:

58 parts["root_path"] = environ.get("SCRIPT_NAME", "")

60 if not root_only:

61 parts["path"] = environ.get("PATH_INFO", "")

63 if not strip_querystring:

64 parts["query_string"] = environ.get("QUERY_STRING", "").encode("latin1")

66 return _sansio_utils.get_current_url(**parts)

69def _get_server(

70 environ: WSGIEnvironment,

71) -> tuple[str, int | None] | None:

72 name = environ.get("SERVER_NAME")

74 if name is None:

75 return None

77 try:

78 port: int | None = int(environ.get("SERVER_PORT", None)) # type: ignore[arg-type]

79 except (TypeError, ValueError):

80 # unix socket

81 port = None

83 return name, port

86def get_host(

87 environ: WSGIEnvironment, trusted_hosts: t.Collection[str] | None = None

88) -> str:

89 """Get and validate a request's ``host:port`` based on the values in the

90 given WSGI environ.

92 The ``Host`` header sent by the client is preferred. Otherwise, the server's

93 configured address is used. If the server address is a Unix socket, it is

94 ignored. The port is omitted if it matches the standard HTTP or HTTPS ports.

96 The value is passed through :func:`host_is_trusted`. The host must be made

97 up of valid characters, but this does not check validity beyond that. If a

98 list of trusted domains is given, the domain must match one.

100 :param environ: The WSGI environ.

101 :param trusted_hosts: A list of trusted domains to match. These should

102 already be IDNA encoded, but will be encoded if needed. The port is

103 ignored for this check. If a name starts with a dot it will match as a

104 suffix, accepting all subdomains. If empty or ``None``, all domains are

105 allowed.

106

107 :return: Host, with port if necessary.

108 :raise .SecurityError: If the host is not trusted.

109

110 .. versionchanged:: 3.2

111 The characters of the host value are validated. The empty string is no

112 longer allowed if no header value is available.

113

114 .. versionchanged:: 3.2

115 When using the server address, Unix sockets are ignored.

116

117 .. versionchanged:: 3.1.3

118 If ``SERVER_NAME`` is IPv6, it is wrapped in ``[]``.

119 """

120 return _sansio_utils.get_host(

121 environ["wsgi.url_scheme"],

122 environ.get("HTTP_HOST"),

123 _get_server(environ),

124 trusted_hosts,

125 )

126

127

128def get_content_length(environ: WSGIEnvironment) -> int | None:

129 """Return the ``Content-Length`` header value as an int. If the header is not given

130 or the ``Transfer-Encoding`` header is ``chunked``, ``None`` is returned to indicate

131 a streaming request. If the value is not an integer, or negative, 0 is returned.

132

133 :param environ: The WSGI environ to get the content length from.

134

135 .. versionadded:: 0.9

136 """

137 return _sansio_utils.get_content_length(

138 http_content_length=environ.get("CONTENT_LENGTH"),

139 http_transfer_encoding=environ.get("HTTP_TRANSFER_ENCODING"),

140 )

141

142

143def get_input_stream(

144 environ: WSGIEnvironment,

145 safe_fallback: bool = True,

146 max_content_length: int | None = None,

147) -> t.IO[bytes]:

148 """Return the WSGI input stream, wrapped so that it may be read safely

149 without going past the ``Content-Length`` header value or

150 ``max_content_length``. This must be used to pass a safe stream to other

151 functions, which are not responsible for enforcing this limit.

152

153 If ``Content-Length`` exceeds ``max_content_length``, a

154 :exc:`RequestEntityTooLarge`` ``413 Content Too Large`` error is raised.

155

156 If the WSGI server sets ``environ["wsgi.input_terminated"]``, it indicates that the

157 server handles terminating the stream, so it is safe to read directly. For example,

158 a server that knows how to handle chunked requests safely would set this.

159

160 If ``max_content_length`` is set, it can be enforced on streams if

161 ``wsgi.input_terminated`` is set. Otherwise, an empty stream is returned unless the

162 user explicitly disables this safe fallback.

163

164 If the limit is reached before the underlying stream is exhausted (such as a file

165 that is too large, or an infinite stream), the remaining contents of the stream

166 cannot be read safely. Depending on how the server handles this, clients may show a

167 "connection reset" failure instead of seeing the 413 response.

168

169 :param environ: The WSGI environ containing the stream.

170 :param safe_fallback: Return an empty stream when ``Content-Length`` is not set.

171 Disabling this allows infinite streams, which can be a denial-of-service risk.

172 :param max_content_length: The maximum length that content-length or streaming

173 requests may not exceed.

174

175 .. versionchanged:: 2.3.2

176 ``max_content_length`` is only applied to streaming requests if the server sets

177 ``wsgi.input_terminated``.

178

179 .. versionchanged:: 2.3

180 Check ``max_content_length`` and raise an error if it is exceeded.

181

182 .. versionadded:: 0.9

183 """

184 stream = t.cast(t.IO[bytes], environ["wsgi.input"])

185 content_length = get_content_length(environ)

186

187 if content_length is not None and max_content_length is not None:

188 if content_length > max_content_length:

189 raise RequestEntityTooLarge()

190

191 # A WSGI server can set this to indicate that it terminates the input stream. In

192 # that case the stream is safe without wrapping, or can enforce a max length.

193 if "wsgi.input_terminated" in environ:

194 if max_content_length is not None:

195 # If this is moved above, it can cause the stream to hang if a read attempt

196 # is made when the client sends no data. For example, the development server

197 # does not handle buffering except for chunked encoding.

198 return t.cast(

199 t.IO[bytes], LimitedStream(stream, max_content_length, is_max=True)

200 )

201

202 return stream

203

204 # No limit given, return an empty stream unless the user explicitly allows the

205 # potentially infinite stream. An infinite stream is dangerous if it's not expected,

206 # as it can tie up a worker indefinitely.

207 if content_length is None:

208 return io.BytesIO() if safe_fallback else stream

209

210 return t.cast(t.IO[bytes], LimitedStream(stream, content_length))

211

212

213def get_path_info(environ: WSGIEnvironment) -> str:

214 """Return ``PATH_INFO`` from the WSGI environment.

215

216 :param environ: WSGI environment to get the path from.

217

218 .. versionchanged:: 3.0

219 The ``charset`` and ``errors`` parameters were removed.

220

221 .. versionadded:: 0.9

222 """

223 path: bytes = environ.get("PATH_INFO", "").encode("latin1")

224 return path.decode(errors="replace")

225

226

227class ClosingIterator:

228 """The WSGI specification requires that all middlewares and gateways

229 respect the `close` callback of the iterable returned by the application.

230 Because it is useful to add another close action to a returned iterable

231 and adding a custom iterable is a boring task this class can be used for

232 that::

233

234 return ClosingIterator(app(environ, start_response), [cleanup_session,

235 cleanup_locals])

236

237 If there is just one close function it can be passed instead of the list.

238

239 A closing iterator is not needed if the application uses response objects

240 and finishes the processing if the response is started::

241

242 try:

243 return response(environ, start_response)

244 finally:

245 cleanup_session()

246 cleanup_locals()

247 """

248

249 def __init__(

250 self,

251 iterable: t.Iterable[bytes],

252 callbacks: None

253 | (t.Callable[[], None] | t.Iterable[t.Callable[[], None]]) = None,

254 ) -> None:

255 iterator = iter(iterable)

256 self._next = t.cast(t.Callable[[], bytes], partial(next, iterator))

257 if callbacks is None:

258 callbacks = []

259 elif callable(callbacks):

260 callbacks = [callbacks]

261 else:

262 callbacks = list(callbacks)

263 iterable_close = getattr(iterable, "close", None)

264 if iterable_close:

265 callbacks.insert(0, iterable_close)

266 self._callbacks = callbacks

267

268 def __iter__(self) -> ClosingIterator:

269 return self

270

271 def __next__(self) -> bytes:

272 return self._next()

273

274 def close(self) -> None:

275 for callback in self._callbacks:

276 callback()

277

278

279def wrap_file(

280 environ: WSGIEnvironment, file: t.IO[bytes], buffer_size: int = 8192

281) -> t.Iterable[bytes]:

282 """Wraps a file. This uses the WSGI server's file wrapper if available

283 or otherwise the generic :class:`FileWrapper`.

284

285 .. versionadded:: 0.5

286

287 If the file wrapper from the WSGI server is used it's important to not

288 iterate over it from inside the application but to pass it through

289 unchanged. If you want to pass out a file wrapper inside a response

290 object you have to set :attr:`Response.direct_passthrough` to `True`.

291

292 More information about file wrappers are available in :pep:`333`.

293

294 :param file: a :class:`file`-like object with a :meth:`~file.read` method.

295 :param buffer_size: number of bytes for one iteration.

296 """

297 return environ.get("wsgi.file_wrapper", FileWrapper)( # type: ignore

298 file, buffer_size

299 )

300

301

302class FileWrapper:

303 """This class can be used to convert a :class:`file`-like object into

304 an iterable. It yields `buffer_size` blocks until the file is fully

305 read.

306

307 You should not use this class directly but rather use the

308 :func:`wrap_file` function that uses the WSGI server's file wrapper

309 support if it's available.

310

311 .. versionadded:: 0.5

312

313 If you're using this object together with a :class:`Response` you have

314 to use the `direct_passthrough` mode.

315

316 :param file: a :class:`file`-like object with a :meth:`~file.read` method.

317 :param buffer_size: number of bytes for one iteration.

318 """

319

320 def __init__(self, file: t.IO[bytes], buffer_size: int = 8192) -> None:

321 self.file = file

322 self.buffer_size = buffer_size

323

324 def close(self) -> None:

325 if hasattr(self.file, "close"):

326 self.file.close()

327

328 def seekable(self) -> bool:

329 if hasattr(self.file, "seekable"):

330 return self.file.seekable()

331 if hasattr(self.file, "seek"):

332 return True

333 return False

334

335 def seek(self, *args: t.Any) -> None:

336 if hasattr(self.file, "seek"):

337 self.file.seek(*args)

338

339 def tell(self) -> int | None:

340 if hasattr(self.file, "tell"):

341 return self.file.tell()

342 return None

343

344 def __iter__(self) -> FileWrapper:

345 return self

346

347 def __next__(self) -> bytes:

348 data = self.file.read(self.buffer_size)

349 if data:

350 return data

351 raise StopIteration()

352

353

354class _RangeWrapper:

355 # private for now, but should we make it public in the future ?

356

357 """This class can be used to convert an iterable object into

358 an iterable that will only yield a piece of the underlying content.

359 It yields blocks until the underlying stream range is fully read.

360 The yielded blocks will have a size that can't exceed the original

361 iterator defined block size, but that can be smaller.

362

363 If you're using this object together with a :class:`Response` you have

364 to use the `direct_passthrough` mode.

365

366 :param iterable: an iterable object with a :meth:`__next__` method.

367 :param start_byte: byte from which read will start.

368 :param byte_range: how many bytes to read.

369 """

370

371 def __init__(

372 self,

373 iterable: t.Iterable[bytes] | t.IO[bytes],

374 start_byte: int = 0,

375 byte_range: int | None = None,

376 ):

377 self.iterable = iter(iterable)

378 self.byte_range = byte_range

379 self.start_byte = start_byte

380 self.end_byte = None

381

382 if byte_range is not None:

383 self.end_byte = start_byte + byte_range

384

385 self.read_length = 0

386 self.seekable = hasattr(iterable, "seekable") and iterable.seekable()

387 self.end_reached = False

388

389 def __iter__(self) -> _RangeWrapper:

390 return self

391

392 def _next_chunk(self) -> bytes:

393 try:

394 chunk = next(self.iterable)

395 self.read_length += len(chunk)

396 return chunk

397 except StopIteration:

398 self.end_reached = True

399 raise

400

401 def _first_iteration(self) -> tuple[bytes | None, int]:

402 chunk = None

403 if self.seekable:

404 self.iterable.seek(self.start_byte) # type: ignore

405 self.read_length = self.iterable.tell() # type: ignore

406 contextual_read_length = self.read_length

407 else:

408 while self.read_length <= self.start_byte:

409 chunk = self._next_chunk()

410 if chunk is not None:

411 chunk = chunk[self.start_byte - self.read_length :]

412 contextual_read_length = self.start_byte

413 return chunk, contextual_read_length

414

415 def _next(self) -> bytes:

416 if self.end_reached:

417 raise StopIteration()

418 chunk = None

419 contextual_read_length = self.read_length

420 if self.read_length == 0:

421 chunk, contextual_read_length = self._first_iteration()

422 if chunk is None:

423 chunk = self._next_chunk()

424 if self.end_byte is not None and self.read_length >= self.end_byte:

425 self.end_reached = True

426 return chunk[: self.end_byte - contextual_read_length]

427 return chunk

428

429 def __next__(self) -> bytes:

430 chunk = self._next()

431 if chunk:

432 return chunk

433 self.end_reached = True

434 raise StopIteration()

435

436 def close(self) -> None:

437 if hasattr(self.iterable, "close"):

438 self.iterable.close()

439

440

441class LimitedStream(io.RawIOBase):

442 """Wrap a stream so that it doesn't read more than a given limit. This is used to

443 limit ``wsgi.input`` to the ``Content-Length`` header value or

444 :attr:`.Request.max_content_length`.

445

446 When attempting to read after the limit has been reached, :meth:`on_exhausted` is

447 called. When the limit is a maximum, this raises :exc:`.RequestEntityTooLarge`.

448

449 If reading from the stream returns zero bytes or raises an error,

450 :meth:`on_disconnect` is called, which raises :exc:`.ClientDisconnected`. When the

451 limit is a maximum and zero bytes were read, no error is raised, since it may be the

452 end of the stream.

453

454 If the limit is reached before the underlying stream is exhausted (such as a file

455 that is too large, or an infinite stream), the remaining contents of the stream

456 cannot be read safely. Depending on how the server handles this, clients may show a

457 "connection reset" failure instead of seeing the 413 response.

458

459 :param stream: The stream to read from. Must be a readable binary IO object.

460 :param limit: The limit in bytes to not read past. Should be either the

461 ``Content-Length`` header value or ``request.max_content_length``.

462 :param is_max: Whether the given ``limit`` is ``request.max_content_length`` instead

463 of the ``Content-Length`` header value. This changes how exhausted and

464 disconnect events are handled.

465

466 .. versionchanged:: 2.3

467 Handle ``max_content_length`` differently than ``Content-Length``.

468

469 .. versionchanged:: 2.3

470 Implements ``io.RawIOBase`` rather than ``io.IOBase``.

471 """

472

473 def __init__(self, stream: t.IO[bytes], limit: int, is_max: bool = False) -> None:

474 self._stream = stream

475 self._pos = 0

476 self.limit = limit

477 self._limit_is_max = is_max

478

479 @property

480 def is_exhausted(self) -> bool:

481 """Whether the current stream position has reached the limit."""

482 return self._pos >= self.limit

483

484 def on_exhausted(self) -> None:

485 """Called when attempting to read after the limit has been reached.

486

487 The default behavior is to do nothing, unless the limit is a maximum, in which

488 case it raises :exc:`.RequestEntityTooLarge`.

489

490 .. versionchanged:: 2.3

491 Raises ``RequestEntityTooLarge`` if the limit is a maximum.

492

493 .. versionchanged:: 2.3

494 Any return value is ignored.

495 """

496 if self._limit_is_max:

497 raise RequestEntityTooLarge()

498

499 def on_disconnect(self, error: Exception | None = None) -> None:

500 """Called when an attempted read receives zero bytes before the limit was

501 reached. This indicates that the client disconnected before sending the full

502 request body.

503

504 The default behavior is to raise :exc:`.ClientDisconnected`, unless the limit is

505 a maximum and no error was raised.

506

507 .. versionchanged:: 2.3

508 Added the ``error`` parameter. Do nothing if the limit is a maximum and no

509 error was raised.

510

511 .. versionchanged:: 2.3

512 Any return value is ignored.

513 """

514 if not self._limit_is_max or error is not None:

515 raise ClientDisconnected()

516

517 # If the limit is a maximum, then we may have read zero bytes because the

518 # streaming body is complete. There's no way to distinguish that from the

519 # client disconnecting early.

520

521 def exhaust(self) -> bytes:

522 """Exhaust the stream by reading until the limit is reached or the client

523 disconnects, returning the remaining data.

524

525 .. versionchanged:: 2.3

526 Return the remaining data.

527

528 .. versionchanged:: 2.2.3

529 Handle case where wrapped stream returns fewer bytes than requested.

530 """

531 if not self.is_exhausted:

532 return self.readall()

533

534 return b""

535

536 def readinto(self, b: bytearray) -> int | None: # type: ignore[override]

537 size = len(b)

538 remaining = self.limit - self._pos

539

540 if remaining <= 0:

541 self.on_exhausted()

542 return 0

543

544 if hasattr(self._stream, "readinto"):

545 # Use stream.readinto if it's available.

546 if size <= remaining:

547 # The size fits in the remaining limit, use the buffer directly.

548 try:

549 out_size: int | None = self._stream.readinto(b)

550 except (OSError, ValueError) as e:

551 self.on_disconnect(error=e)

552 return 0

553 else:

554 # Use a temp buffer with the remaining limit as the size.

555 temp_b = bytearray(remaining)

556

557 try:

558 out_size = self._stream.readinto(temp_b)

559 except (OSError, ValueError) as e:

560 self.on_disconnect(error=e)

561 return 0

562

563 if out_size:

564 # Need to slice both sides. Math here is complicated.

565 # size=10, remaining=5, out_size=3, remaining > out_size

566 # Without slicing temp_b (b[:s] = tb), would try to

567 # put 5 bytes into 3 bytes, causing b to resize to 12.

568 b[:out_size] = temp_b[:out_size]

569 else:

570 # WSGI requires that stream.read is available.

571 try:

572 data = self._stream.read(min(size, remaining))

573 except (OSError, ValueError) as e:

574 self.on_disconnect(error=e)

575 return 0

576

577 out_size = len(data)

578 b[:out_size] = data

579

580 if not out_size:

581 # Read zero bytes from the stream.

582 self.on_disconnect()

583 return 0

584

585 self._pos += out_size

586 return out_size

587

588 def readall(self) -> bytes:

589 if self.is_exhausted:

590 self.on_exhausted()

591 return b""

592

593 out = bytearray()

594

595 # The parent implementation uses "while True", which results in an extra read.

596 while not self.is_exhausted:

597 data = self.read(1024 * 64)

598

599 # Stream may return empty before a max limit is reached.

600 if not data:

601 break

602

603 out.extend(data)

604

605 return bytes(out)

606

607 def tell(self) -> int:

608 """Return the current stream position.

609

610 .. versionadded:: 0.9

611 """

612 return self._pos

613

614 def readable(self) -> bool:

615 return True