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

1from __future__ import annotations

3import hashlib

4import http.cookies

5import json

6import os

7import stat

8import sys

9from collections.abc import AsyncIterable, Awaitable, Callable, Iterable, Mapping, Sequence

10from datetime import datetime

11from email.utils import format_datetime, formatdate

12from functools import partial

13from mimetypes import guess_type

14from secrets import token_hex

15from typing import Any, Literal

16from urllib.parse import quote

18import anyio

19import anyio.to_thread

21from starlette._utils import create_collapsing_task_group

22from starlette.background import BackgroundTask

23from starlette.concurrency import iterate_in_threadpool

24from starlette.datastructures import URL, Headers, MutableHeaders

25from starlette.requests import ClientDisconnect

26from starlette.types import Message, Receive, Scope, Send

29class Response:

30 media_type = None

31 charset = "utf-8"

33 def __init__(

34 self,

35 content: Any = None,

36 status_code: int = 200,

37 headers: Mapping[str, str] | None = None,

38 media_type: str | None = None,

39 background: BackgroundTask | None = None,

40 ) -> None:

41 self.status_code = status_code

42 if media_type is not None:

43 self.media_type = media_type

44 self.background = background

45 self.body = self.render(content)

46 self.init_headers(headers)

48 def render(self, content: Any) -> bytes | memoryview:

49 if content is None:

50 return b""

51 if isinstance(content, bytes | memoryview):

52 return content

53 return content.encode(self.charset) # type: ignore

55 def init_headers(self, headers: Mapping[str, str] | None = None) -> None:

56 if headers is None:

57 raw_headers: list[tuple[bytes, bytes]] = []

58 populate_content_length = True

59 populate_content_type = True

60 else:

61 raw_headers = [(k.lower().encode("latin-1"), v.encode("latin-1")) for k, v in headers.items()]

62 keys = [h[0] for h in raw_headers]

63 populate_content_length = b"content-length" not in keys

64 populate_content_type = b"content-type" not in keys

66 body = getattr(self, "body", None)

67 if (

68 body is not None

69 and populate_content_length

70 and not (self.status_code < 200 or self.status_code in (204, 304))

71 ):

72 content_length = str(len(body))

73 raw_headers.append((b"content-length", content_length.encode("latin-1")))

75 content_type = self.media_type

76 if content_type is not None and populate_content_type:

77 if content_type.startswith("text/") and "charset=" not in content_type.lower():

78 content_type += "; charset=" + self.charset

79 raw_headers.append((b"content-type", content_type.encode("latin-1")))

81 self.raw_headers = raw_headers

83 @property

84 def headers(self) -> MutableHeaders:

85 if not hasattr(self, "_headers"):

86 self._headers = MutableHeaders(raw=self.raw_headers)

87 return self._headers

89 def set_cookie(

90 self,

91 key: str,

92 value: str = "",

93 max_age: int | None = None,

94 expires: datetime | str | int | None = None,

95 path: str | None = "/",

96 domain: str | None = None,

97 secure: bool = False,

98 httponly: bool = False,

99 samesite: Literal["lax", "strict", "none"] | None = "lax",

100 partitioned: bool = False,

101 ) -> None:

102 cookie: http.cookies.BaseCookie[str] = http.cookies.SimpleCookie()

103 cookie[key] = value

104 if max_age is not None:

105 cookie[key]["max-age"] = max_age

106 if expires is not None:

107 if isinstance(expires, datetime):

108 cookie[key]["expires"] = format_datetime(expires, usegmt=True)

109 else:

110 cookie[key]["expires"] = expires

111 if path is not None:

112 cookie[key]["path"] = path

113 if domain is not None:

114 cookie[key]["domain"] = domain

115 if secure:

116 cookie[key]["secure"] = True

117 if httponly:

118 cookie[key]["httponly"] = True

119 if samesite is not None:

120 assert samesite.lower() in [

121 "strict",

122 "lax",

123 "none",

124 ], "samesite must be either 'strict', 'lax' or 'none'"

125 cookie[key]["samesite"] = samesite

126 if partitioned:

127 if sys.version_info < (3, 14):

128 raise ValueError("Partitioned cookies are only supported in Python 3.14 and above.") # pragma: no cover

129 cookie[key]["partitioned"] = True # pragma: no cover

130

131 cookie_val = cookie.output(header="").strip()

132 self.raw_headers.append((b"set-cookie", cookie_val.encode("latin-1")))

133

134 def delete_cookie(

135 self,

136 key: str,

137 path: str = "/",

138 domain: str | None = None,

139 secure: bool = False,

140 httponly: bool = False,

141 samesite: Literal["lax", "strict", "none"] | None = "lax",

142 ) -> None:

143 self.set_cookie(

144 key,

145 max_age=0,

146 expires=0,

147 path=path,

148 domain=domain,

149 secure=secure,

150 httponly=httponly,

151 samesite=samesite,

152 )

153

154 def _wrap_websocket_denial_send(self, send: Send) -> Send:

155 async def wrapped(message: Message) -> None:

156 message_type = message["type"]

157 if message_type in {"http.response.start", "http.response.body"}: # pragma: no branch

158 message = {**message, "type": "websocket." + message_type}

159 await send(message)

160

161 return wrapped

162

163 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

164 if scope["type"] == "websocket":

165 send = self._wrap_websocket_denial_send(send)

166 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})

167 await send({"type": "http.response.body", "body": self.body})

168

169 if self.background is not None:

170 await self.background()

171

172

173class HTMLResponse(Response):

174 media_type = "text/html"

175

176

177class PlainTextResponse(Response):

178 media_type = "text/plain"

179

180

181class JSONResponse(Response):

182 media_type = "application/json"

183

184 def __init__(

185 self,

186 content: Any,

187 status_code: int = 200,

188 headers: Mapping[str, str] | None = None,

189 media_type: str | None = None,

190 background: BackgroundTask | None = None,

191 ) -> None:

192 super().__init__(content, status_code, headers, media_type, background)

193

194 def render(self, content: Any) -> bytes:

195 return json.dumps(

196 content,

197 ensure_ascii=False,

198 allow_nan=False,

199 indent=None,

200 separators=(",", ":"),

201 ).encode("utf-8")

202

203

204class RedirectResponse(Response):

205 def __init__(

206 self,

207 url: str | URL,

208 status_code: int = 307,

209 headers: Mapping[str, str] | None = None,

210 background: BackgroundTask | None = None,

211 ) -> None:

212 super().__init__(content=b"", status_code=status_code, headers=headers, background=background)

213 self.headers["location"] = quote(str(url), safe=":/%#?=@[]!$&'()*+,;")

214

215

216Content = str | bytes | memoryview

217SyncContentStream = Iterable[Content]

218AsyncContentStream = AsyncIterable[Content]

219ContentStream = AsyncContentStream | SyncContentStream

220

221

222class StreamingResponse(Response):

223 body_iterator: AsyncContentStream

224

225 def __init__(

226 self,

227 content: ContentStream,

228 status_code: int = 200,

229 headers: Mapping[str, str] | None = None,

230 media_type: str | None = None,

231 background: BackgroundTask | None = None,

232 ) -> None:

233 if isinstance(content, AsyncIterable):

234 self.body_iterator = content

235 else:

236 self.body_iterator = iterate_in_threadpool(content)

237 self.status_code = status_code

238 self.media_type = self.media_type if media_type is None else media_type

239 self.background = background

240 self.init_headers(headers)

241

242 async def listen_for_disconnect(self, receive: Receive) -> None:

243 while True:

244 message = await receive()

245 if message["type"] == "http.disconnect":

246 break

247

248 async def stream_response(self, send: Send) -> None:

249 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})

250 async for chunk in self.body_iterator:

251 if not isinstance(chunk, bytes | memoryview):

252 chunk = chunk.encode(self.charset)

253 await send({"type": "http.response.body", "body": chunk, "more_body": True})

254

255 await send({"type": "http.response.body", "body": b"", "more_body": False})

256

257 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

258 if scope["type"] == "websocket":

259 send = self._wrap_websocket_denial_send(send)

260 await self.stream_response(send)

261 if self.background is not None:

262 await self.background()

263 return

264

265 spec_version = tuple(map(int, scope.get("asgi", {}).get("spec_version", "2.0").split(".")))

266

267 if spec_version >= (2, 4):

268 try:

269 await self.stream_response(send)

270 except OSError:

271 raise ClientDisconnect()

272 else:

273 async with create_collapsing_task_group() as task_group:

274

275 async def wrap(func: Callable[[], Awaitable[None]]) -> None:

276 await func()

277 task_group.cancel_scope.cancel()

278

279 task_group.start_soon(wrap, partial(self.stream_response, send))

280 await wrap(partial(self.listen_for_disconnect, receive))

281

282 if self.background is not None:

283 await self.background()

284

285

286class MalformedRangeHeader(Exception):

287 def __init__(self, content: str = "Malformed range header.") -> None:

288 self.content = content

289

290

291class RangeNotSatisfiable(Exception):

292 def __init__(self, max_size: int) -> None:

293 self.max_size = max_size

294

295

296class FileResponse(Response):

297 chunk_size = 64 * 1024

298

299 def __init__(

300 self,

301 path: str | os.PathLike[str],

302 status_code: int = 200,

303 headers: Mapping[str, str] | None = None,

304 media_type: str | None = None,

305 background: BackgroundTask | None = None,

306 filename: str | None = None,

307 stat_result: os.stat_result | None = None,

308 content_disposition_type: str = "attachment",

309 ) -> None:

310 self.path = path

311 self.status_code = status_code

312 self.filename = filename

313 if media_type is None:

314 media_type = guess_type(filename or path)[0] or "application/octet-stream"

315 self.media_type = media_type

316 self.background = background

317 self.init_headers(headers)

318 self.headers.setdefault("accept-ranges", "bytes")

319 if self.filename is not None:

320 content_disposition_filename = quote(self.filename)

321 if content_disposition_filename != self.filename:

322 content_disposition = f"{content_disposition_type}; filename*=utf-8''{content_disposition_filename}"

323 else:

324 content_disposition = f'{content_disposition_type}; filename="{self.filename}"'

325 self.headers.setdefault("content-disposition", content_disposition)

326 self.stat_result = stat_result

327 if stat_result is not None:

328 self.set_stat_headers(stat_result)

329

330 def set_stat_headers(self, stat_result: os.stat_result) -> None:

331 content_length = str(stat_result.st_size)

332 last_modified = formatdate(stat_result.st_mtime, usegmt=True)

333 etag_base = str(stat_result.st_mtime) + "-" + str(stat_result.st_size)

334 etag = f'"{hashlib.md5(etag_base.encode(), usedforsecurity=False).hexdigest()}"'

335

336 self.headers.setdefault("content-length", content_length)

337 self.headers.setdefault("last-modified", last_modified)

338 self.headers.setdefault("etag", etag)

339

340 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

341 scope_type = scope["type"]

342 send_header_only = scope_type == "http" and scope["method"].upper() == "HEAD"

343 send_pathsend = scope_type == "http" and "http.response.pathsend" in scope.get("extensions", {})

344 if scope_type == "websocket":

345 send = self._wrap_websocket_denial_send(send)

346

347 if self.stat_result is None:

348 try:

349 stat_result = await anyio.to_thread.run_sync(os.stat, self.path)

350 self.set_stat_headers(stat_result)

351 except FileNotFoundError:

352 raise RuntimeError(f"File at path {self.path} does not exist.")

353 else:

354 mode = stat_result.st_mode

355 if not stat.S_ISREG(mode):

356 raise RuntimeError(f"File at path {self.path} is not a file.")

357 else:

358 stat_result = self.stat_result

359

360 headers = Headers(scope=scope)

361 http_range = headers.get("range")

362 http_if_range = headers.get("if-range")

363

364 if http_range is None or (http_if_range is not None and not self._should_use_range(http_if_range)):

365 await self._handle_simple(send, send_header_only, send_pathsend)

366 else:

367 try:

368 ranges = self._parse_range_header(http_range, stat_result.st_size)

369 except MalformedRangeHeader as exc:

370 return await PlainTextResponse(exc.content, status_code=400)(scope, receive, send)

371 except RangeNotSatisfiable as exc:

372 response = PlainTextResponse(status_code=416, headers={"Content-Range": f"bytes */{exc.max_size}"})

373 return await response(scope, receive, send)

374

375 if len(ranges) == 1:

376 start, end = ranges[0]

377 await self._handle_single_range(send, start, end, stat_result.st_size, send_header_only)

378 else:

379 await self._handle_multiple_ranges(send, ranges, stat_result.st_size, send_header_only)

380

381 if self.background is not None:

382 await self.background()

383

384 async def _handle_simple(self, send: Send, send_header_only: bool, send_pathsend: bool) -> None:

385 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})

386 if send_header_only:

387 await send({"type": "http.response.body", "body": b"", "more_body": False})

388 elif send_pathsend:

389 await send({"type": "http.response.pathsend", "path": str(self.path)})

390 else:

391 async with await anyio.open_file(self.path, mode="rb") as file:

392 more_body = True

393 while more_body:

394 chunk = await file.read(self.chunk_size)

395 more_body = len(chunk) == self.chunk_size

396 await send({"type": "http.response.body", "body": chunk, "more_body": more_body})

397

398 async def _handle_single_range(

399 self, send: Send, start: int, end: int, file_size: int, send_header_only: bool

400 ) -> None:

401 headers = MutableHeaders(raw=list(self.raw_headers))

402 headers["content-range"] = f"bytes {start}-{end - 1}/{file_size}"

403 headers["content-length"] = str(end - start)

404 await send({"type": "http.response.start", "status": 206, "headers": headers.raw})

405 if send_header_only:

406 await send({"type": "http.response.body", "body": b"", "more_body": False})

407 else:

408 async with await anyio.open_file(self.path, mode="rb") as file:

409 await file.seek(start)

410 more_body = True

411 while more_body:

412 chunk = await file.read(min(self.chunk_size, end - start))

413 start += len(chunk)

414 more_body = len(chunk) == self.chunk_size and start < end

415 await send({"type": "http.response.body", "body": chunk, "more_body": more_body})

416

417 async def _handle_multiple_ranges(

418 self,

419 send: Send,

420 ranges: list[tuple[int, int]],

421 file_size: int,

422 send_header_only: bool,

423 ) -> None:

424 # In firefox and chrome, they use boundary with 95-96 bits entropy (that's roughly 13 bytes).

425 boundary = token_hex(13)

426 content_length, header_generator = self.generate_multipart(

427 ranges, boundary, file_size, self.headers["content-type"]

428 )

429 headers = MutableHeaders(raw=list(self.raw_headers))

430 headers["content-type"] = f"multipart/byteranges; boundary={boundary}"

431 headers["content-length"] = str(content_length)

432 await send({"type": "http.response.start", "status": 206, "headers": headers.raw})

433 if send_header_only:

434 await send({"type": "http.response.body", "body": b"", "more_body": False})

435 else:

436 async with await anyio.open_file(self.path, mode="rb") as file:

437 for start, end in ranges:

438 await send({"type": "http.response.body", "body": header_generator(start, end), "more_body": True})

439 await file.seek(start)

440 while start < end:

441 chunk = await file.read(min(self.chunk_size, end - start))

442 start += len(chunk)

443 await send({"type": "http.response.body", "body": chunk, "more_body": True})

444 await send({"type": "http.response.body", "body": b"\r\n", "more_body": True})

445 await send(

446 {

447 "type": "http.response.body",

448 "body": f"--{boundary}--".encode("latin-1"),

449 "more_body": False,

450 }

451 )

452

453 def _should_use_range(self, http_if_range: str) -> bool:

454 return http_if_range == self.headers["last-modified"] or http_if_range == self.headers["etag"]

455

456 @classmethod

457 def _parse_range_header(cls, http_range: str, file_size: int) -> list[tuple[int, int]]:

458 ranges: list[tuple[int, int]] = []

459 try:

460 units, range_ = http_range.split("=", 1)

461 except ValueError:

462 raise MalformedRangeHeader()

463

464 units = units.strip().lower()

465

466 if units != "bytes":

467 raise MalformedRangeHeader("Only support bytes range")

468

469 ranges = cls._parse_ranges(range_, file_size)

470

471 if len(ranges) == 0:

472 raise MalformedRangeHeader("Range header: range must be requested")

473

474 if any(not (0 <= start < file_size) for start, _ in ranges):

475 raise RangeNotSatisfiable(file_size)

476

477 if any(start > end for start, end in ranges):

478 raise MalformedRangeHeader("Range header: start must be less than end")

479

480 if len(ranges) == 1:

481 return ranges

482

483 # Merge overlapping ranges

484 ranges.sort()

485 result: list[tuple[int, int]] = [ranges[0]]

486 for start, end in ranges[1:]:

487 last_start, last_end = result[-1]

488 if start <= last_end:

489 result[-1] = (last_start, max(last_end, end))

490 else:

491 result.append((start, end))

492

493 return result

494

495 @classmethod

496 def _parse_ranges(cls, range_: str, file_size: int) -> list[tuple[int, int]]:

497 ranges: list[tuple[int, int]] = []

498

499 for part in range_.split(","):

500 part = part.strip()

501

502 # If the range is empty or a single dash, we ignore it.

503 if not part or part == "-":

504 continue

505

506 # If the range is not in the format "start-end", we ignore it.

507 if "-" not in part:

508 continue

509

510 start_str, end_str = part.split("-", 1)

511 start_str = start_str.strip()

512 end_str = end_str.strip()

513

514 try:

515 start = int(start_str) if start_str else max(file_size - int(end_str), 0)

516 end = int(end_str) + 1 if start_str and end_str and int(end_str) < file_size else file_size

517 ranges.append((start, end))

518 except ValueError:

519 # If the range is not numeric, we ignore it.

520 continue

521

522 return ranges

523

524 def generate_multipart(

525 self,

526 ranges: Sequence[tuple[int, int]],

527 boundary: str,

528 max_size: int,

529 content_type: str,

530 ) -> tuple[int, Callable[[int, int], bytes]]:

531 r"""

532 Multipart response headers generator.

533

534 ```

535 --{boundary}\r\n

536 Content-Type: {content_type}\r\n

537 Content-Range: bytes {start}-{end-1}/{max_size}\r\n

538 \r\n

539 ..........content...........\r\n

540 --{boundary}\r\n

541 Content-Type: {content_type}\r\n

542 Content-Range: bytes {start}-{end-1}/{max_size}\r\n

543 \r\n

544 ..........content...........\r\n

545 --{boundary}--

546 ```

547 """

548 boundary_len = len(boundary)

549 static_header_part_len = 49 + boundary_len + len(content_type) + len(str(max_size))

550 content_length = sum(

551 (len(str(start)) + len(str(end - 1)) + static_header_part_len) # Headers

552 + (end - start) # Content

553 for start, end in ranges

554 ) + (

555 4 + boundary_len # --boundary--

556 )

557 return (

558 content_length,

559 lambda start, end: (

560 f"""\

561--{boundary}\r

562Content-Type: {content_type}\r

563Content-Range: bytes {start}-{end - 1}/{max_size}\r

564\r

565"""

566 ).encode("latin-1"),

567 )