(
self,
scope: _ASGIScope,
receive: _ASGIReceive,
send: _ASGISend,
)
| 543 | self.app = app |
| 544 | |
| 545 | async def __call__( |
| 546 | self, |
| 547 | scope: _ASGIScope, |
| 548 | receive: _ASGIReceive, |
| 549 | send: _ASGISend, |
| 550 | ) -> None: |
| 551 | scope_type = scope.get("type") |
| 552 | if scope_type not in ("http", "websocket"): |
| 553 | # Non-HTTP/WebSocket traffic is forwarded verbatim |
| 554 | await self.app(scope, receive, send) |
| 555 | return |
| 556 | |
| 557 | if scope_type == "http" and scope.get("path") == "/_vercel/ping": |
| 558 | await send( |
| 559 | { |
| 560 | "type": "http.response.start", |
| 561 | "status": 200, |
| 562 | "headers": [], |
| 563 | } |
| 564 | ) |
| 565 | await send( |
| 566 | { |
| 567 | "type": "http.response.body", |
| 568 | "body": b"", |
| 569 | "more_body": False, |
| 570 | } |
| 571 | ) |
| 572 | return |
| 573 | |
| 574 | # Extract internal headers and set per-request context |
| 575 | headers_list: list[tuple[bytes | str, bytes | str]] = ( |
| 576 | scope.get("headers", []) or [] |
| 577 | ) |
| 578 | new_headers: list[tuple[bytes, bytes]] = [] |
| 579 | invocation_id = "0" |
| 580 | request_id = 0 |
| 581 | internal_oidc_token: str | None = None |
| 582 | sc_pairs: list[tuple[bytes, bytes]] = [] |
| 583 | sc_no_header_leak = False |
| 584 | |
| 585 | for raw_k, raw_v in headers_list: |
| 586 | key_bytes = raw_k if isinstance(raw_k, bytes) else raw_k.encode() |
| 587 | val_bytes = raw_v if isinstance(raw_v, bytes) else raw_v.encode() |
| 588 | key = decode_header_bytes(key_bytes).lower() |
| 589 | val = decode_header_bytes(val_bytes) |
| 590 | if key == "x-vercel-internal-invocation-id": |
| 591 | invocation_id = val |
| 592 | continue |
| 593 | if key == "x-vercel-internal-request-id": |
| 594 | request_id = int(val) if val.isdigit() else 0 |
| 595 | continue |
| 596 | if key in ( |
| 597 | "x-vercel-internal-span-id", |
| 598 | "x-vercel-internal-trace-id", |
| 599 | ): |
| 600 | continue |
| 601 | if key == INTERNAL_OIDC_HEADER_NAME: |
| 602 | internal_oidc_token = val |
nothing calls this directly
no test coverage detected