# -*- coding: utf-8 -*-
r"""HTTP/1.* - Hypertext Transfer Protocol
============================================
.. module:: pcapkit.protocols.application.httpv1
:mod:`pcapkit.protocols.application.httpv1` contains
:class:`~pcapkit.protocols.application.httpv1.HTTP`
only, which implements extractor for Hypertext Transfer
Protocol (HTTP/1.*) [*]_, whose structure is described
as below:
.. code-block:: text
METHOD URL HTTP/VERSION\r\n :==: REQUEST LINE
<key> : <value>\r\n :==: REQUEST HEADER
............ (Ellipsis) :==: REQUEST HEADER
\r\n :==: REQUEST SEPARATOR
<body> :==: REQUEST BODY (optional)
HTTP/VERSION CODE DESP \r\n :==: RESPONSE LINE
<key> : <value>\r\n :==: RESPONSE HEADER
............ (Ellipsis) :==: RESPONSE HEADER
\r\n :==: RESPONSE SEPARATOR
<body> :==: RESPONSE BODY (optional)
.. [*] https://en.wikipedia.org/wiki/Hypertext_Transfer_Protocol
"""
import re
from typing import TYPE_CHECKING
from pcapkit.const.http.method import Method as Enum_Method
from pcapkit.const.http.status_code import StatusCode as Enum_StatusCode
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.protocols.application.http import HTTP as HTTPBase
from pcapkit.protocols.data.application.httpv1 import HTTP as Data_HTTP
from pcapkit.protocols.data.application.httpv1 import RequestHeader as Data_RequestHeader
from pcapkit.protocols.data.application.httpv1 import ResponseHeader as Data_ResponseHeader
from pcapkit.protocols.schema.application.httpv1 import HTTP as Schema_HTTP
from pcapkit.utilities.compat import StrEnum
from pcapkit.utilities.exceptions import ProtocolError
if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from typing import Any, Optional
from typing import Type as _Type
from aenum import IntEnum as AenumEnum
from typing_extensions import Literal
from pcapkit.protocols.data.application.httpv1 import Header as Data_Header
__all__ = ['HTTP']
# Regular expression to match HTTP methods. Anchored at both ends: :func:`re.match`
# anchors only at the start, so an unanchored pattern prefix-matches and accepts the
# leading ``G`` of ``Get`` as a whole method token. Method tokens are case-sensitive
# per :rfc:`9110#section-9.1`, so ``Get`` is not ``GET`` and must not parse as one.
_RE_METHOD = re.compile(rb"(?P<method>[A-Z][A-Z-]*)\Z") # RFC 9110, section 16.1.1, 9.1, 5.6.2
# Regular expression to match HTTP version string.
_RE_VERSION = re.compile(rb"HTTP/(?P<version>\d\.\d)")
# Regular expression to match HTTP status code. Anchored for the same reason as
# ``_RE_METHOD``, and it matters more here: this pattern is only a guard, and the
# value is taken from ``int(para2)`` on the raw token, so an unanchored prefix
# match let ``200x`` and ``2000`` past the guard and then out of ``int()`` as a
# bare ``ValueError`` -- where ``_read_http_header`` documents ``ProtocolError``.
# :rfc:`9112#section-4` gives ``status-code = 3DIGIT``, exactly three -- the
# grammar is in HTTP/1.1 because ``status-code`` is part of its ``status-line``
# production; :rfc:`9110#section-15` covers the code semantics and registry, not
# the syntax.
_RE_STATUS = re.compile(rb'\d{3}\Z')
[docs]
def _test_start_line(data: 'bytes') -> 'bool':
"""Whether ``data`` opens with an HTTP/1.* start line.
This is a *classification* predicate and parses nothing: it answers "is this
HTTP/1?" for :meth:`HTTP._guess_version
<pcapkit.protocols.application.http.HTTP._guess_version>`, which until #800
answered that question by trial-parsing every version in the family and
keeping whichever one did not object -- so a payload that is not HTTP at all
was classified by which parser happened to fail less loudly.
Args:
data: Payload to classify.
Returns:
Whether the payload's first line is a ``request-line`` or a
``status-line`` (:rfc:`9112#section-2.1`).
Note:
The acceptance rule is deliberately the *same* one
:meth:`HTTP._read_http_header
<pcapkit.protocols.application.httpv1.HTTP._read_http_header>` applies
further down this module -- ``_RE_METHOD`` with ``_RE_VERSION`` for a
request, ``_RE_VERSION`` with ``_RE_STATUS`` for a response -- which is
why this lives beside those three patterns rather than in the dispatcher
that calls it. The two must accept the same start lines: a predicate
looser than the parser classifies payloads the parser then refuses, and
one tighter than the parser hands real HTTP/1 to a later arm.
``test_start_line_predicate_agrees_with_the_httpv1_parser`` pins that
agreement.
Both of the unpackings the parser performs *before* those patterns are
mirrored too, and this is not pedantry -- the second of them is the whole
reason the HTTP/2 connection preface is not claimed here. ``PRI *
HTTP/2.0\\r\\n\\r\\nSM\\r\\n\\r\\n`` is deliberately a well-formed
HTTP/1.1 *request line* (:rfc:`9113#section-3.4`), so a predicate that
tested only the first line would answer :data:`True` for it. Split at the
header/body separator first, as :meth:`HTTP.read
<pcapkit.protocols.application.httpv1.HTTP.read>` does, and the preface's
header is ``PRI * HTTP/2.0`` with no CRLF left in it -- which is exactly
why the parser refuses it, and now why this does. Measured: without the
separator split this returned :data:`True` for the preface.
An HTTP/0.9 request line carries only two tokens and so is not recognised
here either, matching the parser, which raises on fewer than three.
"""
header = data.split(b'\r\n\r\n', maxsplit=1)[0]
if header == data: # no header/body separator -- ``read`` raises
return False
startline = header.split(b'\r\n', maxsplit=1)[0]
if startline == header: # header holds no CRLF -- ``_read_http_header`` raises
return False
try:
para1, para2, para3 = re.split(rb'\s+', startline, maxsplit=2)
except ValueError:
return False
return bool(
(re.match(_RE_METHOD, para1) and re.match(_RE_VERSION, para3)) # request-line
or (re.match(_RE_VERSION, para1) and re.match(_RE_STATUS, para2)) # status-line
)
[docs]
class Type(StrEnum):
"""HTTP packet type."""
#: Request packet.
REQUEST = 'request'
#: Response packet.
RESPONSE = 'response'
[docs]
class HTTP(HTTPBase[Data_HTTP, Schema_HTTP],
data=Data_HTTP, schema=Schema_HTTP):
"""This class implements Hypertext Transfer Protocol (HTTP/1.*)."""
##########################################################################
# Defaults.
##########################################################################
#: Type: Type of HTTP receipt.
_receipt: 'Type'
##########################################################################
# Properties.
##########################################################################
@property
def alias(self) -> 'Literal["HTTP/0.9", "HTTP/1.0", "HTTP/1.1"]':
"""Acronym of current protocol."""
return f'HTTP/{self.version}' # type: ignore[return-value]
@property
def version(self) -> 'Literal["0.9", "1.0", "1.1"]':
"""Version of current protocol."""
return self._info.receipt.version # type: ignore[attr-defined]
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_HTTP': # pylint: disable=unused-argument
"""Read Hypertext Transfer Protocol (HTTP/1.*).
Structure of HTTP/1.* packet [:rfc:`7230`]:
.. code-block:: text
HTTP-message :==: start-line
*( header-field CRLF )
CRLF
[ message-body ]
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Raises:
ProtocolError: If the packet is malformed.
"""
if length is None:
length = len(self)
schema = self.__header__
packet = schema.data
# NOTE: A payload carrying no header/body separator at all unpacks short
# here, and the bare ``ValueError`` that used to escape is what made
# ``HTTP._guess_version``'s HTTP/2 arm unreachable: that dispatcher falls
# through on ``ProtocolError`` alone, so an HTTP/1 attempt on HTTP/2 wire
# bytes aborted the guess rather than failing it, and the HTTP/2 attempt
# never ran (#787). ``ProtocolError`` is what the ``Raises:`` section
# above already promises for a malformed packet, and the same conversion
# the explicit ``version=`` path performs at ``http.py:119``; chained, so
# the underlying unpacking error stays reachable as ``__cause__``.
try:
header, body = packet.split(b'\r\n\r\n', maxsplit=1)
except ValueError as error:
raise ProtocolError('HTTP: invalid format') from error
header_line, header_unpacked = self._read_http_header(header)
body_unpacked = self._read_http_body(body, headers=header_unpacked) or None
http = Data_HTTP(
receipt=header_line,
header=header_unpacked,
body=body_unpacked,
)
self._receipt = header_line.type
self._version = header_line.version # type: ignore[attr-defined]
self._length = len(header)
return http
[docs]
def make(self, # type: ignore[override]
http_version: 'Literal["0.9", "1.0", "1.1", b"0.9", b"1.0", b"1.1"]' = '1.1',
method: 'Optional[Enum_Method | str | bytes]' = None,
uri: 'Optional[str | bytes]' = None,
status: 'Optional[Enum_StatusCode | str | bytes | int]' = None,
status_default: 'Optional[int]' = None,
status_namespace: 'Optional[dict[str, int] | dict[int, str] | _Type[StdlibEnum] | _Type[AenumEnum]]' = None, # pylint: disable=line-too-long
status_reversed: 'bool' = False,
message: 'Optional[str | bytes]' = None,
headers: 'Optional[OrderedMultiDict[str, str]]' = None,
body: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_HTTP':
"""Make (construct) packet data.
Args:
http_version: HTTP version.
method: HTTP method.
uri: HTTP request URI.
status: HTTP status code.
status_default: Default HTTP status code.
status_namespace: Namespace of HTTP status code.
status_reversed: Whether to reverse the namespace.
message: HTTP status message.
headers: HTTP headers.
body: HTTP body.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
"""
version = http_version.encode() if isinstance(http_version, str) else http_version
if method is not None and status is None:
if uri is None:
raise ProtocolError('HTTP request must have URI.')
if isinstance(method, Enum_Method):
meth = method.value.encode()
elif isinstance(method, bytes):
meth = method
elif isinstance(method, str):
meth = method.encode()
else:
meth = method.value.encode()
uri_val = uri.encode() if isinstance(uri, str) else uri
header_line = b'%s %s HTTP/%s\r\n' % (meth, uri_val, version)
elif method is None and status is not None:
status_code = self._make_index(status, status_default, namespace=status_namespace,
reversed=status_reversed, pack=False)
status_code_val = int(status_code)
if message is None:
msg = getattr(status, 'message', None) or getattr(status_code, 'message', b'') or b''
else:
msg = message.encode() if isinstance(message, str) else message
if isinstance(msg, str):
msg = msg.encode()
header_line = b'HTTP/%s %s %s\r\n' % (version, str(status_code_val).encode(), msg)
else:
raise ProtocolError('HTTP packet must be either request or response.')
header_fields = [] # type: list[bytes]
if headers is not None:
header_fields = []
for key, value in headers.items(multi=True):
header_fields.append(b'%s: %s\r\n' % (key.encode(), value.encode()))
return Schema_HTTP(
data=header_line + b''.join(header_fields) + b'\r\n' + body,
)
[docs]
@classmethod
def id(cls) -> 'tuple[Literal["HTTP"], Literal["HTTPv1"]]': # type: ignore[override]
"""Index ID of the protocol.
Returns:
Index ID of the protocol.
"""
return (cls.__name__, 'HTTPv1') # type: ignore[return-value]
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_HTTP') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
return {
'http_version': data.receipt.version, # type: ignore[attr-defined]
'method': getattr(data.receipt, 'method', None),
'uri': getattr(data.receipt, 'uri', None),
'status': getattr(data.receipt, 'status', None),
'message': getattr(data.receipt, 'message', None),
'headers': data.header,
'body': data.body,
}
[docs]
def _read_http_body(self, body: 'bytes', *,
headers: 'OrderedMultiDict[str, str]') -> 'Any':
"""Read HTTP/1.* body.
Args:
body: HTTP body data.
headers: HTTP header fields.
Returns:
Raw HTTP body.
"""
return body