connection.py 85 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110
  1. """
  2. h2/connection
  3. ~~~~~~~~~~~~~
  4. An implementation of a HTTP/2 connection.
  5. """
  6. from __future__ import annotations
  7. import base64
  8. from enum import Enum, IntEnum
  9. from typing import TYPE_CHECKING, Any, Callable
  10. from hpack.exceptions import HPACKError, OversizedHeaderListError
  11. from hpack.hpack import Decoder, Encoder
  12. from hyperframe.exceptions import InvalidPaddingError
  13. from hyperframe.frame import (
  14. AltSvcFrame,
  15. ContinuationFrame,
  16. DataFrame,
  17. ExtensionFrame,
  18. Frame,
  19. GoAwayFrame,
  20. HeadersFrame,
  21. PingFrame,
  22. PriorityFrame,
  23. PushPromiseFrame,
  24. RstStreamFrame,
  25. SettingsFrame,
  26. WindowUpdateFrame,
  27. )
  28. from .config import H2Configuration
  29. from .errors import ErrorCodes, _error_code_from_int
  30. from .events import (
  31. AlternativeServiceAvailable,
  32. ConnectionTerminated,
  33. Event,
  34. InformationalResponseReceived,
  35. PingAckReceived,
  36. PingReceived,
  37. PriorityUpdated,
  38. RemoteSettingsChanged,
  39. RequestReceived,
  40. ResponseReceived,
  41. SettingsAcknowledged,
  42. TrailersReceived,
  43. UnknownFrameReceived,
  44. WindowUpdated,
  45. )
  46. from .exceptions import (
  47. DenialOfServiceError,
  48. FlowControlError,
  49. FrameTooLargeError,
  50. NoAvailableStreamIDError,
  51. NoSuchStreamError,
  52. ProtocolError,
  53. RFC1122Error,
  54. StreamClosedError,
  55. StreamIDTooLowError,
  56. TooManyStreamsError,
  57. )
  58. from .frame_buffer import FrameBuffer
  59. from .settings import ChangedSetting, SettingCodes, Settings
  60. from .stream import H2Stream, StreamClosedBy
  61. from .utilities import SizeLimitDict, guard_increment_window
  62. from .windows import WindowManager
  63. if TYPE_CHECKING: # pragma: no cover
  64. from collections.abc import Iterable
  65. from hpack.struct import Header, HeaderWeaklyTyped
  66. class ConnectionState(Enum):
  67. IDLE = 0
  68. CLIENT_OPEN = 1
  69. SERVER_OPEN = 2
  70. CLOSED = 3
  71. class ConnectionInputs(Enum):
  72. SEND_HEADERS = 0
  73. SEND_PUSH_PROMISE = 1
  74. SEND_DATA = 2
  75. SEND_GOAWAY = 3
  76. SEND_WINDOW_UPDATE = 4
  77. SEND_PING = 5
  78. SEND_SETTINGS = 6
  79. SEND_RST_STREAM = 7
  80. SEND_PRIORITY = 8
  81. RECV_HEADERS = 9
  82. RECV_PUSH_PROMISE = 10
  83. RECV_DATA = 11
  84. RECV_GOAWAY = 12
  85. RECV_WINDOW_UPDATE = 13
  86. RECV_PING = 14
  87. RECV_SETTINGS = 15
  88. RECV_RST_STREAM = 16
  89. RECV_PRIORITY = 17
  90. SEND_ALTERNATIVE_SERVICE = 18 # Added in 2.3.0
  91. RECV_ALTERNATIVE_SERVICE = 19 # Added in 2.3.0
  92. class AllowedStreamIDs(IntEnum):
  93. EVEN = 0
  94. ODD = 1
  95. class H2ConnectionStateMachine:
  96. """
  97. A single HTTP/2 connection state machine.
  98. This state machine, while defined in its own class, is logically part of
  99. the H2Connection class also defined in this file. The state machine itself
  100. maintains very little state directly, instead focusing entirely on managing
  101. state transitions.
  102. """
  103. # For the purposes of this state machine we treat HEADERS and their
  104. # associated CONTINUATION frames as a single jumbo frame. The protocol
  105. # allows/requires this by preventing other frames from being interleved in
  106. # between HEADERS/CONTINUATION frames.
  107. #
  108. # The _transitions dictionary contains a mapping of tuples of
  109. # (state, input) to tuples of (side_effect_function, end_state). This map
  110. # contains all allowed transitions: anything not in this map is invalid
  111. # and immediately causes a transition to ``closed``.
  112. _transitions = {
  113. # State: idle
  114. (ConnectionState.IDLE, ConnectionInputs.SEND_HEADERS):
  115. (None, ConnectionState.CLIENT_OPEN),
  116. (ConnectionState.IDLE, ConnectionInputs.RECV_HEADERS):
  117. (None, ConnectionState.SERVER_OPEN),
  118. (ConnectionState.IDLE, ConnectionInputs.SEND_SETTINGS):
  119. (None, ConnectionState.IDLE),
  120. (ConnectionState.IDLE, ConnectionInputs.RECV_SETTINGS):
  121. (None, ConnectionState.IDLE),
  122. (ConnectionState.IDLE, ConnectionInputs.SEND_WINDOW_UPDATE):
  123. (None, ConnectionState.IDLE),
  124. (ConnectionState.IDLE, ConnectionInputs.RECV_WINDOW_UPDATE):
  125. (None, ConnectionState.IDLE),
  126. (ConnectionState.IDLE, ConnectionInputs.SEND_PING):
  127. (None, ConnectionState.IDLE),
  128. (ConnectionState.IDLE, ConnectionInputs.RECV_PING):
  129. (None, ConnectionState.IDLE),
  130. (ConnectionState.IDLE, ConnectionInputs.SEND_GOAWAY):
  131. (None, ConnectionState.CLOSED),
  132. (ConnectionState.IDLE, ConnectionInputs.RECV_GOAWAY):
  133. (None, ConnectionState.CLOSED),
  134. (ConnectionState.IDLE, ConnectionInputs.SEND_PRIORITY):
  135. (None, ConnectionState.IDLE),
  136. (ConnectionState.IDLE, ConnectionInputs.RECV_PRIORITY):
  137. (None, ConnectionState.IDLE),
  138. (ConnectionState.IDLE, ConnectionInputs.SEND_ALTERNATIVE_SERVICE):
  139. (None, ConnectionState.SERVER_OPEN),
  140. (ConnectionState.IDLE, ConnectionInputs.RECV_ALTERNATIVE_SERVICE):
  141. (None, ConnectionState.CLIENT_OPEN),
  142. # State: open, client side.
  143. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_HEADERS):
  144. (None, ConnectionState.CLIENT_OPEN),
  145. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_DATA):
  146. (None, ConnectionState.CLIENT_OPEN),
  147. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_GOAWAY):
  148. (None, ConnectionState.CLOSED),
  149. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_WINDOW_UPDATE):
  150. (None, ConnectionState.CLIENT_OPEN),
  151. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_PING):
  152. (None, ConnectionState.CLIENT_OPEN),
  153. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_SETTINGS):
  154. (None, ConnectionState.CLIENT_OPEN),
  155. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_PRIORITY):
  156. (None, ConnectionState.CLIENT_OPEN),
  157. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_HEADERS):
  158. (None, ConnectionState.CLIENT_OPEN),
  159. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_PUSH_PROMISE):
  160. (None, ConnectionState.CLIENT_OPEN),
  161. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_DATA):
  162. (None, ConnectionState.CLIENT_OPEN),
  163. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_GOAWAY):
  164. (None, ConnectionState.CLOSED),
  165. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_WINDOW_UPDATE):
  166. (None, ConnectionState.CLIENT_OPEN),
  167. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_PING):
  168. (None, ConnectionState.CLIENT_OPEN),
  169. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_SETTINGS):
  170. (None, ConnectionState.CLIENT_OPEN),
  171. (ConnectionState.CLIENT_OPEN, ConnectionInputs.SEND_RST_STREAM):
  172. (None, ConnectionState.CLIENT_OPEN),
  173. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_RST_STREAM):
  174. (None, ConnectionState.CLIENT_OPEN),
  175. (ConnectionState.CLIENT_OPEN, ConnectionInputs.RECV_PRIORITY):
  176. (None, ConnectionState.CLIENT_OPEN),
  177. (ConnectionState.CLIENT_OPEN,
  178. ConnectionInputs.RECV_ALTERNATIVE_SERVICE):
  179. (None, ConnectionState.CLIENT_OPEN),
  180. # State: open, server side.
  181. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_HEADERS):
  182. (None, ConnectionState.SERVER_OPEN),
  183. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_PUSH_PROMISE):
  184. (None, ConnectionState.SERVER_OPEN),
  185. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_DATA):
  186. (None, ConnectionState.SERVER_OPEN),
  187. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_GOAWAY):
  188. (None, ConnectionState.CLOSED),
  189. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_WINDOW_UPDATE):
  190. (None, ConnectionState.SERVER_OPEN),
  191. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_PING):
  192. (None, ConnectionState.SERVER_OPEN),
  193. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_SETTINGS):
  194. (None, ConnectionState.SERVER_OPEN),
  195. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_PRIORITY):
  196. (None, ConnectionState.SERVER_OPEN),
  197. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_HEADERS):
  198. (None, ConnectionState.SERVER_OPEN),
  199. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_DATA):
  200. (None, ConnectionState.SERVER_OPEN),
  201. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_GOAWAY):
  202. (None, ConnectionState.CLOSED),
  203. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_WINDOW_UPDATE):
  204. (None, ConnectionState.SERVER_OPEN),
  205. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_PING):
  206. (None, ConnectionState.SERVER_OPEN),
  207. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_SETTINGS):
  208. (None, ConnectionState.SERVER_OPEN),
  209. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_PRIORITY):
  210. (None, ConnectionState.SERVER_OPEN),
  211. (ConnectionState.SERVER_OPEN, ConnectionInputs.SEND_RST_STREAM):
  212. (None, ConnectionState.SERVER_OPEN),
  213. (ConnectionState.SERVER_OPEN, ConnectionInputs.RECV_RST_STREAM):
  214. (None, ConnectionState.SERVER_OPEN),
  215. (ConnectionState.SERVER_OPEN,
  216. ConnectionInputs.SEND_ALTERNATIVE_SERVICE):
  217. (None, ConnectionState.SERVER_OPEN),
  218. (ConnectionState.SERVER_OPEN,
  219. ConnectionInputs.RECV_ALTERNATIVE_SERVICE):
  220. (None, ConnectionState.SERVER_OPEN),
  221. # State: closed
  222. (ConnectionState.CLOSED, ConnectionInputs.SEND_GOAWAY):
  223. (None, ConnectionState.CLOSED),
  224. (ConnectionState.CLOSED, ConnectionInputs.RECV_GOAWAY):
  225. (None, ConnectionState.CLOSED),
  226. }
  227. def __init__(self) -> None:
  228. self.state = ConnectionState.IDLE
  229. def process_input(self, input_: ConnectionInputs) -> list[Event]:
  230. """
  231. Process a specific input in the state machine.
  232. """
  233. if not isinstance(input_, ConnectionInputs):
  234. msg = "Input must be an instance of ConnectionInputs"
  235. raise ValueError(msg) # noqa: TRY004
  236. try:
  237. func, target_state = self._transitions[(self.state, input_)]
  238. except KeyError as e:
  239. old_state = self.state
  240. self.state = ConnectionState.CLOSED
  241. msg = f"Invalid input {input_} in state {old_state}"
  242. raise ProtocolError(msg) from e
  243. else:
  244. self.state = target_state
  245. if func is not None: # pragma: no cover
  246. return func()
  247. return []
  248. class H2Connection:
  249. """
  250. A low-level HTTP/2 connection object. This handles building and receiving
  251. frames and maintains both connection and per-stream state for all streams
  252. on this connection.
  253. This wraps a HTTP/2 Connection state machine implementation, ensuring that
  254. frames can only be sent/received when the connection is in a valid state.
  255. It also builds stream state machines on demand to ensure that the
  256. constraints of those state machines are met as well. Attempts to create
  257. frames that cannot be sent will raise a ``ProtocolError``.
  258. .. versionchanged:: 2.3.0
  259. Added the ``header_encoding`` keyword argument.
  260. .. versionchanged:: 2.5.0
  261. Added the ``config`` keyword argument. Deprecated the ``client_side``
  262. and ``header_encoding`` parameters.
  263. .. versionchanged:: 3.0.0
  264. Removed deprecated parameters and properties.
  265. :param config: The configuration for the HTTP/2 connection.
  266. .. versionadded:: 2.5.0
  267. :type config: :class:`H2Configuration <h2.config.H2Configuration>`
  268. """
  269. # The initial maximum outbound frame size. This can be changed by receiving
  270. # a settings frame.
  271. DEFAULT_MAX_OUTBOUND_FRAME_SIZE = 65535
  272. # The initial maximum inbound frame size. This is somewhat arbitrarily
  273. # chosen.
  274. DEFAULT_MAX_INBOUND_FRAME_SIZE = 2**24
  275. # The highest acceptable stream ID.
  276. HIGHEST_ALLOWED_STREAM_ID = 2**31 - 1
  277. # The largest acceptable window increment.
  278. MAX_WINDOW_INCREMENT = 2**31 - 1
  279. # The initial default value of SETTINGS_MAX_HEADER_LIST_SIZE.
  280. DEFAULT_MAX_HEADER_LIST_SIZE = 2**16
  281. # Keep in memory limited amount of results for streams closes
  282. MAX_CLOSED_STREAMS = 2**16
  283. def __init__(self, config: H2Configuration | None = None) -> None:
  284. self.state_machine = H2ConnectionStateMachine()
  285. self.streams: dict[int, H2Stream] = {}
  286. self.highest_inbound_stream_id = 0
  287. self.highest_outbound_stream_id = 0
  288. self.encoder = Encoder()
  289. self.decoder = Decoder()
  290. # This won't always actually do anything: for versions of HPACK older
  291. # than 2.3.0 it does nothing. However, we have to try!
  292. self.decoder.max_header_list_size = self.DEFAULT_MAX_HEADER_LIST_SIZE
  293. #: The configuration for this HTTP/2 connection object.
  294. #:
  295. #: .. versionadded:: 2.5.0
  296. self.config = config or H2Configuration(client_side=True)
  297. # Objects that store settings, including defaults.
  298. #
  299. # We set the MAX_CONCURRENT_STREAMS value to 100 because its default is
  300. # unbounded, and that's a dangerous default because it allows
  301. # essentially unbounded resources to be allocated regardless of how
  302. # they will be used. 100 should be suitable for the average
  303. # application. This default obviously does not apply to the remote
  304. # peer's settings: the remote peer controls them!
  305. #
  306. # We also set MAX_HEADER_LIST_SIZE to a reasonable value. This is to
  307. # advertise our defence against CVE-2016-6581. However, not all
  308. # versions of HPACK will let us do it. That's ok: we should at least
  309. # suggest that we're not vulnerable.
  310. self.local_settings = Settings(
  311. client=self.config.client_side,
  312. initial_values={
  313. SettingCodes.MAX_CONCURRENT_STREAMS: 100,
  314. SettingCodes.MAX_HEADER_LIST_SIZE:
  315. self.DEFAULT_MAX_HEADER_LIST_SIZE,
  316. },
  317. )
  318. self.remote_settings = Settings(client=not self.config.client_side)
  319. # The current value of the connection flow control windows on the
  320. # connection.
  321. self.outbound_flow_control_window = (
  322. self.remote_settings.initial_window_size
  323. )
  324. #: The maximum size of a frame that can be emitted by this peer, in
  325. #: bytes.
  326. self.max_outbound_frame_size = self.remote_settings.max_frame_size
  327. #: The maximum size of a frame that can be received by this peer, in
  328. #: bytes.
  329. self.max_inbound_frame_size = self.local_settings.max_frame_size
  330. # Buffer for incoming data.
  331. self.incoming_buffer = FrameBuffer(server=not self.config.client_side)
  332. # A private variable to store a sequence of received header frames
  333. # until completion.
  334. self._header_frames: list[Frame] = []
  335. # Data that needs to be sent.
  336. self._data_to_send = bytearray()
  337. # Keeps track of how streams are closed.
  338. # Used to ensure that we don't blow up in the face of frames that were
  339. # in flight when a RST_STREAM was sent.
  340. # Also used to determine whether we should consider a frame received
  341. # while a stream is closed as either a stream error or a connection
  342. # error.
  343. self._closed_streams: dict[int, StreamClosedBy | None] = SizeLimitDict(
  344. size_limit=self.MAX_CLOSED_STREAMS,
  345. )
  346. # The flow control window manager for the connection.
  347. self._inbound_flow_control_window_manager = WindowManager(
  348. max_window_size=self.local_settings.initial_window_size,
  349. )
  350. # When in doubt use dict-dispatch.
  351. self._frame_dispatch_table: dict[type[Frame], Callable] = { # type: ignore
  352. HeadersFrame: self._receive_headers_frame,
  353. PushPromiseFrame: self._receive_push_promise_frame,
  354. SettingsFrame: self._receive_settings_frame,
  355. DataFrame: self._receive_data_frame,
  356. WindowUpdateFrame: self._receive_window_update_frame,
  357. PingFrame: self._receive_ping_frame,
  358. RstStreamFrame: self._receive_rst_stream_frame,
  359. PriorityFrame: self._receive_priority_frame,
  360. GoAwayFrame: self._receive_goaway_frame,
  361. ContinuationFrame: self._receive_naked_continuation,
  362. AltSvcFrame: self._receive_alt_svc_frame,
  363. ExtensionFrame: self._receive_unknown_frame,
  364. }
  365. def _prepare_for_sending(self, frames: list[Frame]) -> None:
  366. if not frames:
  367. return
  368. self._data_to_send += b"".join(f.serialize() for f in frames)
  369. assert all(f.body_len <= self.max_outbound_frame_size for f in frames)
  370. def _open_streams(self, remainder: int) -> int:
  371. """
  372. A common method of counting number of open streams. Returns the number
  373. of streams that are open *and* that have (stream ID % 2) == remainder.
  374. While it iterates, also deletes any closed streams.
  375. """
  376. count = 0
  377. to_delete = []
  378. for stream_id, stream in self.streams.items():
  379. if stream.open and (stream_id % 2 == remainder):
  380. count += 1
  381. elif stream.closed:
  382. to_delete.append(stream_id)
  383. for stream_id in to_delete:
  384. stream = self.streams.pop(stream_id)
  385. self._closed_streams[stream_id] = stream.closed_by
  386. return count
  387. @property
  388. def open_outbound_streams(self) -> int:
  389. """
  390. The current number of open outbound streams.
  391. """
  392. outbound_numbers = int(self.config.client_side)
  393. return self._open_streams(outbound_numbers)
  394. @property
  395. def open_inbound_streams(self) -> int:
  396. """
  397. The current number of open inbound streams.
  398. """
  399. inbound_numbers = int(not self.config.client_side)
  400. return self._open_streams(inbound_numbers)
  401. @property
  402. def inbound_flow_control_window(self) -> int:
  403. """
  404. The size of the inbound flow control window for the connection. This is
  405. rarely publicly useful: instead, use :meth:`remote_flow_control_window
  406. <h2.connection.H2Connection.remote_flow_control_window>`. This
  407. shortcut is largely present to provide a shortcut to this data.
  408. """
  409. return self._inbound_flow_control_window_manager.current_window_size
  410. def _begin_new_stream(self, stream_id: int, allowed_ids: AllowedStreamIDs) -> H2Stream:
  411. """
  412. Initiate a new stream.
  413. .. versionchanged:: 2.0.0
  414. Removed this function from the public API.
  415. :param stream_id: The ID of the stream to open.
  416. :param allowed_ids: What kind of stream ID is allowed.
  417. """
  418. self.config.logger.debug(
  419. "Attempting to initiate stream ID %d", stream_id,
  420. )
  421. outbound = self._stream_id_is_outbound(stream_id)
  422. highest_stream_id = (
  423. self.highest_outbound_stream_id if outbound else
  424. self.highest_inbound_stream_id
  425. )
  426. if stream_id <= highest_stream_id:
  427. raise StreamIDTooLowError(stream_id, highest_stream_id)
  428. if (stream_id % 2) != int(allowed_ids):
  429. msg = "Invalid stream ID for peer."
  430. raise ProtocolError(msg)
  431. s = H2Stream(
  432. stream_id,
  433. config=self.config,
  434. inbound_window_size=self.local_settings.initial_window_size,
  435. outbound_window_size=self.remote_settings.initial_window_size,
  436. )
  437. self.config.logger.debug("Stream ID %d created", stream_id)
  438. s.max_outbound_frame_size = self.max_outbound_frame_size
  439. self.streams[stream_id] = s
  440. self.config.logger.debug("Current streams: %s", self.streams.keys())
  441. if outbound:
  442. self.highest_outbound_stream_id = stream_id
  443. else:
  444. self.highest_inbound_stream_id = stream_id
  445. return s
  446. def initiate_connection(self) -> None:
  447. """
  448. Provides any data that needs to be sent at the start of the connection.
  449. Must be called for both clients and servers.
  450. """
  451. self.config.logger.debug("Initializing connection")
  452. self.state_machine.process_input(ConnectionInputs.SEND_SETTINGS)
  453. if self.config.client_side:
  454. preamble = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
  455. else:
  456. preamble = b""
  457. f = SettingsFrame(0)
  458. for setting, value in self.local_settings.items():
  459. f.settings[setting] = value
  460. self.config.logger.debug(
  461. "Send Settings frame: %s", self.local_settings,
  462. )
  463. self._data_to_send += preamble + f.serialize()
  464. def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> bytes | None:
  465. """
  466. Call to initialise the connection object for use with an upgraded
  467. HTTP/2 connection (i.e. a connection negotiated using the
  468. ``Upgrade: h2c`` HTTP header).
  469. This method differs from :meth:`initiate_connection
  470. <h2.connection.H2Connection.initiate_connection>` in several ways.
  471. Firstly, it handles the additional SETTINGS frame that is sent in the
  472. ``HTTP2-Settings`` header field. When called on a client connection,
  473. this method will return a bytestring that the caller can put in the
  474. ``HTTP2-Settings`` field they send on their initial request. When
  475. called on a server connection, the user **must** provide the value they
  476. received from the client in the ``HTTP2-Settings`` header field to the
  477. ``settings_header`` argument, which will be used appropriately.
  478. Additionally, this method sets up stream 1 in a half-closed state
  479. appropriate for this side of the connection, to reflect the fact that
  480. the request is already complete.
  481. Finally, this method also prepares the appropriate preamble to be sent
  482. after the upgrade.
  483. .. versionadded:: 2.3.0
  484. :param settings_header: (optional, server-only): The value of the
  485. ``HTTP2-Settings`` header field received from the client.
  486. :type settings_header: ``bytes``
  487. :returns: For clients, a bytestring to put in the ``HTTP2-Settings``.
  488. For servers, returns nothing.
  489. :rtype: ``bytes`` or ``None``
  490. """
  491. self.config.logger.debug(
  492. "Upgrade connection. Current settings: %s", self.local_settings,
  493. )
  494. frame_data = None
  495. # Begin by getting the preamble in place.
  496. self.initiate_connection()
  497. if self.config.client_side:
  498. f = SettingsFrame(0)
  499. for setting, value in self.local_settings.items():
  500. f.settings[setting] = value
  501. frame_data = f.serialize_body()
  502. frame_data = base64.urlsafe_b64encode(frame_data)
  503. elif settings_header:
  504. # We have a settings header from the client. This needs to be
  505. # applied, but we want to throw away the ACK. We do this by
  506. # inserting the data into a Settings frame and then passing it to
  507. # the state machine, but ignoring the return value.
  508. settings_header = base64.urlsafe_b64decode(settings_header)
  509. f = SettingsFrame(0)
  510. f.parse_body(memoryview(settings_header))
  511. self._receive_settings_frame(f)
  512. # Set up appropriate state. Stream 1 in a half-closed state:
  513. # half-closed(local) for clients, half-closed(remote) for servers.
  514. # Additionally, we need to set up the Connection state machine.
  515. connection_input = (
  516. ConnectionInputs.SEND_HEADERS if self.config.client_side
  517. else ConnectionInputs.RECV_HEADERS
  518. )
  519. self.config.logger.debug("Process input %s", connection_input)
  520. self.state_machine.process_input(connection_input)
  521. # Set up stream 1.
  522. self._begin_new_stream(stream_id=1, allowed_ids=AllowedStreamIDs.ODD)
  523. self.streams[1].upgrade(self.config.client_side)
  524. return frame_data
  525. def _get_or_create_stream(self, stream_id: int, allowed_ids: AllowedStreamIDs) -> H2Stream:
  526. """
  527. Gets a stream by its stream ID. Will create one if one does not already
  528. exist. Use allowed_ids to circumvent the usual stream ID rules for
  529. clients and servers.
  530. .. versionchanged:: 2.0.0
  531. Removed this function from the public API.
  532. """
  533. try:
  534. return self.streams[stream_id]
  535. except KeyError:
  536. return self._begin_new_stream(stream_id, allowed_ids)
  537. def _get_stream_by_id(self, stream_id: int | None) -> H2Stream:
  538. """
  539. Gets a stream by its stream ID. Raises NoSuchStreamError if the stream
  540. ID does not correspond to a known stream and is higher than the current
  541. maximum: raises if it is lower than the current maximum.
  542. .. versionchanged:: 2.0.0
  543. Removed this function from the public API.
  544. """
  545. if not stream_id:
  546. raise NoSuchStreamError(-1) # pragma: no cover
  547. try:
  548. return self.streams[stream_id]
  549. except KeyError as e:
  550. outbound = self._stream_id_is_outbound(stream_id)
  551. highest_stream_id = (
  552. self.highest_outbound_stream_id if outbound else
  553. self.highest_inbound_stream_id
  554. )
  555. if stream_id > highest_stream_id:
  556. raise NoSuchStreamError(stream_id) from e
  557. raise StreamClosedError(stream_id) from e
  558. def get_next_available_stream_id(self) -> int:
  559. """
  560. Returns an integer suitable for use as the stream ID for the next
  561. stream created by this endpoint. For server endpoints, this stream ID
  562. will be even. For client endpoints, this stream ID will be odd. If no
  563. stream IDs are available, raises :class:`NoAvailableStreamIDError
  564. <h2.exceptions.NoAvailableStreamIDError>`.
  565. .. warning:: The return value from this function does not change until
  566. the stream ID has actually been used by sending or pushing
  567. headers on that stream. For that reason, it should be
  568. called as close as possible to the actual use of the
  569. stream ID.
  570. .. versionadded:: 2.0.0
  571. :raises: :class:`NoAvailableStreamIDError
  572. <h2.exceptions.NoAvailableStreamIDError>`
  573. :returns: The next free stream ID this peer can use to initiate a
  574. stream.
  575. :rtype: ``int``
  576. """
  577. # No streams have been opened yet, so return the lowest allowed stream
  578. # ID.
  579. if not self.highest_outbound_stream_id:
  580. next_stream_id = 1 if self.config.client_side else 2
  581. else:
  582. next_stream_id = self.highest_outbound_stream_id + 2
  583. self.config.logger.debug(
  584. "Next available stream ID %d", next_stream_id,
  585. )
  586. if next_stream_id > self.HIGHEST_ALLOWED_STREAM_ID:
  587. msg = "Exhausted allowed stream IDs"
  588. raise NoAvailableStreamIDError(msg)
  589. return next_stream_id
  590. def send_headers(self,
  591. stream_id: int,
  592. headers: Iterable[HeaderWeaklyTyped],
  593. end_stream: bool = False,
  594. priority_weight: int | None = None,
  595. priority_depends_on: int | None = None,
  596. priority_exclusive: bool | None = None) -> None:
  597. """
  598. Send headers on a given stream.
  599. This function can be used to send request or response headers: the kind
  600. that are sent depends on whether this connection has been opened as a
  601. client or server connection, and whether the stream was opened by the
  602. remote peer or not.
  603. If this is a client connection, calling ``send_headers`` will send the
  604. headers as a request. It will also implicitly open the stream being
  605. used. If this is a client connection and ``send_headers`` has *already*
  606. been called, this will send trailers instead.
  607. If this is a server connection, calling ``send_headers`` will send the
  608. headers as a response. It is a protocol error for a server to open a
  609. stream by sending headers. If this is a server connection and
  610. ``send_headers`` has *already* been called, this will send trailers
  611. instead.
  612. When acting as a server, you may call ``send_headers`` any number of
  613. times allowed by the following rules, in this order:
  614. - zero or more times with ``(':status', '1XX')`` (where ``1XX`` is a
  615. placeholder for any 100-level status code).
  616. - once with any other status header.
  617. - zero or one time for trailers.
  618. That is, you are allowed to send as many informational responses as you
  619. like, followed by one complete response and zero or one HTTP trailer
  620. blocks.
  621. Clients may send one or two header blocks: one request block, and
  622. optionally one trailer block.
  623. If it is important to send HPACK "never indexed" header fields (as
  624. defined in `RFC 7451 Section 7.1.3
  625. <https://tools.ietf.org/html/rfc7541#section-7.1.3>`_), the user may
  626. instead provide headers using the HPACK library's :class:`HeaderTuple
  627. <hpack:hpack.HeaderTuple>` and :class:`NeverIndexedHeaderTuple
  628. <hpack:hpack.NeverIndexedHeaderTuple>` objects.
  629. This method also allows users to prioritize the stream immediately,
  630. by sending priority information on the HEADERS frame directly. To do
  631. this, any one of ``priority_weight``, ``priority_depends_on``, or
  632. ``priority_exclusive`` must be set to a value that is not ``None``. For
  633. more information on the priority fields, see :meth:`prioritize
  634. <h2.connection.H2Connection.prioritize>`.
  635. .. warning:: In HTTP/2, it is mandatory that all the HTTP/2 special
  636. headers (that is, ones whose header keys begin with ``:``) appear
  637. at the start of the header block, before any normal headers.
  638. .. versionchanged:: 2.3.0
  639. Added support for using :class:`HeaderTuple
  640. <hpack:hpack.HeaderTuple>` objects to store headers.
  641. .. versionchanged:: 2.4.0
  642. Added the ability to provide priority keyword arguments:
  643. ``priority_weight``, ``priority_depends_on``, and
  644. ``priority_exclusive``.
  645. :param stream_id: The stream ID to send the headers on. If this stream
  646. does not currently exist, it will be created.
  647. :type stream_id: ``int``
  648. :param headers: The request/response headers to send.
  649. :type headers: An iterable of two tuples of bytestrings or
  650. :class:`HeaderTuple <hpack:hpack.HeaderTuple>` objects.
  651. :param end_stream: Whether this headers frame should end the stream
  652. immediately (that is, whether no more data will be sent after this
  653. frame). Defaults to ``False``.
  654. :type end_stream: ``bool``
  655. :param priority_weight: Sets the priority weight of the stream. See
  656. :meth:`prioritize <h2.connection.H2Connection.prioritize>` for more
  657. about how this field works. Defaults to ``None``, which means that
  658. no priority information will be sent.
  659. :type priority_weight: ``int`` or ``None``
  660. :param priority_depends_on: Sets which stream this one depends on for
  661. priority purposes. See :meth:`prioritize
  662. <h2.connection.H2Connection.prioritize>` for more about how this
  663. field works. Defaults to ``None``, which means that no priority
  664. information will be sent.
  665. :type priority_depends_on: ``int`` or ``None``
  666. :param priority_exclusive: Sets whether this stream exclusively depends
  667. on the stream given in ``priority_depends_on`` for priority
  668. purposes. See :meth:`prioritize
  669. <h2.connection.H2Connection.prioritize>` for more about how this
  670. field workds. Defaults to ``None``, which means that no priority
  671. information will be sent.
  672. :type priority_depends_on: ``bool`` or ``None``
  673. :returns: Nothing
  674. """
  675. self.config.logger.debug(
  676. "Send headers on stream ID %d", stream_id,
  677. )
  678. # Check we can open the stream.
  679. if stream_id not in self.streams:
  680. max_open_streams = self.remote_settings.max_concurrent_streams
  681. value = self.open_outbound_streams # take a copy due to the property accessor having side affects
  682. if (value + 1) > max_open_streams:
  683. msg = f"Max outbound streams is {max_open_streams}, {value} open"
  684. raise TooManyStreamsError(msg)
  685. self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)
  686. stream = self._get_or_create_stream(
  687. stream_id, AllowedStreamIDs(self.config.client_side),
  688. )
  689. frames: list[Frame] = []
  690. frames.extend(stream.send_headers(
  691. headers, self.encoder, end_stream,
  692. ))
  693. # We may need to send priority information.
  694. priority_present = (
  695. (priority_weight is not None) or
  696. (priority_depends_on is not None) or
  697. (priority_exclusive is not None)
  698. )
  699. if priority_present:
  700. if not self.config.client_side:
  701. msg = "Servers SHOULD NOT prioritize streams."
  702. raise RFC1122Error(msg)
  703. headers_frame = frames[0]
  704. assert isinstance(headers_frame, HeadersFrame)
  705. headers_frame.flags.add("PRIORITY")
  706. frames[0] = _add_frame_priority(
  707. headers_frame,
  708. priority_weight,
  709. priority_depends_on,
  710. priority_exclusive,
  711. )
  712. self._prepare_for_sending(frames)
  713. def send_data(self,
  714. stream_id: int,
  715. data: bytes | memoryview,
  716. end_stream: bool = False,
  717. pad_length: Any = None) -> None:
  718. """
  719. Send data on a given stream.
  720. This method does no breaking up of data: if the data is larger than the
  721. value returned by :meth:`local_flow_control_window
  722. <h2.connection.H2Connection.local_flow_control_window>` for this stream
  723. then a :class:`FlowControlError <h2.exceptions.FlowControlError>` will
  724. be raised. If the data is larger than :data:`max_outbound_frame_size
  725. <h2.connection.H2Connection.max_outbound_frame_size>` then a
  726. :class:`FrameTooLargeError <h2.exceptions.FrameTooLargeError>` will be
  727. raised.
  728. h2 does this to avoid buffering the data internally. If the user
  729. has more data to send than h2 will allow, consider breaking it up
  730. and buffering it externally.
  731. :param stream_id: The ID of the stream on which to send the data.
  732. :type stream_id: ``int``
  733. :param data: The data to send on the stream.
  734. :type data: ``bytes``
  735. :param end_stream: (optional) Whether this is the last data to be sent
  736. on the stream. Defaults to ``False``.
  737. :type end_stream: ``bool``
  738. :param pad_length: (optional) Length of the padding to apply to the
  739. data frame. Defaults to ``None`` for no use of padding. Note that
  740. a value of ``0`` results in padding of length ``0``
  741. (with the "padding" flag set on the frame).
  742. .. versionadded:: 2.6.0
  743. :type pad_length: ``int``
  744. :returns: Nothing
  745. """
  746. self.config.logger.debug(
  747. "Send data on stream ID %d with len %d", stream_id, len(data),
  748. )
  749. frame_size = len(data)
  750. if pad_length is not None:
  751. if not isinstance(pad_length, int):
  752. msg = "pad_length must be an int"
  753. raise TypeError(msg)
  754. if pad_length < 0 or pad_length > 255:
  755. msg = "pad_length must be within range: [0, 255]"
  756. raise ValueError(msg)
  757. # Account for padding bytes plus the 1-byte padding length field.
  758. frame_size += pad_length + 1
  759. self.config.logger.debug(
  760. "Frame size on stream ID %d is %d", stream_id, frame_size,
  761. )
  762. if frame_size > self.local_flow_control_window(stream_id):
  763. msg = f"Cannot send {frame_size} bytes, flow control window is {self.local_flow_control_window(stream_id)}"
  764. raise FlowControlError(msg)
  765. if frame_size > self.max_outbound_frame_size:
  766. msg = f"Cannot send frame size {frame_size}, max frame size is {self.max_outbound_frame_size}"
  767. raise FrameTooLargeError(msg)
  768. self.state_machine.process_input(ConnectionInputs.SEND_DATA)
  769. frames = self.streams[stream_id].send_data(
  770. data, end_stream, pad_length=pad_length,
  771. )
  772. self._prepare_for_sending(frames)
  773. self.outbound_flow_control_window -= frame_size
  774. self.config.logger.debug(
  775. "Outbound flow control window size is %d",
  776. self.outbound_flow_control_window,
  777. )
  778. assert self.outbound_flow_control_window >= 0
  779. def end_stream(self, stream_id: int) -> None:
  780. """
  781. Cleanly end a given stream.
  782. This method ends a stream by sending an empty DATA frame on that stream
  783. with the ``END_STREAM`` flag set.
  784. :param stream_id: The ID of the stream to end.
  785. :type stream_id: ``int``
  786. :returns: Nothing
  787. """
  788. self.config.logger.debug("End stream ID %d", stream_id)
  789. self.state_machine.process_input(ConnectionInputs.SEND_DATA)
  790. frames = self.streams[stream_id].end_stream()
  791. self._prepare_for_sending(frames)
  792. def increment_flow_control_window(self, increment: int, stream_id: int | None = None) -> None:
  793. """
  794. Increment a flow control window, optionally for a single stream. Allows
  795. the remote peer to send more data.
  796. .. versionchanged:: 2.0.0
  797. Rejects attempts to increment the flow control window by out of
  798. range values with a ``ValueError``.
  799. :param increment: The amount to increment the flow control window by.
  800. :type increment: ``int``
  801. :param stream_id: (optional) The ID of the stream that should have its
  802. flow control window opened. If not present or ``None``, the
  803. connection flow control window will be opened instead.
  804. :type stream_id: ``int`` or ``None``
  805. :returns: Nothing
  806. :raises: ``ValueError``
  807. """
  808. if not (1 <= increment <= self.MAX_WINDOW_INCREMENT):
  809. msg = f"Flow control increment must be between 1 and {self.MAX_WINDOW_INCREMENT}"
  810. raise ValueError(msg)
  811. self.state_machine.process_input(ConnectionInputs.SEND_WINDOW_UPDATE)
  812. if stream_id is not None:
  813. stream = self.streams[stream_id]
  814. frames = stream.increase_flow_control_window(
  815. increment,
  816. )
  817. self.config.logger.debug(
  818. "Increase stream ID %d flow control window by %d",
  819. stream_id, increment,
  820. )
  821. else:
  822. self._inbound_flow_control_window_manager.window_opened(increment)
  823. f = WindowUpdateFrame(0)
  824. f.window_increment = increment
  825. frames = [f]
  826. self.config.logger.debug(
  827. "Increase connection flow control window by %d", increment,
  828. )
  829. self._prepare_for_sending(frames)
  830. def push_stream(self,
  831. stream_id: int,
  832. promised_stream_id: int,
  833. request_headers: Iterable[HeaderWeaklyTyped]) -> None:
  834. """
  835. Push a response to the client by sending a PUSH_PROMISE frame.
  836. If it is important to send HPACK "never indexed" header fields (as
  837. defined in `RFC 7451 Section 7.1.3
  838. <https://tools.ietf.org/html/rfc7541#section-7.1.3>`_), the user may
  839. instead provide headers using the HPACK library's :class:`HeaderTuple
  840. <hpack:hpack.HeaderTuple>` and :class:`NeverIndexedHeaderTuple
  841. <hpack:hpack.NeverIndexedHeaderTuple>` objects.
  842. :param stream_id: The ID of the stream that this push is a response to.
  843. :type stream_id: ``int``
  844. :param promised_stream_id: The ID of the stream that the pushed
  845. response will be sent on.
  846. :type promised_stream_id: ``int``
  847. :param request_headers: The headers of the request that the pushed
  848. response will be responding to.
  849. :type request_headers: An iterable of two tuples of bytestrings or
  850. :class:`HeaderTuple <hpack:hpack.HeaderTuple>` objects.
  851. :returns: Nothing
  852. """
  853. self.config.logger.debug(
  854. "Send Push Promise frame on stream ID %d", stream_id,
  855. )
  856. if not self.remote_settings.enable_push:
  857. msg = "Remote peer has disabled stream push"
  858. raise ProtocolError(msg)
  859. self.state_machine.process_input(ConnectionInputs.SEND_PUSH_PROMISE)
  860. stream = self._get_stream_by_id(stream_id)
  861. # We need to prevent users pushing streams in response to streams that
  862. # they themselves have already pushed: see #163 and RFC 7540 § 6.6. The
  863. # easiest way to do that is to assert that the stream_id is not even:
  864. # this shortcut works because only servers can push and the state
  865. # machine will enforce this.
  866. if (stream_id % 2) == 0:
  867. msg = "Cannot recursively push streams."
  868. raise ProtocolError(msg)
  869. new_stream = self._begin_new_stream(
  870. promised_stream_id, AllowedStreamIDs.EVEN,
  871. )
  872. self.streams[promised_stream_id] = new_stream
  873. frames = stream.push_stream_in_band(
  874. promised_stream_id, request_headers, self.encoder,
  875. )
  876. new_frames = new_stream.locally_pushed()
  877. self._prepare_for_sending(frames + new_frames)
  878. def ping(self, opaque_data: bytes | str) -> None:
  879. """
  880. Send a PING frame.
  881. :param opaque_data: A bytestring of length 8 that will be sent in the
  882. PING frame.
  883. :returns: Nothing
  884. """
  885. self.config.logger.debug("Send Ping frame")
  886. if not isinstance(opaque_data, bytes) or len(opaque_data) != 8:
  887. msg = f"Invalid value for ping data: {opaque_data!r}"
  888. raise ValueError(msg)
  889. self.state_machine.process_input(ConnectionInputs.SEND_PING)
  890. f = PingFrame(0)
  891. f.opaque_data = opaque_data
  892. self._prepare_for_sending([f])
  893. def reset_stream(self, stream_id: int, error_code: ErrorCodes | int = 0) -> None:
  894. """
  895. Reset a stream.
  896. This method forcibly closes a stream by sending a RST_STREAM frame for
  897. a given stream. This is not a graceful closure. To gracefully end a
  898. stream, try the :meth:`end_stream
  899. <h2.connection.H2Connection.end_stream>` method.
  900. :param stream_id: The ID of the stream to reset.
  901. :type stream_id: ``int``
  902. :param error_code: (optional) The error code to use to reset the
  903. stream. Defaults to :data:`ErrorCodes.NO_ERROR
  904. <h2.errors.ErrorCodes.NO_ERROR>`.
  905. :type error_code: ``int``
  906. :returns: Nothing
  907. """
  908. self.config.logger.debug("Reset stream ID %d", stream_id)
  909. self.state_machine.process_input(ConnectionInputs.SEND_RST_STREAM)
  910. stream = self._get_stream_by_id(stream_id)
  911. frames = stream.reset_stream(error_code)
  912. self._prepare_for_sending(frames)
  913. def close_connection(self,
  914. error_code: ErrorCodes | int = 0,
  915. additional_data: bytes | None = None,
  916. last_stream_id: int | None = None) -> None:
  917. """
  918. Close a connection, emitting a GOAWAY frame.
  919. .. versionchanged:: 2.4.0
  920. Added ``additional_data`` and ``last_stream_id`` arguments.
  921. :param error_code: (optional) The error code to send in the GOAWAY
  922. frame.
  923. :param additional_data: (optional) Additional debug data indicating
  924. a reason for closing the connection. Must be a bytestring.
  925. :param last_stream_id: (optional) The last stream which was processed
  926. by the sender. Defaults to ``highest_inbound_stream_id``.
  927. :returns: Nothing
  928. """
  929. self.config.logger.debug("Close connection")
  930. self.state_machine.process_input(ConnectionInputs.SEND_GOAWAY)
  931. # Additional_data must be bytes
  932. if additional_data is not None:
  933. assert isinstance(additional_data, bytes)
  934. if last_stream_id is None:
  935. last_stream_id = self.highest_inbound_stream_id
  936. f = GoAwayFrame(
  937. stream_id=0,
  938. last_stream_id=last_stream_id,
  939. error_code=error_code,
  940. additional_data=(additional_data or b""),
  941. )
  942. self._prepare_for_sending([f])
  943. def update_settings(self, new_settings: dict[SettingCodes | int, int]) -> None:
  944. """
  945. Update the local settings. This will prepare and emit the appropriate
  946. SETTINGS frame.
  947. :param new_settings: A dictionary of {setting: new value}
  948. """
  949. self.config.logger.debug(
  950. "Update connection settings to %s", new_settings,
  951. )
  952. self.state_machine.process_input(ConnectionInputs.SEND_SETTINGS)
  953. self.local_settings.update(new_settings)
  954. s = SettingsFrame(0)
  955. s.settings = new_settings
  956. self._prepare_for_sending([s])
  957. def advertise_alternative_service(self,
  958. field_value: bytes | str,
  959. origin: bytes | None = None,
  960. stream_id: int | None = None) -> None:
  961. """
  962. Notify a client about an available Alternative Service.
  963. An Alternative Service is defined in `RFC 7838
  964. <https://tools.ietf.org/html/rfc7838>`_. An Alternative Service
  965. notification informs a client that a given origin is also available
  966. elsewhere.
  967. Alternative Services can be advertised in two ways. Firstly, they can
  968. be advertised explicitly: that is, a server can say "origin X is also
  969. available at Y". To advertise like this, set the ``origin`` argument
  970. and not the ``stream_id`` argument. Alternatively, they can be
  971. advertised implicitly: that is, a server can say "the origin you're
  972. contacting on stream X is also available at Y". To advertise like this,
  973. set the ``stream_id`` argument and not the ``origin`` argument.
  974. The explicit method of advertising can be done as long as the
  975. connection is active. The implicit method can only be done after the
  976. client has sent the request headers and before the server has sent the
  977. response headers: outside of those points, h2 will forbid sending
  978. the Alternative Service advertisement by raising a ProtocolError.
  979. The ``field_value`` parameter is specified in RFC 7838. h2 does
  980. not validate or introspect this argument: the user is required to
  981. ensure that it's well-formed. ``field_value`` corresponds to RFC 7838's
  982. "Alternative Service Field Value".
  983. .. note:: It is strongly preferred to use the explicit method of
  984. advertising Alternative Services. The implicit method of
  985. advertising Alternative Services has a number of subtleties
  986. and can lead to inconsistencies between the server and
  987. client. h2 allows both mechanisms, but caution is
  988. strongly advised.
  989. .. versionadded:: 2.3.0
  990. :param field_value: The RFC 7838 Alternative Service Field Value. This
  991. argument is not introspected by h2: the user is responsible
  992. for ensuring that it is well-formed.
  993. :type field_value: ``bytes``
  994. :param origin: The origin/authority to which the Alternative Service
  995. being advertised applies. Must not be provided at the same time as
  996. ``stream_id``.
  997. :type origin: ``bytes`` or ``None``
  998. :param stream_id: The ID of the stream which was sent to the authority
  999. for which this Alternative Service advertisement applies. Must not
  1000. be provided at the same time as ``origin``.
  1001. :type stream_id: ``int`` or ``None``
  1002. :returns: Nothing.
  1003. """
  1004. if not isinstance(field_value, bytes):
  1005. msg = "Field must be bytestring."
  1006. raise ValueError(msg) # noqa: TRY004
  1007. if origin is not None and stream_id is not None:
  1008. msg = "Must not provide both origin and stream_id"
  1009. raise ValueError(msg)
  1010. self.state_machine.process_input(
  1011. ConnectionInputs.SEND_ALTERNATIVE_SERVICE,
  1012. )
  1013. if origin is not None:
  1014. # This ALTSVC is sent on stream zero.
  1015. f = AltSvcFrame(stream_id=0)
  1016. f.origin = origin
  1017. f.field = field_value
  1018. frames: list[Frame] = [f]
  1019. else:
  1020. stream = self._get_stream_by_id(stream_id)
  1021. frames = stream.advertise_alternative_service(field_value)
  1022. self._prepare_for_sending(frames)
  1023. def prioritize(self,
  1024. stream_id: int,
  1025. weight: int | None = None,
  1026. depends_on: int | None = None,
  1027. exclusive: bool | None = None) -> None:
  1028. """
  1029. Notify a server about the priority of a stream.
  1030. Stream priorities are a form of guidance to a remote server: they
  1031. inform the server about how important a given response is, so that the
  1032. server may allocate its resources (e.g. bandwidth, CPU time, etc.)
  1033. accordingly. This exists to allow clients to ensure that the most
  1034. important data arrives earlier, while less important data does not
  1035. starve out the more important data.
  1036. Stream priorities are explained in depth in `RFC 7540 Section 5.3
  1037. <https://tools.ietf.org/html/rfc7540#section-5.3>`_.
  1038. This method updates the priority information of a single stream. It may
  1039. be called well before a stream is actively in use, or well after a
  1040. stream is closed.
  1041. .. warning:: RFC 7540 allows for servers to change the priority of
  1042. streams. However, h2 **does not** allow server
  1043. stacks to do this. This is because most clients do not
  1044. adequately know how to respond when provided conflicting
  1045. priority information, and relatively little utility is
  1046. provided by making that functionality available.
  1047. .. note:: h2 **does not** maintain any information about the
  1048. RFC 7540 priority tree. That means that h2 does not
  1049. prevent incautious users from creating invalid priority
  1050. trees, particularly by creating priority loops. While some
  1051. basic error checking is provided by h2, users are
  1052. strongly recommended to understand their prioritisation
  1053. strategies before using the priority tools here.
  1054. .. note:: Priority information is strictly advisory. Servers are
  1055. allowed to disregard it entirely. Avoid relying on the idea
  1056. that your priority signaling will definitely be obeyed.
  1057. .. versionadded:: 2.4.0
  1058. :param stream_id: The ID of the stream to prioritize.
  1059. :type stream_id: ``int``
  1060. :param weight: The weight to give the stream. Defaults to ``16``, the
  1061. default weight of any stream. May be any value between ``1`` and
  1062. ``256`` inclusive. The relative weight of a stream indicates what
  1063. proportion of available resources will be allocated to that
  1064. stream.
  1065. :type weight: ``int``
  1066. :param depends_on: The ID of the stream on which this stream depends.
  1067. This stream will only be progressed if it is impossible to
  1068. progress the parent stream (the one on which this one depends).
  1069. Passing the value ``0`` means that this stream does not depend on
  1070. any other. Defaults to ``0``.
  1071. :type depends_on: ``int``
  1072. :param exclusive: Whether this stream is an exclusive dependency of its
  1073. "parent" stream (i.e. the stream given by ``depends_on``). If a
  1074. stream is an exclusive dependency of another, that means that all
  1075. previously-set children of the parent are moved to become children
  1076. of the new exclusively-dependent stream. Defaults to ``False``.
  1077. :type exclusive: ``bool``
  1078. """
  1079. if not self.config.client_side:
  1080. msg = "Servers SHOULD NOT prioritize streams."
  1081. raise RFC1122Error(msg)
  1082. self.state_machine.process_input(
  1083. ConnectionInputs.SEND_PRIORITY,
  1084. )
  1085. frame = PriorityFrame(stream_id)
  1086. frame_prio = _add_frame_priority(frame, weight, depends_on, exclusive)
  1087. self._prepare_for_sending([frame_prio])
  1088. def local_flow_control_window(self, stream_id: int) -> int:
  1089. """
  1090. Returns the maximum amount of data that can be sent on stream
  1091. ``stream_id``.
  1092. This value will never be larger than the total data that can be sent on
  1093. the connection: even if the given stream allows more data, the
  1094. connection window provides a logical maximum to the amount of data that
  1095. can be sent.
  1096. The maximum data that can be sent in a single data frame on a stream
  1097. is either this value, or the maximum frame size, whichever is
  1098. *smaller*.
  1099. :param stream_id: The ID of the stream whose flow control window is
  1100. being queried.
  1101. :type stream_id: ``int``
  1102. :returns: The amount of data in bytes that can be sent on the stream
  1103. before the flow control window is exhausted.
  1104. :rtype: ``int``
  1105. """
  1106. stream = self._get_stream_by_id(stream_id)
  1107. return min(
  1108. self.outbound_flow_control_window,
  1109. stream.outbound_flow_control_window,
  1110. )
  1111. def remote_flow_control_window(self, stream_id: int) -> int:
  1112. """
  1113. Returns the maximum amount of data the remote peer can send on stream
  1114. ``stream_id``.
  1115. This value will never be larger than the total data that can be sent on
  1116. the connection: even if the given stream allows more data, the
  1117. connection window provides a logical maximum to the amount of data that
  1118. can be sent.
  1119. The maximum data that can be sent in a single data frame on a stream
  1120. is either this value, or the maximum frame size, whichever is
  1121. *smaller*.
  1122. :param stream_id: The ID of the stream whose flow control window is
  1123. being queried.
  1124. :type stream_id: ``int``
  1125. :returns: The amount of data in bytes that can be received on the
  1126. stream before the flow control window is exhausted.
  1127. :rtype: ``int``
  1128. """
  1129. stream = self._get_stream_by_id(stream_id)
  1130. return min(
  1131. self.inbound_flow_control_window,
  1132. stream.inbound_flow_control_window,
  1133. )
  1134. def acknowledge_received_data(self, acknowledged_size: int, stream_id: int) -> None:
  1135. """
  1136. Inform the :class:`H2Connection <h2.connection.H2Connection>` that a
  1137. certain number of flow-controlled bytes have been processed, and that
  1138. the space should be handed back to the remote peer at an opportune
  1139. time.
  1140. .. versionadded:: 2.5.0
  1141. :param acknowledged_size: The total *flow-controlled size* of the data
  1142. that has been processed. Note that this must include the amount of
  1143. padding that was sent with that data.
  1144. :type acknowledged_size: ``int``
  1145. :param stream_id: The ID of the stream on which this data was received.
  1146. :type stream_id: ``int``
  1147. :returns: Nothing
  1148. :rtype: ``None``
  1149. """
  1150. self.config.logger.debug(
  1151. "Ack received data on stream ID %d with size %d",
  1152. stream_id, acknowledged_size,
  1153. )
  1154. if stream_id <= 0:
  1155. msg = f"Stream ID {stream_id} is not valid for acknowledge_received_data"
  1156. raise ValueError(msg)
  1157. if acknowledged_size < 0:
  1158. msg = "Cannot acknowledge negative data"
  1159. raise ValueError(msg)
  1160. frames: list[Frame] = []
  1161. conn_manager = self._inbound_flow_control_window_manager
  1162. conn_increment = conn_manager.process_bytes(acknowledged_size)
  1163. if conn_increment:
  1164. f = WindowUpdateFrame(0)
  1165. f.window_increment = conn_increment
  1166. frames.append(f)
  1167. try:
  1168. stream = self._get_stream_by_id(stream_id)
  1169. except StreamClosedError:
  1170. # The stream is already gone. We're not worried about incrementing
  1171. # the window in this case.
  1172. pass
  1173. else:
  1174. # No point incrementing the windows of closed streams.
  1175. if stream.open:
  1176. frames.extend(
  1177. stream.acknowledge_received_data(acknowledged_size),
  1178. )
  1179. self._prepare_for_sending(frames)
  1180. def data_to_send(self, amount: int | None = None) -> bytes:
  1181. """
  1182. Returns some data for sending out of the internal data buffer.
  1183. This method is analogous to ``read`` on a file-like object, but it
  1184. doesn't block. Instead, it returns as much data as the user asks for,
  1185. or less if that much data is not available. It does not perform any
  1186. I/O, and so uses a different name.
  1187. :param amount: (optional) The maximum amount of data to return. If not
  1188. set, or set to ``None``, will return as much data as possible.
  1189. :type amount: ``int``
  1190. :returns: A bytestring containing the data to send on the wire.
  1191. :rtype: ``bytes``
  1192. """
  1193. if amount is None:
  1194. data = bytes(self._data_to_send)
  1195. self._data_to_send = bytearray()
  1196. return data
  1197. data = bytes(self._data_to_send[:amount])
  1198. self._data_to_send = self._data_to_send[amount:]
  1199. return data
  1200. def clear_outbound_data_buffer(self) -> None:
  1201. """
  1202. Clears the outbound data buffer, such that if this call was immediately
  1203. followed by a call to
  1204. :meth:`data_to_send <h2.connection.H2Connection.data_to_send>`, that
  1205. call would return no data.
  1206. This method should not normally be used, but is made available to avoid
  1207. exposing implementation details.
  1208. """
  1209. self._data_to_send = bytearray()
  1210. def _acknowledge_settings(self) -> list[Frame]:
  1211. """
  1212. Acknowledge settings that have been received.
  1213. .. versionchanged:: 2.0.0
  1214. Removed from public API, removed useless ``event`` parameter, made
  1215. automatic.
  1216. :returns: Nothing
  1217. """
  1218. self.state_machine.process_input(ConnectionInputs.SEND_SETTINGS)
  1219. changes = self.remote_settings.acknowledge()
  1220. if SettingCodes.INITIAL_WINDOW_SIZE in changes:
  1221. setting = changes[SettingCodes.INITIAL_WINDOW_SIZE]
  1222. self._flow_control_change_from_settings(
  1223. setting.original_value,
  1224. setting.new_value,
  1225. )
  1226. # HEADER_TABLE_SIZE changes by the remote part affect our encoder: cf.
  1227. # RFC 7540 Section 6.5.2.
  1228. if SettingCodes.HEADER_TABLE_SIZE in changes:
  1229. setting = changes[SettingCodes.HEADER_TABLE_SIZE]
  1230. self.encoder.header_table_size = setting.new_value
  1231. if SettingCodes.MAX_FRAME_SIZE in changes:
  1232. setting = changes[SettingCodes.MAX_FRAME_SIZE]
  1233. self.max_outbound_frame_size = setting.new_value
  1234. for stream in self.streams.values():
  1235. stream.max_outbound_frame_size = setting.new_value
  1236. f = SettingsFrame(0)
  1237. f.flags.add("ACK")
  1238. return [f]
  1239. def _flow_control_change_from_settings(self, old_value: int | None, new_value: int) -> None:
  1240. """
  1241. Update flow control windows in response to a change in the value of
  1242. SETTINGS_INITIAL_WINDOW_SIZE.
  1243. When this setting is changed, it automatically updates all flow control
  1244. windows by the delta in the settings values. Note that it does not
  1245. increment the *connection* flow control window, per section 6.9.2 of
  1246. RFC 7540.
  1247. """
  1248. delta = new_value - (old_value or 0)
  1249. for stream in self.streams.values():
  1250. stream.outbound_flow_control_window = guard_increment_window(
  1251. stream.outbound_flow_control_window,
  1252. delta,
  1253. )
  1254. def _inbound_flow_control_change_from_settings(self, old_value: int | None, new_value: int) -> None:
  1255. """
  1256. Update remote flow control windows in response to a change in the value
  1257. of SETTINGS_INITIAL_WINDOW_SIZE.
  1258. When this setting is changed, it automatically updates all remote flow
  1259. control windows by the delta in the settings values.
  1260. """
  1261. delta = new_value - (old_value or 0)
  1262. for stream in self.streams.values():
  1263. stream._inbound_flow_control_change_from_settings(delta)
  1264. def receive_data(self, data: bytes) -> list[Event]:
  1265. """
  1266. Pass some received HTTP/2 data to the connection for handling.
  1267. :param data: The data received from the remote peer on the network.
  1268. :type data: ``bytes``
  1269. :returns: A list of events that the remote peer triggered by sending
  1270. this data.
  1271. """
  1272. self.config.logger.trace(
  1273. "Process received data on connection. Received data: %r", data,
  1274. )
  1275. events: list[Event] = []
  1276. self.incoming_buffer.add_data(data)
  1277. self.incoming_buffer.max_frame_size = self.max_inbound_frame_size
  1278. try:
  1279. for frame in self.incoming_buffer:
  1280. events.extend(self._receive_frame(frame))
  1281. except InvalidPaddingError as e:
  1282. self._terminate_connection(ErrorCodes.PROTOCOL_ERROR)
  1283. msg = "Received frame with invalid padding."
  1284. raise ProtocolError(msg) from e
  1285. except ProtocolError as e:
  1286. # For whatever reason, receiving the frame caused a protocol error.
  1287. # We should prepare to emit a GoAway frame before throwing the
  1288. # exception up further. No need for an event: the exception will
  1289. # do fine.
  1290. self._terminate_connection(e.error_code)
  1291. raise
  1292. return events
  1293. def _receive_frame(self, frame: Frame) -> list[Event]:
  1294. """
  1295. Handle a frame received on the connection.
  1296. .. versionchanged:: 2.0.0
  1297. Removed from the public API.
  1298. """
  1299. events: list[Event]
  1300. self.config.logger.trace("Received frame: %s", repr(frame))
  1301. try:
  1302. # I don't love using __class__ here, maybe reconsider it.
  1303. frames, events = self._frame_dispatch_table[frame.__class__](frame)
  1304. except StreamClosedError as e:
  1305. # If the stream was closed by RST_STREAM, we just send a RST_STREAM
  1306. # to the remote peer. Otherwise, this is a connection error, and so
  1307. # we will re-raise to trigger one.
  1308. if self._stream_is_closed_by_reset(e.stream_id):
  1309. f = RstStreamFrame(e.stream_id)
  1310. f.error_code = e.error_code
  1311. self._prepare_for_sending([f])
  1312. events = e._events
  1313. else:
  1314. raise
  1315. except StreamIDTooLowError as e:
  1316. # The stream ID seems invalid. This may happen when the closed
  1317. # stream has been cleaned up, or when the remote peer has opened a
  1318. # new stream with a higher stream ID than this one, forcing it
  1319. # closed implicitly.
  1320. #
  1321. # Check how the stream was closed: depending on the mechanism, it
  1322. # is either a stream error or a connection error.
  1323. if self._stream_is_closed_by_reset(e.stream_id):
  1324. # Closed by RST_STREAM is a stream error.
  1325. f = RstStreamFrame(e.stream_id)
  1326. f.error_code = ErrorCodes.STREAM_CLOSED
  1327. self._prepare_for_sending([f])
  1328. events = []
  1329. elif self._stream_is_closed_by_end(e.stream_id):
  1330. # Closed by END_STREAM is a connection error.
  1331. raise StreamClosedError(e.stream_id) from e
  1332. else:
  1333. # Closed implicitly, also a connection error, but of type
  1334. # PROTOCOL_ERROR.
  1335. raise
  1336. else:
  1337. self._prepare_for_sending(frames)
  1338. return events
  1339. def _terminate_connection(self, error_code: ErrorCodes) -> None:
  1340. """
  1341. Terminate the connection early. Used in error handling blocks to send
  1342. GOAWAY frames.
  1343. """
  1344. f = GoAwayFrame(0)
  1345. f.last_stream_id = self.highest_inbound_stream_id
  1346. f.error_code = error_code
  1347. self.state_machine.process_input(ConnectionInputs.SEND_GOAWAY)
  1348. self._prepare_for_sending([f])
  1349. def _receive_headers_frame(self, frame: HeadersFrame) -> tuple[list[Frame], list[Event]]:
  1350. """
  1351. Receive a headers frame on the connection.
  1352. """
  1353. # If necessary, check we can open the stream. Also validate that the
  1354. # stream ID is valid.
  1355. if frame.stream_id not in self.streams:
  1356. max_open_streams = self.local_settings.max_concurrent_streams
  1357. value = self.open_inbound_streams # take a copy due to the property accessor having side affects
  1358. if (value + 1) > max_open_streams:
  1359. msg = f"Max inbound streams is {max_open_streams}, {value} open"
  1360. raise TooManyStreamsError(msg)
  1361. # Let's decode the headers. We handle headers as bytes internally up
  1362. # until we hang them off the event, at which point we may optionally
  1363. # convert them to unicode.
  1364. headers = _decode_headers(self.decoder, frame.data)
  1365. events = self.state_machine.process_input(
  1366. ConnectionInputs.RECV_HEADERS,
  1367. )
  1368. stream = self._get_or_create_stream(
  1369. frame.stream_id, AllowedStreamIDs(not self.config.client_side),
  1370. )
  1371. frames, stream_events = stream.receive_headers(
  1372. headers,
  1373. "END_STREAM" in frame.flags,
  1374. self.config.header_encoding,
  1375. )
  1376. if "PRIORITY" in frame.flags:
  1377. p_frames, p_events = self._receive_priority_frame(frame)
  1378. expected_frame_types = (RequestReceived, ResponseReceived, TrailersReceived, InformationalResponseReceived)
  1379. assert isinstance(stream_events[0], expected_frame_types)
  1380. assert isinstance(p_events[0], PriorityUpdated)
  1381. stream_events[0].priority_updated = p_events[0]
  1382. stream_events.extend(p_events)
  1383. assert not p_frames
  1384. return frames, events + stream_events
  1385. def _receive_push_promise_frame(self, frame: PushPromiseFrame) -> tuple[list[Frame], list[Event]]:
  1386. """
  1387. Receive a push-promise frame on the connection.
  1388. """
  1389. if not self.local_settings.enable_push:
  1390. msg = "Received pushed stream"
  1391. raise ProtocolError(msg)
  1392. pushed_headers = _decode_headers(self.decoder, frame.data)
  1393. events = self.state_machine.process_input(
  1394. ConnectionInputs.RECV_PUSH_PROMISE,
  1395. )
  1396. try:
  1397. stream = self._get_stream_by_id(frame.stream_id)
  1398. except NoSuchStreamError as e:
  1399. # We need to check if the parent stream was reset by us. If it was
  1400. # then we presume that the PUSH_PROMISE was in flight when we reset
  1401. # the parent stream. Rather than accept the new stream, just reset
  1402. # it.
  1403. #
  1404. # If this was closed naturally, however, we should call this a
  1405. # PROTOCOL_ERROR: pushing a stream on a naturally closed stream is
  1406. # a real problem because it creates a brand new stream that the
  1407. # remote peer now believes exists.
  1408. if (self._stream_closed_by(frame.stream_id) ==
  1409. StreamClosedBy.SEND_RST_STREAM):
  1410. f = RstStreamFrame(frame.promised_stream_id)
  1411. f.error_code = ErrorCodes.REFUSED_STREAM
  1412. return [f], events
  1413. msg = "Attempted to push on closed stream."
  1414. raise ProtocolError(msg) from e
  1415. # We need to prevent peers pushing streams in response to streams that
  1416. # they themselves have already pushed: see #163 and RFC 7540 § 6.6. The
  1417. # easiest way to do that is to assert that the stream_id is not even:
  1418. # this shortcut works because only servers can push and the state
  1419. # machine will enforce this.
  1420. if (frame.stream_id % 2) == 0:
  1421. msg = "Cannot recursively push streams."
  1422. raise ProtocolError(msg)
  1423. try:
  1424. frames, stream_events = stream.receive_push_promise_in_band(
  1425. frame.promised_stream_id,
  1426. pushed_headers,
  1427. self.config.header_encoding,
  1428. )
  1429. except StreamClosedError:
  1430. # The parent stream was reset by us, so we presume that
  1431. # PUSH_PROMISE was in flight when we reset the parent stream.
  1432. # So we just reset the new stream.
  1433. f = RstStreamFrame(frame.promised_stream_id)
  1434. f.error_code = ErrorCodes.REFUSED_STREAM
  1435. return [f], events
  1436. new_stream = self._begin_new_stream(
  1437. frame.promised_stream_id, AllowedStreamIDs.EVEN,
  1438. )
  1439. self.streams[frame.promised_stream_id] = new_stream
  1440. new_stream.remotely_pushed(pushed_headers)
  1441. return frames, events + stream_events
  1442. def _handle_data_on_closed_stream(self,
  1443. events: list[Event],
  1444. exc: StreamClosedError,
  1445. frame: DataFrame) -> tuple[list[Frame], list[Event]]:
  1446. # This stream is already closed - and yet we received a DATA frame.
  1447. # The received DATA frame counts towards the connection flow window.
  1448. # We need to manually to acknowledge the DATA frame to update the flow
  1449. # window of the connection. Otherwise the whole connection stalls due
  1450. # the inbound flow window being 0.
  1451. frames: list[Frame] = []
  1452. conn_manager = self._inbound_flow_control_window_manager
  1453. conn_increment = conn_manager.process_bytes(
  1454. frame.flow_controlled_length,
  1455. )
  1456. if conn_increment:
  1457. window_update_frame = WindowUpdateFrame(0)
  1458. window_update_frame.window_increment = conn_increment
  1459. frames.append(window_update_frame)
  1460. self.config.logger.debug(
  1461. "Received DATA frame on closed stream %d - "
  1462. "auto-emitted a WINDOW_UPDATE by %d",
  1463. frame.stream_id, conn_increment,
  1464. )
  1465. rst_stream_frame = RstStreamFrame(exc.stream_id)
  1466. rst_stream_frame.error_code = exc.error_code
  1467. frames.append(rst_stream_frame)
  1468. self.config.logger.debug(
  1469. "Stream %s already CLOSED or cleaned up - auto-emitted a RST_FRAME",
  1470. frame.stream_id,
  1471. )
  1472. return frames, events + exc._events
  1473. def _receive_data_frame(self, frame: DataFrame) -> tuple[list[Frame], list[Event]]:
  1474. """
  1475. Receive a data frame on the connection.
  1476. """
  1477. flow_controlled_length = frame.flow_controlled_length
  1478. events = self.state_machine.process_input(
  1479. ConnectionInputs.RECV_DATA,
  1480. )
  1481. self._inbound_flow_control_window_manager.window_consumed(
  1482. flow_controlled_length,
  1483. )
  1484. try:
  1485. stream = self._get_stream_by_id(frame.stream_id)
  1486. frames, stream_events = stream.receive_data(
  1487. frame.data,
  1488. "END_STREAM" in frame.flags,
  1489. flow_controlled_length,
  1490. )
  1491. except StreamClosedError as e:
  1492. # This stream is either marked as CLOSED or already gone from our
  1493. # internal state.
  1494. return self._handle_data_on_closed_stream(events, e, frame)
  1495. return frames, events + stream_events
  1496. def _receive_settings_frame(self, frame: SettingsFrame) -> tuple[list[Frame], list[Event]]:
  1497. """
  1498. Receive a SETTINGS frame on the connection.
  1499. """
  1500. events = self.state_machine.process_input(
  1501. ConnectionInputs.RECV_SETTINGS,
  1502. )
  1503. # This is an ack of the local settings.
  1504. if "ACK" in frame.flags:
  1505. changed_settings = self._local_settings_acked()
  1506. ack_event = SettingsAcknowledged()
  1507. ack_event.changed_settings = changed_settings
  1508. events.append(ack_event)
  1509. return [], events
  1510. # Add the new settings.
  1511. self.remote_settings.update(frame.settings)
  1512. events.append(
  1513. RemoteSettingsChanged.from_settings(
  1514. self.remote_settings, frame.settings,
  1515. ),
  1516. )
  1517. frames = self._acknowledge_settings()
  1518. return frames, events
  1519. def _receive_window_update_frame(self, frame: WindowUpdateFrame) -> tuple[list[Frame], list[Event]]:
  1520. """
  1521. Receive a WINDOW_UPDATE frame on the connection.
  1522. """
  1523. # hyperframe will take care of validating the window_increment.
  1524. # If we reach in here, we can assume a valid value.
  1525. events = self.state_machine.process_input(
  1526. ConnectionInputs.RECV_WINDOW_UPDATE,
  1527. )
  1528. if frame.stream_id:
  1529. try:
  1530. stream = self._get_stream_by_id(frame.stream_id)
  1531. frames, stream_events = stream.receive_window_update(
  1532. frame.window_increment,
  1533. )
  1534. except StreamClosedError:
  1535. return [], events
  1536. else:
  1537. # Increment our local flow control window.
  1538. self.outbound_flow_control_window = guard_increment_window(
  1539. self.outbound_flow_control_window,
  1540. frame.window_increment,
  1541. )
  1542. # FIXME: Should we split this into one event per active stream?
  1543. window_updated_event = WindowUpdated(stream_id=0, delta=frame.window_increment)
  1544. stream_events = [window_updated_event]
  1545. frames = []
  1546. return frames, events + stream_events
  1547. def _receive_ping_frame(self, frame: PingFrame) -> tuple[list[Frame], list[Event]]:
  1548. """
  1549. Receive a PING frame on the connection.
  1550. """
  1551. events = self.state_machine.process_input(
  1552. ConnectionInputs.RECV_PING,
  1553. )
  1554. frames: list[Frame] = []
  1555. evt: PingReceived | PingAckReceived
  1556. if "ACK" in frame.flags:
  1557. evt = PingAckReceived(ping_data=frame.opaque_data)
  1558. else:
  1559. evt = PingReceived(ping_data=frame.opaque_data)
  1560. # automatically ACK the PING with the same 'opaque data'
  1561. f = PingFrame(0)
  1562. f.flags.add("ACK")
  1563. f.opaque_data = frame.opaque_data
  1564. frames.append(f)
  1565. events.append(evt)
  1566. return frames, events
  1567. def _receive_rst_stream_frame(self, frame: RstStreamFrame) -> tuple[list[Frame], list[Event]]:
  1568. """
  1569. Receive a RST_STREAM frame on the connection.
  1570. """
  1571. events = self.state_machine.process_input(
  1572. ConnectionInputs.RECV_RST_STREAM,
  1573. )
  1574. try:
  1575. stream = self._get_stream_by_id(frame.stream_id)
  1576. except NoSuchStreamError:
  1577. # The stream is missing. That's ok, we just do nothing here.
  1578. stream_frames: list[Frame] = []
  1579. stream_events: list[Event] = []
  1580. else:
  1581. stream_frames, stream_events = stream.stream_reset(frame)
  1582. return stream_frames, events + stream_events
  1583. def _receive_priority_frame(self, frame: HeadersFrame | PriorityFrame) -> tuple[list[Frame], list[Event]]:
  1584. """
  1585. Receive a PRIORITY frame on the connection.
  1586. """
  1587. events = self.state_machine.process_input(
  1588. ConnectionInputs.RECV_PRIORITY,
  1589. )
  1590. event = PriorityUpdated()
  1591. event.stream_id = frame.stream_id
  1592. event.depends_on = frame.depends_on
  1593. event.exclusive = frame.exclusive
  1594. # Weight is an integer between 1 and 256, but the byte only allows
  1595. # 0 to 255: add one.
  1596. event.weight = frame.stream_weight + 1
  1597. # A stream may not depend on itself.
  1598. if event.depends_on == frame.stream_id:
  1599. msg = f"Stream {frame.stream_id} may not depend on itself"
  1600. raise ProtocolError(msg)
  1601. events.append(event)
  1602. return [], events
  1603. def _receive_goaway_frame(self, frame: GoAwayFrame) -> tuple[list[Frame], list[Event]]:
  1604. """
  1605. Receive a GOAWAY frame on the connection.
  1606. """
  1607. events = self.state_machine.process_input(
  1608. ConnectionInputs.RECV_GOAWAY,
  1609. )
  1610. # Clear the outbound data buffer: we cannot send further data now.
  1611. self.clear_outbound_data_buffer()
  1612. # Fire an appropriate ConnectionTerminated event.
  1613. new_event = ConnectionTerminated()
  1614. new_event.error_code = _error_code_from_int(frame.error_code)
  1615. new_event.last_stream_id = frame.last_stream_id
  1616. new_event.additional_data = (frame.additional_data
  1617. if frame.additional_data else None)
  1618. events.append(new_event)
  1619. return [], events
  1620. def _receive_naked_continuation(self, frame: ContinuationFrame) -> None:
  1621. """
  1622. A naked CONTINUATION frame has been received. This is always an error,
  1623. but the type of error it is depends on the state of the stream and must
  1624. transition the state of the stream, so we need to pass it to the
  1625. appropriate stream.
  1626. """
  1627. stream = self._get_stream_by_id(frame.stream_id)
  1628. stream.receive_continuation()
  1629. msg = "Should not be reachable" # pragma: no cover
  1630. raise AssertionError(msg) # pragma: no cover
  1631. def _receive_alt_svc_frame(self, frame: AltSvcFrame) -> tuple[list[Frame], list[Event]]:
  1632. """
  1633. An ALTSVC frame has been received. This frame, specified in RFC 7838,
  1634. is used to advertise alternative places where the same service can be
  1635. reached.
  1636. This frame can optionally be received either on a stream or on stream
  1637. 0, and its semantics are different in each case.
  1638. """
  1639. events = self.state_machine.process_input(
  1640. ConnectionInputs.RECV_ALTERNATIVE_SERVICE,
  1641. )
  1642. frames = []
  1643. if frame.stream_id:
  1644. # Given that it makes no sense to receive ALTSVC on a stream
  1645. # before that stream has been opened with a HEADERS frame, the
  1646. # ALTSVC frame cannot create a stream. If the stream is not
  1647. # present, we simply ignore the frame.
  1648. try:
  1649. stream = self._get_stream_by_id(frame.stream_id)
  1650. except (NoSuchStreamError, StreamClosedError):
  1651. pass
  1652. else:
  1653. stream_frames, stream_events = stream.receive_alt_svc(frame)
  1654. frames.extend(stream_frames)
  1655. events.extend(stream_events)
  1656. else:
  1657. # This frame is sent on stream 0. The origin field on the frame
  1658. # must be present, though if it isn't it's not a ProtocolError
  1659. # (annoyingly), we just need to ignore it.
  1660. if not frame.origin:
  1661. return frames, events
  1662. # If we're a server, we want to ignore this (RFC 7838 says so).
  1663. if not self.config.client_side:
  1664. return frames, events
  1665. event = AlternativeServiceAvailable()
  1666. event.origin = frame.origin
  1667. event.field_value = frame.field
  1668. events.append(event)
  1669. return frames, events
  1670. def _receive_unknown_frame(self, frame: ExtensionFrame) -> tuple[list[Frame], list[Event]]:
  1671. """
  1672. We have received a frame that we do not understand. This is almost
  1673. certainly an extension frame, though it's impossible to be entirely
  1674. sure.
  1675. RFC 7540 § 5.5 says that we MUST ignore unknown frame types: so we
  1676. do. We do notify the user that we received one, however.
  1677. """
  1678. # All we do here is log.
  1679. self.config.logger.debug(
  1680. "Received unknown extension frame (ID %d)", frame.stream_id,
  1681. )
  1682. event = UnknownFrameReceived(frame=frame)
  1683. return [], [event]
  1684. def _local_settings_acked(self) -> dict[SettingCodes | int, ChangedSetting]:
  1685. """
  1686. Handle the local settings being ACKed, update internal state.
  1687. """
  1688. changes = self.local_settings.acknowledge()
  1689. if SettingCodes.INITIAL_WINDOW_SIZE in changes:
  1690. setting = changes[SettingCodes.INITIAL_WINDOW_SIZE]
  1691. self._inbound_flow_control_change_from_settings(
  1692. setting.original_value,
  1693. setting.new_value,
  1694. )
  1695. if SettingCodes.MAX_HEADER_LIST_SIZE in changes:
  1696. setting = changes[SettingCodes.MAX_HEADER_LIST_SIZE]
  1697. self.decoder.max_header_list_size = setting.new_value
  1698. if SettingCodes.MAX_FRAME_SIZE in changes:
  1699. setting = changes[SettingCodes.MAX_FRAME_SIZE]
  1700. self.max_inbound_frame_size = setting.new_value
  1701. if SettingCodes.HEADER_TABLE_SIZE in changes:
  1702. setting = changes[SettingCodes.HEADER_TABLE_SIZE]
  1703. # This is safe across all hpack versions: some versions just won't
  1704. # respect it.
  1705. self.decoder.max_allowed_table_size = setting.new_value
  1706. return changes
  1707. def _stream_id_is_outbound(self, stream_id: int) -> bool:
  1708. """
  1709. Returns ``True`` if the stream ID corresponds to an outbound stream
  1710. (one initiated by this peer), returns ``False`` otherwise.
  1711. """
  1712. return (stream_id % 2 == int(self.config.client_side))
  1713. def _stream_closed_by(self, stream_id: int) -> StreamClosedBy | None:
  1714. """
  1715. Returns how the stream was closed.
  1716. The return value will be either a member of
  1717. ``h2.stream.StreamClosedBy`` or ``None``. If ``None``, the stream was
  1718. closed implicitly by the peer opening a stream with a higher stream ID
  1719. before opening this one.
  1720. """
  1721. if stream_id in self.streams:
  1722. return self.streams[stream_id].closed_by
  1723. if stream_id in self._closed_streams:
  1724. return self._closed_streams[stream_id]
  1725. return None
  1726. def _stream_is_closed_by_reset(self, stream_id: int) -> bool:
  1727. """
  1728. Returns ``True`` if the stream was closed by sending or receiving a
  1729. RST_STREAM frame. Returns ``False`` otherwise.
  1730. """
  1731. return self._stream_closed_by(stream_id) in (
  1732. StreamClosedBy.RECV_RST_STREAM, StreamClosedBy.SEND_RST_STREAM,
  1733. )
  1734. def _stream_is_closed_by_end(self, stream_id: int) -> bool:
  1735. """
  1736. Returns ``True`` if the stream was closed by sending or receiving an
  1737. END_STREAM flag in a HEADERS or DATA frame. Returns ``False``
  1738. otherwise.
  1739. """
  1740. return self._stream_closed_by(stream_id) in (
  1741. StreamClosedBy.RECV_END_STREAM, StreamClosedBy.SEND_END_STREAM,
  1742. )
  1743. def _add_frame_priority(frame: PriorityFrame | HeadersFrame,
  1744. weight: int | None = None,
  1745. depends_on: int | None = None,
  1746. exclusive: bool | None = None) -> PriorityFrame | HeadersFrame:
  1747. """
  1748. Adds priority data to a given frame. Does not change any flags set on that
  1749. frame: if the caller is adding priority information to a HEADERS frame they
  1750. must set that themselves.
  1751. This method also deliberately sets defaults for anything missing.
  1752. This method validates the input values.
  1753. """
  1754. # A stream may not depend on itself.
  1755. if depends_on == frame.stream_id:
  1756. msg = f"Stream {frame.stream_id} may not depend on itself"
  1757. raise ProtocolError(msg)
  1758. # Weight must be between 1 and 256.
  1759. if weight is not None:
  1760. if weight > 256 or weight < 1:
  1761. msg = f"Weight must be between 1 and 256, not {weight}"
  1762. raise ProtocolError(msg)
  1763. # Weight is an integer between 1 and 256, but the byte only allows
  1764. # 0 to 255: subtract one.
  1765. weight -= 1
  1766. # Set defaults for anything not provided.
  1767. weight = weight if weight is not None else 15
  1768. depends_on = depends_on if depends_on is not None else 0
  1769. exclusive = exclusive if exclusive is not None else False
  1770. frame.stream_weight = weight
  1771. frame.depends_on = depends_on
  1772. frame.exclusive = exclusive
  1773. return frame
  1774. def _decode_headers(decoder: Decoder, encoded_header_block: bytes) -> Iterable[Header]:
  1775. """
  1776. Decode a HPACK-encoded header block, translating HPACK exceptions into
  1777. sensible h2 errors.
  1778. This only ever returns bytestring headers: h2 may emit them as
  1779. unicode later, but internally it processes them as bytestrings only.
  1780. """
  1781. try:
  1782. return decoder.decode(encoded_header_block, raw=True)
  1783. except OversizedHeaderListError as e:
  1784. # This is a symptom of a HPACK bomb attack: the user has
  1785. # disregarded our requirements on how large a header block we'll
  1786. # accept.
  1787. msg = f"Oversized header block: {e}"
  1788. raise DenialOfServiceError(msg) from e
  1789. except (HPACKError, IndexError, TypeError, UnicodeDecodeError) as e:
  1790. # We should only need HPACKError here, but versions of HPACK older
  1791. # than 2.1.0 throw all three others as well. For maximum
  1792. # compatibility, catch all of them.
  1793. msg = f"Error decoding header block: {e}"
  1794. raise ProtocolError(msg) from e