(extstr: str | None, isserver: bool = False)
| 76 | |
| 77 | |
| 78 | def ws_ext_parse(extstr: str | None, isserver: bool = False) -> tuple[int, bool]: |
| 79 | if not extstr: |
| 80 | return 0, False |
| 81 | |
| 82 | compress = 0 |
| 83 | notakeover = False |
| 84 | for ext in _WS_EXT_RE_SPLIT.finditer(extstr): |
| 85 | defext = ext.group(1) |
| 86 | # Return compress = 15 when get `permessage-deflate` |
| 87 | if not defext: |
| 88 | compress = 15 |
| 89 | break |
| 90 | match = _WS_EXT_RE.match(defext) |
| 91 | if match: |
| 92 | compress = 15 |
| 93 | if isserver: |
| 94 | # Server never fail to detect compress handshake. |
| 95 | # Server does not need to send max wbit to client |
| 96 | if match.group(4): |
| 97 | compress = int(match.group(4)) |
| 98 | # Group3 must match if group4 matches |
| 99 | # Compress wbit 8 does not support in zlib |
| 100 | # If compress level not support, |
| 101 | # CONTINUE to next extension |
| 102 | if compress > 15 or compress < 9: |
| 103 | compress = 0 |
| 104 | continue |
| 105 | if match.group(1): |
| 106 | notakeover = True |
| 107 | # Ignore regex group 5 & 6 for client_max_window_bits |
| 108 | break |
| 109 | else: |
| 110 | if match.group(6): |
| 111 | compress = int(match.group(6)) |
| 112 | # Group5 must match if group6 matches |
| 113 | # Compress wbit 8 does not support in zlib |
| 114 | # If compress level not support, |
| 115 | # FAIL the parse progress |
| 116 | if compress > 15 or compress < 9: |
| 117 | raise WSHandshakeError("Invalid window size") |
| 118 | if match.group(2): |
| 119 | notakeover = True |
| 120 | # Ignore regex group 5 & 6 for client_max_window_bits |
| 121 | break |
| 122 | # Return Fail if client side and not match |
| 123 | elif not isserver: |
| 124 | raise WSHandshakeError("Extension for deflate not supported" + ext.group(1)) |
| 125 | |
| 126 | return compress, notakeover |
| 127 | |
| 128 | |
| 129 | def ws_ext_gen( |
no test coverage detected
searching dependent graphs…