Handle the Origin HTTP request header. Args: headers: WebSocket handshake request headers. Returns: origin, if it is acceptable. Raises: InvalidHandshake: If the Origin header is invalid. InvalidOrigin: If the origin
(self, headers: Headers)
| 275 | return (accept_header, extensions_header, protocol_header) |
| 276 | |
| 277 | def process_origin(self, headers: Headers) -> Origin | None: |
| 278 | """ |
| 279 | Handle the Origin HTTP request header. |
| 280 | |
| 281 | Args: |
| 282 | headers: WebSocket handshake request headers. |
| 283 | |
| 284 | Returns: |
| 285 | origin, if it is acceptable. |
| 286 | |
| 287 | Raises: |
| 288 | InvalidHandshake: If the Origin header is invalid. |
| 289 | InvalidOrigin: If the origin isn't acceptable. |
| 290 | |
| 291 | """ |
| 292 | # "The user agent MUST NOT include more than one Origin header field" |
| 293 | # per https://datatracker.ietf.org/doc/html/rfc6454#section-7.3. |
| 294 | try: |
| 295 | origin = headers.get("Origin") |
| 296 | except MultipleValuesError: |
| 297 | raise InvalidHeader("Origin", "multiple values") from None |
| 298 | if origin is not None: |
| 299 | origin = cast(Origin, origin) |
| 300 | if self.origins is not None: |
| 301 | for origin_or_regex in self.origins: |
| 302 | if origin_or_regex == origin or ( |
| 303 | isinstance(origin_or_regex, re.Pattern) |
| 304 | and origin is not None |
| 305 | and origin_or_regex.fullmatch(origin) is not None |
| 306 | ): |
| 307 | break |
| 308 | else: |
| 309 | raise InvalidOrigin(origin) |
| 310 | return origin |
| 311 | |
| 312 | def process_extensions( |
| 313 | self, |
no test coverage detected