Vendor http_sfv

Author: kislyuk

README.rst

@@ -16,8 +16,8 @@ Synopsis
 
 .. code-block:: python
 
-    from http_message_signatures import HTTPMessageSigner, HTTPMessageVerifier, HTTPSignatureKeyResolver, algorithms
-    import requests, base64, hashlib, http_sfv
+    from http_message_signatures import HTTPMessageSigner, HTTPMessageVerifier, HTTPSignatureKeyResolver, algorithms, http_sfv
+    import requests, base64, hashlib
 
     class MyHTTPSignatureKeyResolver(HTTPSignatureKeyResolver):
         keys = {"my-key": b"top-secret-key"}

http_message_signatures/http_sfv/__init__.py

@@ -0,0 +1,33 @@
+"""
+Structured HTTP Field Values
+
+Copyright (c) 2018-2020 Mark Nottingham
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+"""
+
+# Top-level structures
+from .dictionary import Dictionary
+from .item import InnerList, Item
+from .list import List
+
+# Item type wrappers
+from .types import DisplayString, Token
+
+structures = {"dictionary": Dictionary, "list": List, "item": Item}

http_message_signatures/http_sfv/boolean.py

@@ -0,0 +1,19 @@
+from typing import Tuple
+
+QUESTION = ord(b"?")
+ONE = ord(b"1")
+ZERO = ord(b"0")
+
+_boolean_map = {ONE: (2, True), ZERO: (2, False)}
+
+
+def parse_boolean(data: bytes) -> Tuple[int, bool]:
+    try:
+        return _boolean_map[data[1]]
+    except (KeyError, IndexError):
+        pass
+    raise ValueError("No Boolean value found")
+
+
+def ser_boolean(inval: bool) -> str:
+    return f"?{inval and '1' or '0'}"

http_message_signatures/http_sfv/byteseq.py

@@ -0,0 +1,28 @@
+import base64
+import binascii
+from string import ascii_letters, digits
+from typing import Tuple
+
+BYTE_DELIMIT = ord(b":")
+B64CONTENT = set((ascii_letters + digits + "+/=").encode("ascii"))
+
+
+def parse_byteseq(data: bytes) -> Tuple[int, bytes]:
+    bytes_consumed = 1
+    try:
+        end_delimit = data[bytes_consumed:].index(BYTE_DELIMIT)
+    except ValueError as why:
+        raise ValueError("Binary Sequence didn't contain ending ':'") from why
+    b64_content = data[bytes_consumed : bytes_consumed + end_delimit]
+    bytes_consumed += end_delimit + 1
+    if not all(c in B64CONTENT for c in b64_content):
+        raise ValueError("Binary Sequence contained disallowed character")
+    try:
+        binary_content = base64.standard_b64decode(b64_content)
+    except binascii.Error as why:
+        raise ValueError("Binary Sequence failed to decode") from why
+    return bytes_consumed, binary_content
+
+
+def ser_byteseq(byteseq: bytes) -> str:
+    return f":{base64.standard_b64encode(byteseq).decode('ascii')}:"

http_message_signatures/http_sfv/date.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from typing import Tuple
+
+from .integer import parse_integer
+
+
+def parse_date(data: bytes) -> Tuple[int, datetime]:
+    bytes_consumed, value = parse_integer(data[1:])
+    if not isinstance(value, int):
+        raise ValueError("Non-integer Date")
+    return bytes_consumed + 1, datetime.fromtimestamp(value)
+
+
+def ser_date(inval: datetime) -> str:
+    return f"@{int(inval.timestamp())}"

http_message_signatures/http_sfv/decimal.py

@@ -0,0 +1,29 @@
+from decimal import Decimal
+from typing import Tuple, Union
+
+from .integer import parse_number
+
+INT_DIGITS = 12
+FRAC_DIGITS = 3
+PRECISION = Decimal(10) ** -FRAC_DIGITS
+
+
+def parse_decimal(data: bytes) -> Tuple[int, Decimal]:
+    return parse_number(data)  # type: ignore
+
+
+def ser_decimal(input_decimal: Union[Decimal, float]) -> str:
+    if isinstance(input_decimal, float):
+        input_decimal = Decimal(input_decimal)
+    if not isinstance(input_decimal, Decimal):
+        raise ValueError("decimal input is not decimal")
+    input_decimal = round(input_decimal, FRAC_DIGITS)
+    abs_decimal = input_decimal.copy_abs()
+    integer_component_s = str(int(abs_decimal))
+    if len(integer_component_s) > INT_DIGITS:
+        raise ValueError(f"decimal with oversize integer component {integer_component_s}")
+    fractional_component = abs_decimal.quantize(PRECISION).normalize() % 1
+    return (
+        f"{'-' if input_decimal < 0 else ''}{integer_component_s}."
+        f"{str(fractional_component)[2:] if fractional_component else '0'}"
+    )

http_message_signatures/http_sfv/dictionary.py

@@ -0,0 +1,74 @@
+from collections import UserDict
+
+from .item import Item, InnerList, itemise, AllItemType
+from .list import parse_item_or_inner_list
+from .types import JsonDictType
+from .util import (
+    StructuredFieldValue,
+    discard_http_ows,
+    ser_key,
+    parse_key,
+)
+
+EQUALS = ord(b"=")
+COMMA = ord(b",")
+
+
+class Dictionary(UserDict, StructuredFieldValue):
+    def parse_content(self, data: bytes) -> int:
+        bytes_consumed = 0
+        data_len = len(data)
+        try:
+            while True:
+                offset, this_key = parse_key(data[bytes_consumed:])
+                bytes_consumed += offset
+                try:
+                    is_equals = data[bytes_consumed] == EQUALS
+                except IndexError:
+                    is_equals = False
+                if is_equals:
+                    bytes_consumed += 1  # consume the "="
+                    offset, member = parse_item_or_inner_list(data[bytes_consumed:])
+                    bytes_consumed += offset
+                else:
+                    member = Item()
+                    member.value = True
+                    bytes_consumed += member.params.parse(data[bytes_consumed:])
+                self[this_key] = member
+                bytes_consumed += discard_http_ows(data[bytes_consumed:])
+                if bytes_consumed == data_len:
+                    return bytes_consumed
+                if data[bytes_consumed] != COMMA:
+                    raise ValueError(f"Dictionary member '{this_key}' has trailing characters")
+                bytes_consumed += 1
+                bytes_consumed += discard_http_ows(data[bytes_consumed:])
+                if bytes_consumed == data_len:
+                    raise ValueError("Dictionary has trailing comma")
+        except Exception as why:
+            self.clear()
+            raise ValueError from why
+
+    def __setitem__(self, key: str, value: AllItemType) -> None:
+        self.data[key] = itemise(value)
+
+    def __str__(self) -> str:
+        if len(self) == 0:
+            raise ValueError("No contents; field should not be emitted")
+        return ", ".join(
+            [
+                f"{ser_key(m)}"
+                f"""{n.params if (isinstance(n, Item) and n.value is True) else f"={n}"}"""
+                for m, n in self.items()
+            ]
+        )
+
+    def to_json(self) -> JsonDictType:
+        return [(key, val.to_json()) for (key, val) in self.items()]
+
+    def from_json(self, json_data: JsonDictType) -> None:
+        for key, val in json_data:
+            if isinstance(val[0], list):
+                self[key] = InnerList()
+            else:
+                self[key] = Item()
+            self[key].from_json(val)

http_message_signatures/http_sfv/display_string.py

@@ -0,0 +1,53 @@
+from typing import Tuple
+
+from .types import DisplayString
+
+PERCENT = ord("%")
+DQUOTE = ord('"')
+
+
+def parse_display_string(data: bytes) -> Tuple[int, DisplayString]:
+    output_array = bytearray([])
+    if data[:2] != b'%"':
+        raise ValueError('Display string does not start with %"')
+    bytes_consumed = 2  # consume PERCENT DQUOTE
+    while True:
+        try:
+            char = data[bytes_consumed]
+        except IndexError as why:
+            raise ValueError("Reached end of input without finding a closing DQUOTE") from why
+        bytes_consumed += 1
+        if char == PERCENT:
+            try:
+                next_chars = data[bytes_consumed : bytes_consumed + 2]
+            except IndexError as why:
+                raise ValueError("Incomplete percent encoding") from why
+            bytes_consumed += 2
+            if next_chars.lower() != next_chars:
+                raise ValueError("Uppercase percent encoding")
+            try:
+                octet = int(next_chars, base=16)
+            except ValueError as why:
+                raise ValueError("Invalid percent encoding") from why
+            output_array.append(octet)
+        elif char == DQUOTE:
+            try:
+                output_string = output_array.decode("utf-8")
+            except UnicodeDecodeError as why:
+                raise ValueError("Invalid UTF-8") from why
+            return bytes_consumed, DisplayString(output_string)
+        elif 31 < char < 127:
+            output_array.append(char)
+        else:
+            raise ValueError("String contains disallowed character")
+
+
+def ser_display_string(inval: DisplayString) -> str:
+    byte_array = inval.encode("utf-8")
+    escaped = []
+    for byte in byte_array:
+        if byte in [PERCENT, DQUOTE] or not 31 <= byte <= 127:
+            escaped.append(f"%{byte:x}")
+        else:
+            escaped.append(chr(byte))
+    return DisplayString(f'%"{"".join(escaped)}"')

http_message_signatures/http_sfv/integer.py

@@ -0,0 +1,80 @@
+from decimal import Decimal
+from string import digits
+from typing import Tuple, Union
+
+
+MAX_INT = 999999999999999
+MIN_INT = -999999999999999
+
+DIGITS = set(digits.encode("ascii"))
+NUMBER_START_CHARS = set((digits + "-").encode("ascii"))
+PERIOD = ord(b".")
+MINUS = ord(b"-")
+
+
+def parse_integer(data: bytes) -> Tuple[int, int]:
+    return parse_number(data)  # type: ignore
+
+
+def ser_integer(inval: int) -> str:
+    if not MIN_INT <= inval <= MAX_INT:
+        raise ValueError("Input is out of Integer range.")
+    output = ""
+    if inval < 0:
+        output += "-"
+    output += str(abs(inval))
+    return output
+
+
+INTEGER = "integer"
+DECIMAL = "decimal"
+
+
+def parse_number(data: bytes) -> Tuple[int, Union[int, Decimal]]:
+    _type = INTEGER
+    _sign = 1
+    bytes_consumed = 0
+    num_start = 0
+    decimal_index = 0
+    num_length = 0
+    if data[0] == MINUS:
+        bytes_consumed += 1
+        num_start += 1
+        _sign = -1
+    if not data[bytes_consumed:]:
+        raise ValueError("Number input lacked a number")
+    if not data[num_start] in DIGITS:
+        raise ValueError("Number doesn't start with a DIGIT")
+    while True:
+        try:
+            char = data[bytes_consumed]
+        except IndexError:
+            break
+        bytes_consumed += 1
+        num_length = bytes_consumed - num_start - 1
+        if char in DIGITS:
+            pass
+        elif _type is INTEGER and char == PERIOD:
+            if num_length > 12:
+                raise ValueError("Decimal too long.")
+            _type = DECIMAL
+            decimal_index = bytes_consumed
+        else:
+            bytes_consumed -= 1
+            break
+    if _type == INTEGER:
+        if num_length > 15:
+            raise ValueError("Integer too long.")
+        output_int = int(data[num_start:bytes_consumed]) * _sign
+        if not MIN_INT <= output_int <= MAX_INT:
+            raise ValueError("Integer outside allowed range")
+        return bytes_consumed, output_int
+    # Decimal
+    if num_length > 16:
+        raise ValueError("Decimal too long.")
+    if data[bytes_consumed - 1] == MINUS:
+        raise ValueError("Decimal ends in '.'")
+    if bytes_consumed - decimal_index > 3:
+        raise ValueError("Decimal fractional component too long")
+    output_float = Decimal(data[num_start:bytes_consumed].decode("ascii")) * _sign
+    return bytes_consumed, output_float