File size: 13,564 Bytes
add8f0b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 |
import enum
import logging
import ssl
import time
from types import TracebackType
from typing import (
Any,
Iterable,
Iterator,
List,
Optional,
Tuple,
Type,
Union,
)
import h11
from .._backends.base import NetworkStream
from .._exceptions import (
ConnectionNotAvailable,
LocalProtocolError,
RemoteProtocolError,
WriteError,
map_exceptions,
)
from .._models import Origin, Request, Response
from .._synchronization import Lock, ShieldCancellation
from .._trace import Trace
from .interfaces import ConnectionInterface
logger = logging.getLogger("httpcore.http11")
# A subset of `h11.Event` types supported by `_send_event`
H11SendEvent = Union[
h11.Request,
h11.Data,
h11.EndOfMessage,
]
class HTTPConnectionState(enum.IntEnum):
NEW = 0
ACTIVE = 1
IDLE = 2
CLOSED = 3
class HTTP11Connection(ConnectionInterface):
READ_NUM_BYTES = 64 * 1024
MAX_INCOMPLETE_EVENT_SIZE = 100 * 1024
def __init__(
self,
origin: Origin,
stream: NetworkStream,
keepalive_expiry: Optional[float] = None,
) -> None:
self._origin = origin
self._network_stream = stream
self._keepalive_expiry: Optional[float] = keepalive_expiry
self._expire_at: Optional[float] = None
self._state = HTTPConnectionState.NEW
self._state_lock = Lock()
self._request_count = 0
self._h11_state = h11.Connection(
our_role=h11.CLIENT,
max_incomplete_event_size=self.MAX_INCOMPLETE_EVENT_SIZE,
)
def handle_request(self, request: Request) -> Response:
if not self.can_handle_request(request.url.origin):
raise RuntimeError(
f"Attempted to send request to {request.url.origin} on connection "
f"to {self._origin}"
)
with self._state_lock:
if self._state in (HTTPConnectionState.NEW, HTTPConnectionState.IDLE):
self._request_count += 1
self._state = HTTPConnectionState.ACTIVE
self._expire_at = None
else:
raise ConnectionNotAvailable()
try:
kwargs = {"request": request}
try:
with Trace(
"send_request_headers", logger, request, kwargs
) as trace:
self._send_request_headers(**kwargs)
with Trace("send_request_body", logger, request, kwargs) as trace:
self._send_request_body(**kwargs)
except WriteError:
# If we get a write error while we're writing the request,
# then we supress this error and move on to attempting to
# read the response. Servers can sometimes close the request
# pre-emptively and then respond with a well formed HTTP
# error response.
pass
with Trace(
"receive_response_headers", logger, request, kwargs
) as trace:
(
http_version,
status,
reason_phrase,
headers,
trailing_data,
) = self._receive_response_headers(**kwargs)
trace.return_value = (
http_version,
status,
reason_phrase,
headers,
)
network_stream = self._network_stream
# CONNECT or Upgrade request
if (status == 101) or (
(request.method == b"CONNECT") and (200 <= status < 300)
):
network_stream = HTTP11UpgradeStream(network_stream, trailing_data)
return Response(
status=status,
headers=headers,
content=HTTP11ConnectionByteStream(self, request),
extensions={
"http_version": http_version,
"reason_phrase": reason_phrase,
"network_stream": network_stream,
},
)
except BaseException as exc:
with ShieldCancellation():
with Trace("response_closed", logger, request) as trace:
self._response_closed()
raise exc
# Sending the request...
def _send_request_headers(self, request: Request) -> None:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("write", None)
with map_exceptions({h11.LocalProtocolError: LocalProtocolError}):
event = h11.Request(
method=request.method,
target=request.url.target,
headers=request.headers,
)
self._send_event(event, timeout=timeout)
def _send_request_body(self, request: Request) -> None:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("write", None)
assert isinstance(request.stream, Iterable)
for chunk in request.stream:
event = h11.Data(data=chunk)
self._send_event(event, timeout=timeout)
self._send_event(h11.EndOfMessage(), timeout=timeout)
def _send_event(
self, event: h11.Event, timeout: Optional[float] = None
) -> None:
bytes_to_send = self._h11_state.send(event)
if bytes_to_send is not None:
self._network_stream.write(bytes_to_send, timeout=timeout)
# Receiving the response...
def _receive_response_headers(
self, request: Request
) -> Tuple[bytes, int, bytes, List[Tuple[bytes, bytes]], bytes]:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("read", None)
while True:
event = self._receive_event(timeout=timeout)
if isinstance(event, h11.Response):
break
if (
isinstance(event, h11.InformationalResponse)
and event.status_code == 101
):
break
http_version = b"HTTP/" + event.http_version
# h11 version 0.11+ supports a `raw_items` interface to get the
# raw header casing, rather than the enforced lowercase headers.
headers = event.headers.raw_items()
trailing_data, _ = self._h11_state.trailing_data
return http_version, event.status_code, event.reason, headers, trailing_data
def _receive_response_body(self, request: Request) -> Iterator[bytes]:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("read", None)
while True:
event = self._receive_event(timeout=timeout)
if isinstance(event, h11.Data):
yield bytes(event.data)
elif isinstance(event, (h11.EndOfMessage, h11.PAUSED)):
break
def _receive_event(
self, timeout: Optional[float] = None
) -> Union[h11.Event, Type[h11.PAUSED]]:
while True:
with map_exceptions({h11.RemoteProtocolError: RemoteProtocolError}):
event = self._h11_state.next_event()
if event is h11.NEED_DATA:
data = self._network_stream.read(
self.READ_NUM_BYTES, timeout=timeout
)
# If we feed this case through h11 we'll raise an exception like:
#
# httpcore.RemoteProtocolError: can't handle event type
# ConnectionClosed when role=SERVER and state=SEND_RESPONSE
#
# Which is accurate, but not very informative from an end-user
# perspective. Instead we handle this case distinctly and treat
# it as a ConnectError.
if data == b"" and self._h11_state.their_state == h11.SEND_RESPONSE:
msg = "Server disconnected without sending a response."
raise RemoteProtocolError(msg)
self._h11_state.receive_data(data)
else:
# mypy fails to narrow the type in the above if statement above
return event # type: ignore[return-value]
def _response_closed(self) -> None:
with self._state_lock:
if (
self._h11_state.our_state is h11.DONE
and self._h11_state.their_state is h11.DONE
):
self._state = HTTPConnectionState.IDLE
self._h11_state.start_next_cycle()
if self._keepalive_expiry is not None:
now = time.monotonic()
self._expire_at = now + self._keepalive_expiry
else:
self.close()
# Once the connection is no longer required...
def close(self) -> None:
# Note that this method unilaterally closes the connection, and does
# not have any kind of locking in place around it.
self._state = HTTPConnectionState.CLOSED
self._network_stream.close()
# The ConnectionInterface methods provide information about the state of
# the connection, allowing for a connection pooling implementation to
# determine when to reuse and when to close the connection...
def can_handle_request(self, origin: Origin) -> bool:
return origin == self._origin
def is_available(self) -> bool:
# Note that HTTP/1.1 connections in the "NEW" state are not treated as
# being "available". The control flow which created the connection will
# be able to send an outgoing request, but the connection will not be
# acquired from the connection pool for any other request.
return self._state == HTTPConnectionState.IDLE
def has_expired(self) -> bool:
now = time.monotonic()
keepalive_expired = self._expire_at is not None and now > self._expire_at
# If the HTTP connection is idle but the socket is readable, then the
# only valid state is that the socket is about to return b"", indicating
# a server-initiated disconnect.
server_disconnected = (
self._state == HTTPConnectionState.IDLE
and self._network_stream.get_extra_info("is_readable")
)
return keepalive_expired or server_disconnected
def is_idle(self) -> bool:
return self._state == HTTPConnectionState.IDLE
def is_closed(self) -> bool:
return self._state == HTTPConnectionState.CLOSED
def info(self) -> str:
origin = str(self._origin)
return (
f"{origin!r}, HTTP/1.1, {self._state.name}, "
f"Request Count: {self._request_count}"
)
def __repr__(self) -> str:
class_name = self.__class__.__name__
origin = str(self._origin)
return (
f"<{class_name} [{origin!r}, {self._state.name}, "
f"Request Count: {self._request_count}]>"
)
# These context managers are not used in the standard flow, but are
# useful for testing or working with connection instances directly.
def __enter__(self) -> "HTTP11Connection":
return self
def __exit__(
self,
exc_type: Optional[Type[BaseException]] = None,
exc_value: Optional[BaseException] = None,
traceback: Optional[TracebackType] = None,
) -> None:
self.close()
class HTTP11ConnectionByteStream:
def __init__(self, connection: HTTP11Connection, request: Request) -> None:
self._connection = connection
self._request = request
self._closed = False
def __iter__(self) -> Iterator[bytes]:
kwargs = {"request": self._request}
try:
with Trace("receive_response_body", logger, self._request, kwargs):
for chunk in self._connection._receive_response_body(**kwargs):
yield chunk
except BaseException as exc:
# If we get an exception while streaming the response,
# we want to close the response (and possibly the connection)
# before raising that exception.
with ShieldCancellation():
self.close()
raise exc
def close(self) -> None:
if not self._closed:
self._closed = True
with Trace("response_closed", logger, self._request):
self._connection._response_closed()
class HTTP11UpgradeStream(NetworkStream):
def __init__(self, stream: NetworkStream, leading_data: bytes) -> None:
self._stream = stream
self._leading_data = leading_data
def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
if self._leading_data:
buffer = self._leading_data[:max_bytes]
self._leading_data = self._leading_data[max_bytes:]
return buffer
else:
return self._stream.read(max_bytes, timeout)
def write(self, buffer: bytes, timeout: Optional[float] = None) -> None:
self._stream.write(buffer, timeout)
def close(self) -> None:
self._stream.close()
def start_tls(
self,
ssl_context: ssl.SSLContext,
server_hostname: Optional[str] = None,
timeout: Optional[float] = None,
) -> NetworkStream:
return self._stream.start_tls(ssl_context, server_hostname, timeout)
def get_extra_info(self, info: str) -> Any:
return self._stream.get_extra_info(info)
|