(self, raw: str, **kwargs)
| 145 | return _send_output_hook |
| 146 | |
| 147 | def httpraw(self, raw: str, **kwargs): |
| 148 | raw = raw.strip() |
| 149 | proxy = kwargs.get("proxy", None) |
| 150 | real_host = kwargs.get("real_host", None) |
| 151 | ssl = kwargs.get("ssl", False) |
| 152 | |
| 153 | scheme = 'http' |
| 154 | port = 80 |
| 155 | if ssl: |
| 156 | scheme = 'https' |
| 157 | port = 443 |
| 158 | |
| 159 | try: |
| 160 | index = raw.index('\n') |
| 161 | except ValueError: |
| 162 | raise Exception("ValueError") |
| 163 | log = {} |
| 164 | try: |
| 165 | method, path, protocol = raw[:index].split(" ") |
| 166 | except: |
| 167 | raise Exception("Protocol format error") |
| 168 | raw = raw[index + 1:] |
| 169 | |
| 170 | try: |
| 171 | host_start = raw.index("Host: ") |
| 172 | host_end = raw.index('\n', host_start) |
| 173 | |
| 174 | except ValueError: |
| 175 | raise ValueError("Host headers not found") |
| 176 | |
| 177 | if real_host: |
| 178 | host = real_host |
| 179 | if ":" in real_host: |
| 180 | host, port = real_host.split(":") |
| 181 | else: |
| 182 | host = raw[host_start + len("Host: "):host_end] |
| 183 | if ":" in host: |
| 184 | host, port = host.split(":") |
| 185 | raws = raw.splitlines() |
| 186 | headers = {} |
| 187 | |
| 188 | # index = 0 |
| 189 | # for r in raws: |
| 190 | # raws[index] = r.lstrip() |
| 191 | # index += 1 |
| 192 | |
| 193 | index = 0 |
| 194 | for r in raws: |
| 195 | if r == "": |
| 196 | break |
| 197 | try: |
| 198 | k, v = r.split(": ") |
| 199 | except: |
| 200 | k = r |
| 201 | v = "" |
| 202 | headers[k] = v |
| 203 | index += 1 |
| 204 | headers["Connection"] = "close" |
nothing calls this directly
no test coverage detected