# Copyright (c) 2023-2024 Jan Malakhovski <oxij@oxij.org>
#
# This file is a part of `hoardy-web` project.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Parsing and un-parsing of URLs, HTTP headers, HTML attributes, etc."""
import base64 as _base64
import dataclasses as _dc
import logging as _logging
import os as _os
import re as _re
import typing as _t
import urllib.parse as _up
from functools import cached_property
import idna as _idna
from kisstdlib.failure import *
from kisstdlib.parsing import *
from kisstdlib.string_ext import quoter, safe_char
### URL parsing
# URL double-slash scheme
uds_str = r"http|https|ftp|ftps"
# URL auth
uauth_str = r"\S+(?::\S*)?"
# URL hostname
def uhostname_str(d: str) -> str:
return rf"[^:/?#\s{d}]+"
# URL port
uport_str = r":\d+"
# URL path
def upath_str(d: str) -> str:
return rf"/[^?#\s{d}]*"
# URL relative path
def urel_str(d: str) -> str:
return rf"[^?#\s{d}]+"
# URL query
def uquery_str(d: str) -> str:
return rf"[^#\s{d}]*"
# URL fragment/hash
def ufragment_str(d: str) -> str:
return rf"[^\s{d}]*"
# Regexp describing a URL, the second case is for common malformed path-less
# URLs like "https://example.com" (which should actually be given as
# "https://example.com/", but this is almost a universal mistake). The order
# of cases matters, since `_re.match` and co are greedy.
def url_re_str(d: str) -> str:
return rf"""(
(
(?:(?:{uds_str}):)?
//
(?:{uauth_str}@)?
(?:{uhostname_str(d)})
(?:{uport_str})?
(?:{upath_str(d)})
|
(?:(?:{uds_str}):)?
//
(?:{uauth_str}@)?
(?:{uhostname_str(d)})
(?:{uport_str})?
|
(?:{upath_str(d)})
|
(?:{urel_str(d)})
)
(\?{uquery_str(d)})?
(#{ufragment_str(d)})?
)""".replace(
"\n", ""
)
# print(url_re_str(""))
url_re = _re.compile(url_re_str(""))
def parse_path(path: str, encoding: str = "utf-8", errors: str = "replace") -> list[str]:
if path in ("", "/"):
return []
return [_up.unquote(e, encoding=encoding, errors=errors) for e in path[1:].split("/")]
def unparse_path(
path_parts: _t.Sequence[str], encoding: str = "utf-8", errors: str = "strict"
) -> str:
return "/".join(map(lambda p: _up.quote(p, ":=", encoding=encoding, errors=errors), path_parts))
def parse_query(
query: str, encoding: str = "utf-8", errors: str = "replace"
) -> list[tuple[str, str | None]]:
"""Like `urllib.parse.parse_qsl`, but "a=" still parses into ("a", "") while
just "a" parses into ("a", None). I.e. this keeps information about
whether "=" was present.
"""
parts = [s2 for s1 in query.split("&") for s2 in s1.split(";")]
res: list[tuple[str, str | None]] = []
for e in parts:
eqp = e.find("=")
if eqp == -1:
name = _up.unquote_plus(e, encoding=encoding, errors=errors)
res.append((name, None))
continue
n, v = e[:eqp], e[eqp + 1 :]
name = _up.unquote_plus(n, encoding=encoding, errors=errors)
value = _up.unquote_plus(v, encoding=encoding, errors=errors)
res.append((name, value))
return res
def unparse_query(
qsl: _t.Sequence[tuple[str, str | None]], encoding: str = "utf-8", errors: str = "strict"
) -> str:
"""Like `urllib.parse.urlencode`, turns URL query components list back into a
query, except works with our `parse_query`.
"""
l = []
for k, v in qsl:
k = _up.quote_plus(k, encoding=encoding, errors=errors)
if v is None:
l.append(k)
else:
v = _up.quote_plus(v, encoding=encoding, errors=errors)
l.append(k + "=" + v)
return "&".join(l)
quote_path = quoter("%", "", safe_char("", "/?#")).escape
quote_key = quoter("%", "", safe_char("", "/&=#")).escape
quote_value = quoter("%", "", safe_char("", "/&#")).escape
def pp_to_path(parts: list[str]) -> str:
"""Turn URL path components list into a minimally-quoted path."""
return "/".join([quote_path(e) for e in parts])
def qsl_to_path(qsl: _t.Sequence[tuple[str, str | None]]) -> str:
"""Turn URL query components list into a minimally-quoted path."""
l = []
for k, v in qsl:
k = quote_key(k)
if v is None:
l.append(k)
else:
v = quote_value(v)
l.append(k + "=" + v)
return "&".join(l)
@_dc.dataclass
class ParsedURL:
raw_url: str
scheme: str
user: str
password: str
brackets: bool
raw_hostname: str
net_hostname: str
hostname: str
opm: str
port: str
raw_path: str
oqm: str
raw_query: str
ofm: str
fragment: str
@cached_property
def net_auth(self) -> str:
if self.user != "":
if self.password != "":
return f"{self.user}:{self.password}@"
return f"{self.user}@"
return ""
@cached_property
def rhostname(self) -> str:
hparts = self.hostname.split(".")
hparts.reverse()
return ".".join(hparts)
@cached_property
def netloc(self) -> str:
hn = self.hostname
if self.brackets:
hn = "[" + hn + "]"
return "".join([self.net_auth, hn, self.opm, self.port])
@cached_property
def net_netloc(self) -> str:
hn = self.net_hostname
if self.brackets:
hn = "[" + hn + "]"
return "".join([self.net_auth, hn, self.opm, self.port])
@cached_property
def path_parts(self) -> list[str]:
return parse_path(self.raw_path)
@cached_property
def path(self) -> str:
return unparse_path(self.path_parts)
@cached_property
def query_parts(self) -> list[tuple[str, str | None]]:
return parse_query(self.raw_query)
@cached_property
def query(self) -> str:
return unparse_query(self.query_parts)
@cached_property
def net_url(self) -> str:
if self.raw_hostname:
nl = self.net_netloc
if nl != "":
nl = "//" + nl
return _up.quote(
f"{self.scheme}:{nl}/{self.path}{self.oqm}{self.query}",
"%/:=&?~#+!$,;'@()*[]|",
)
return _up.quote(
f"{self.scheme}:{self.path}{self.oqm}{self.query}", "%/:=&?~#+!$,;'@()*[]|"
)
@cached_property
def url(self) -> str:
return f"{self.net_url}{self.ofm}{self.fragment}"
@cached_property
def npath_parts(self) -> list[str]:
parts: list[str] = []
last_es = False
for e in self.path_parts:
last_es = False
if e == "":
last_es = True
continue
# remove dots, interpret double dots
if e == ".":
continue
if e == "..":
if len(parts) > 0:
parts.pop()
continue
parts.append(e)
# re-add the last "" so that things like "/test/" would work as expected
if last_es:
parts.append("")
return parts
def filepath_parts_ext(self, default: str, extensions: list[str]) -> tuple[list[str], str]:
parts = self.npath_parts
if len(parts) == 0:
return parts + [default], extensions[0] if len(extensions) > 0 else ".data"
if parts[len(parts) - 1] == "":
return parts[:-1] + [default], extensions[0] if len(extensions) > 0 else ".data"
last = parts[-1].lower()
last_name, last_ext = _os.path.splitext(last)
if last_ext == "":
return parts + [default], extensions[0] if len(extensions) > 0 else ".data"
if last_ext in extensions:
return parts[:-1] + [last_name], last_ext
if len(extensions) > 0:
return parts[:-1] + [last], extensions[0]
if last_ext == ".data":
return parts[:-1] + [last_name], ".data"
return parts[:-1] + [last], ".data"
@cached_property
def query_nparts(self) -> list[tuple[str, str]]:
res: list[tuple[str, str]] = []
for e in self.query_parts:
x = e[1]
if x is not None and x != "":
res.append(e) # type: ignore
return res
@cached_property
def mq_path(self) -> str:
return pp_to_path(self.path_parts)
@cached_property
def mq_npath(self) -> str:
return pp_to_path(self.npath_parts)
@cached_property
def mq_query(self) -> str:
return qsl_to_path(self.query_parts)
@cached_property
def mq_nquery(self) -> str:
return qsl_to_path(self.query_nparts)
@cached_property
def pretty_net_url(self) -> str:
if self.raw_hostname:
nl = self.netloc
if nl != "":
nl = "//" + nl
return f"{self.scheme}:{nl}/{self.mq_path}{self.oqm}{self.mq_query}"
return f"{self.scheme}:{self.mq_path}{self.oqm}{self.mq_query}"
@cached_property
def pretty_url(self) -> str:
return f"{self.pretty_net_url}{self.ofm}{self.fragment}"
@cached_property
def pretty_net_nurl(self) -> str:
if self.raw_hostname:
nl = self.netloc
if nl != "":
nl = "//" + nl
return f"{self.scheme}:{nl}/{self.mq_npath}{self.oqm}{self.mq_nquery}"
return f"{self.scheme}:{self.mq_npath}{self.oqm}{self.mq_nquery}"
@cached_property
def pretty_nurl(self) -> str:
return f"{self.pretty_net_nurl}{self.ofm}{self.fragment}"
class URLParsingError(ValueError):
pass
def parse_url(url: str) -> ParsedURL:
try:
scheme, netloc, path, query, fragment = _up.urlsplit(url)
except Exception as exc:
raise URLParsingError(url) from exc
userinfo, has_user, hostinfo = netloc.rpartition("@")
if has_user:
user, _, password = userinfo.partition(":")
user = _up.quote(_up.unquote(user), "")
password = _up.quote(_up.unquote(password), "")
else:
user = ""
password = ""
if hostinfo.startswith("["):
brackets = True
raw_hostname, has_endbracket, port = hostinfo[1:].partition("]")
if not has_endbracket or port != "" and not port.startswith(":"):
raise URLParsingError(url)
opm = ":"
port = port[1:]
else:
brackets = False
raw_hostname, opm, port = hostinfo.partition(":")
if raw_hostname == "":
net_hostname = hostname = ""
else:
# Fix common issues by rewriting hostnames like browsers do
ehostname = _up.unquote(raw_hostname).strip().replace("_", "-")
# Yes, this is a bit weird. `_idna.encode` and `_idna.decode` are not bijective.
# So, we turn `raw_hostname` into unicode `str` first.
try:
dehostname = _idna.decode(ehostname, uts46=True)
except _idna.IDNAError as exc:
if ehostname[2:4] == "--":
_logging.warning(
"`parse_url` left `net_hostname` and related attrs of `%s` undecoded because `idna` module failed to decode `%s`: %s",
url,
ehostname,
repr(exc),
)
net_hostname = hostname = ehostname
else:
raise URLParsingError(url) from exc
else:
try:
# Then encode it with uts46 enabled.
net_hostname = _idna.encode(dehostname, uts46=True).decode("ascii")
# And then decode it again to get the canonical unicode hostname for which
# encoding and decoding will be bijective
hostname = _idna.decode(net_hostname)
except _idna.IDNAError as exc:
raise URLParsingError(url) from exc
oqm = "?" if query != "" or (query == "" and url.endswith("?")) else ""
ofm = "#" if fragment != "" or (fragment == "" and url.endswith("#")) else ""
return ParsedURL(
url,
scheme,
user,
password,
brackets,
raw_hostname,
net_hostname,
hostname,
opm,
port,
path,
oqm,
query,
ofm,
fragment,
)
def test_parse_url() -> None:
def check(x: ParsedURL, attr: str, expected: _t.Any) -> None:
# print(attr, getattr(x, attr), expected)
assert getattr(x, attr) == expected
example_org = (
"http://example.org/",
"http://example.org/",
"http://example.org/",
"http://example.org/",
"http://example.org/",
"http://example.org/",
)
example_org_hash = (
"http://example.org/",
"http://example.org/#hash",
"http://example.org/#hash",
"http://example.org/",
"http://example.org/#hash",
"http://example.org/",
)
tests1: list[tuple[str, str, str, str, str, str, str]]
tests1 = [
("http://example.org", *example_org),
("http://example.org/", *example_org),
#
# with fragments
("http://example.org#hash", *example_org_hash),
("http://example.org/#hash", *example_org_hash),
#
# workarounds for common typos silently worked-around by web browsers
("http://%20example.org", *example_org),
("http://%20example.org/", *example_org),
#
# with fragments and quoting
(
"http://example.org/one+two#hash",
"http://example.org/one%2Btwo",
"http://example.org/one%2Btwo#hash",
"http://example.org/one+two#hash",
"http://example.org/one+two",
"http://example.org/one+two#hash",
"http://example.org/one+two",
),
#
# web.archive.org-alikes
(
"http://example.org/web/2/http://archived.example.org/one+two#hash",
"http://example.org/web/2/http://archived.example.org/one%2Btwo",
"http://example.org/web/2/http://archived.example.org/one%2Btwo#hash",
"http://example.org/web/2/http://archived.example.org/one+two#hash",
"http://example.org/web/2/http://archived.example.org/one+two",
"http://example.org/web/2/http:/archived.example.org/one+two#hash",
"http://example.org/web/2/http:/archived.example.org/one+two",
),
]
for raw_url, net_url, url, purl, pnet_url, pnurl, pnet_nurl in tests1:
x = parse_url(raw_url)
check(x, "raw_url", raw_url)
check(x, "net_url", net_url)
check(x, "url", url)
check(x, "pretty_net_url", pnet_url)
check(x, "pretty_url", purl)
check(x, "pretty_net_nurl", pnet_nurl)
check(x, "pretty_nurl", pnurl)
tests2: list[tuple[str, str | None]]
tests2 = [
("http://example.org/", None),
("http://example.org/test", None),
("http://example.org/test/", None),
("http://example.org/unfinished/query?", None),
#
("http://example.org/unfinished/query?param", "http://example.org/unfinished/query?"),
#
("http://example.org/unfinished/query?param=", "http://example.org/unfinished/query?"),
#
("http://example.org/unfinished/query?param=0", None),
("http://example.org/unfinished/query?param=0¶m=1", None),
#
(
"http://example.org/web/2/https://archived.example.org",
"http://example.org/web/2/https:/archived.example.org",
),
#
(
"http://example.org/web/2/https://archived.example.org/",
"http://example.org/web/2/https:/archived.example.org/",
),
#
(
"http://example.org/web/2/https://archived.example.org/test",
"http://example.org/web/2/https:/archived.example.org/test",
),
#
(
"http://example.org/web/2/https://archived.example.org/test/",
"http://example.org/web/2/https:/archived.example.org/test/",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?param",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?param=",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?param=0",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?param=0",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?param=0¶m=",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?param=0",
),
#
(
"http://example.org/web/2/https://archived.example.org/unfinished/query?param=0¶m=1",
"http://example.org/web/2/https:/archived.example.org/unfinished/query?param=0¶m=1",
),
#
# work-arounds for hostnames that `idna` module fails to parse
("http://ab-cd-xxxxxxxxx-yyyy.example.org/", None),
("http://ab--cd-xxxxxxxxx-yyyy.example.org/", None),
("http://ab---cd-xxxxxxxxx-yyyy.example.org/", None),
]
for url, nurl in tests2:
x = parse_url(url)
check(x, "raw_url", url)
check(x, "net_url", url)
check(x, "url", url)
check(x, "pretty_net_url", url)
check(x, "pretty_url", url)
nurl = url if nurl is None else nurl
check(x, "pretty_net_nurl", nurl)
check(x, "pretty_nurl", nurl)
### MIME valuess
Parameters = list[tuple[str, str]]
ParamValueDefaultType = _t.TypeVar("ParamValueDefaultType", str, None)
def get_parameter_value(
ps: Parameters, name: str, default: ParamValueDefaultType
) -> str | ParamValueDefaultType:
for n, v in ps:
if n == name:
return v
return default
def set_parameter(ps: Parameters, name: str, value: str) -> Parameters:
res = []
not_done = True
for n, v in ps:
if n == name:
if not_done:
res.append((name, value))
not_done = False
continue
res.append((n, v))
if not_done:
res.append((name, value))
return res
token_ends = r'\s\t()\[\]<>@,:;\/?="'
token_body_re = _re.compile(rf"([^{token_ends}]+)")
def parse_token(p: Parser) -> str:
return _t.cast(str, p.lexeme_regex(token_body_re)[0])
def parse_mime_type(p: Parser, ends: list[str] = []) -> str:
ends = [";"] + ends
try:
maintype = parse_token(p)
p.string("/")
subtype = parse_token(p)
p.skip_optional_whitespace()
return maintype + "/" + subtype
except ParsingFailure:
p.take_until_at_string_in(ends)
# RFC says invalid content types are to be interpreted as `text/plain`
return "text/plain"
attribute_ends = token_ends + "*'%"
attribute_body_re = _re.compile(rf"([^{attribute_ends}]+)")
def parse_attribute(p: Parser) -> str:
return _t.cast(str, p.lexeme_regex(attribute_body_re)[0])
extended_attribute_ends = token_ends + "*'"
extended_attribute_body_re = _re.compile(rf"([^{extended_attribute_ends}]+)")
def parse_extended_attribute(p: Parser) -> str:
return _t.cast(str, p.lexeme_regex(extended_attribute_body_re)[0])
qcontent_body_re = _re.compile(r'([^"\\]*)')
def qcontent_escape(x: str) -> str:
res = []
for c in x:
if c in '\\"':
res.append("\\" + c)
else:
res.append(c)
return "".join(res)
def parse_value(p: Parser, ends: list[str]) -> str:
ws = p.optional_whitespace()
if p.at_eof() or p.at_string_in(ends):
raise ParsingFailure("expected attribute value, got %s", repr(ws[0]))
try:
p.string('"')
except ParsingFailure:
token = parse_extended_attribute(p)
else:
res = []
while not p.at_string('"'):
if p.at_string("\\"):
p.advance(1)
res.append(p.take(1))
else:
grp = p.regex(qcontent_body_re)
res.append(_t.cast(str, grp[0]))
p.string('"')
p.skip_optional_whitespace()
token = "".join(res)
return ws + token
def parse_parameter(p: Parser, ends: list[str]) -> tuple[str, str]:
key = parse_attribute(p)
try:
p.string("=")
except ParsingFailure:
value = ""
else:
value = parse_value(p, ends)
return key, value
def parse_invalid_parameter(p: Parser, ends: list[str]) -> tuple[str, str]:
key = p.take_until_at_string_in(ends)
return key.rstrip(), ""
def parse_mime_parameters(p: Parser, ends: list[str] = []) -> Parameters:
ends = [";"] + ends
res = []
while p.at_string(";"):
p.advance(1)
p.skip_optional_whitespace()
if p.at_eof() or p.at_string_in(ends):
# empty parameter
continue
save = p.pos
try:
token = parse_parameter(p, ends)
except ParsingFailure:
p.pos = save
token = parse_invalid_parameter(p, ends)
res.append(token)
return res
def unparse_mime_parameters(params: Parameters) -> str:
return "".join(map(lambda v: "; " + v[0] + '="' + v[1].replace('"', '\\"') + '"', params))
### `data:` URLs
def parse_data_url(value: str) -> tuple[str, Parameters, bytes]:
p = Parser(value)
p.string("data:")
if p.at_string(","):
# MDN says:
# > If omitted, defaults to `text/plain;charset=US-ASCII`
mime_type = "text/plain"
params = [("charset", "US-ASCII")]
else:
mime_type = parse_mime_type(p, [","])
params = parse_mime_parameters(p, [","])
p.string(",")
base64 = False
if len(params) > 0 and params[-1][0] == "base64":
params = params[:-1]
base64 = True
data: str | bytes
if base64:
data = _base64.b64decode(p.leftovers())
else:
data = _up.unquote_to_bytes(p.leftovers())
return mime_type, params, data
def unparse_data_url(mime_type: str, params: Parameters, data: bytes) -> str:
res = [
"data:",
mime_type,
]
for n, v in params:
res.append(";")
res.append(n)
if len(v) == 0:
continue
res.append('="')
res.append(qcontent_escape(v))
res.append('"')
res += [
";base64,",
_base64.b64encode(data).decode("ascii"),
]
return "".join(res)
def test_parse_data_url() -> None:
def check(xs: str | list[str], expected: tuple[str, Parameters, bytes]) -> None:
if isinstance(xs, str):
xs = [xs]
for x in xs:
res = parse_data_url(x)
assert res == expected
check(
"data:,Hello%2C%20World%21",
("text/plain", [("charset", "US-ASCII")], b"Hello, World!"),
)
check(
"data:text/plain,Hello%2C%20World%21",
("text/plain", [], b"Hello, World!"),
)
check(
[
"data:text/plain;base64,SGVsbG8sIFdvcmxkIQ==",
"data:text/plain; base64,SGVsbG8sIFdvcmxkIQ==",
"data:text/plain; base64 ,SGVsbG8sIFdvcmxkIQ==",
],
("text/plain", [], b"Hello, World!"),
)
check(
[
"data:text/plain;charset=UTF-8;base64,SGVsbG8sIFdvcmxkIQ==",
"data:text/plain; charset=UTF-8;base64,SGVsbG8sIFdvcmxkIQ==",
"data:text/plain; charset=UTF-8 ;base64,SGVsbG8sIFdvcmxkIQ==",
],
("text/plain", [("charset", "UTF-8")], b"Hello, World!"),
)
# because RFC says invalid content types are to be interpreted as `text/plain`
check(
[
"data: ,Hello%2C%20World%21",
"data:bla,Hello%2C%20World%21",
],
("text/plain", [], b"Hello, World!"),
)
def test_unparse_data_url() -> None:
def check(x: tuple[str, Parameters, bytes], expected: str) -> None:
res = unparse_data_url(*x)
assert res == expected
back = parse_data_url(res)
assert back == x
check(
("text/plain", [], b""),
"data:text/plain;base64,",
)
check(
("text/html", [], b"NYA"),
"data:text/html;base64,TllB",
)
check(
("text/plain", [("charset", "utf-8")], b"ABC"),
'data:text/plain;charset="utf-8";base64,QUJD',
)
check(
("text/plain", [("charset", "US-ASCII"), ("token", '"'), ("token", "'A")], b"data"),
'data:text/plain;charset="US-ASCII";token="\\"";token="\'A";base64,ZGF0YQ==',
)
### HTTP Headers
Headers = list[tuple[str, bytes]]
def get_raw_headers(headers: Headers) -> list[tuple[str, bytes]]:
# split because browsers frequently squish headers together
return [(k, e) for k, v in headers for e in v.split(b"\n")]
def get_headers(headers: Headers) -> list[tuple[str, str]]:
return [(k, v.decode("ascii")) for k, v in get_raw_headers(headers)]
def get_raw_headers_bytes(headers: Headers) -> list[bytes]:
return [k.encode("ascii") + b": " + v for k, v in get_raw_headers(headers)]
def get_raw_header_values(headers: Headers, name: str) -> list[bytes]:
# similarly
return [e for k, v in headers if k.lower() == name for e in v.split(b"\n")]
def get_header_values(headers: Headers, name: str) -> list[str]:
return [v.decode("ascii") for v in get_raw_header_values(headers, name)]
HeaderValueDefaultType = _t.TypeVar("HeaderValueDefaultType", str, None)
def get_header_value(
headers: Headers, name: str, default: HeaderValueDefaultType
) -> str | HeaderValueDefaultType:
res = get_header_values(headers, name)
if len(res) == 0:
return default
return res[0]
def parse_content_type_header(value: str) -> tuple[str, Parameters]:
"""Parse HTTP `Content-Type` header."""
p = Parser(value)
mime_type = parse_mime_type(p)
params = parse_mime_parameters(p)
p.skip_optional_whitespace()
p.eof()
return mime_type, params
def test_parse_content_type_header() -> None:
def check(xs: str | list[str], expected: tuple[str, Parameters]) -> None:
if isinstance(xs, str):
xs = [xs]
for x in xs:
res = parse_content_type_header(x)
assert res == expected
check(
["text/plain"],
("text/plain", []),
)
check(
["text/html"],
("text/html", []),
)
check(
[
"text/html;charset=utf-8",
"text/html ;charset=utf-8",
"text/html; charset=utf-8",
'text/html; charset="utf-8"',
'text/html; charset="utf-8";',
'text/html;; charset="utf-8";; ',
'text/html; ; charset="utf-8"; ; ',
],
("text/html", [("charset", "utf-8")]),
)
check(
[
"text/html;charset=utf-8;lang=en",
"text/html; charset=utf-8; lang=en",
'text/html; charset="utf-8"; lang=en',
'text/html; charset="utf-8"; lang=en;',
'text/html;; charset="utf-8"; lang=en;; ',
'text/html; ; charset="utf-8"; lang=en; ; ',
],
("text/html", [("charset", "utf-8"), ("lang", "en")]),
)
check(
[
'text/html;charset="\\"utf-8\\"";lang=en',
'text/html; charset="\\"utf-8\\""; lang=en',
],
("text/html", [("charset", '"utf-8"'), ("lang", "en")]),
)
check(
[
'text/html; charset="utf-8"; %%; lang=en',
'text/html; charset="utf-8"; %% ; lang=en',
],
("text/html", [("charset", "utf-8"), ("%%", ""), ("lang", "en")]),
)
# because RFC says invalid content types are to be interpreted as `text/plain`
check(
["bla; charset=utf-8; lang=en"],
("text/plain", [("charset", "utf-8"), ("lang", "en")]),
)
link_url_re = _re.compile(url_re_str("<>"))
def parse_link_value(p: Parser) -> tuple[str, Parameters]:
"""Parse single sub-value of HTTP `Link` header."""
p.string("<")
p.skip_optional_whitespace()
grp = p.regex(link_url_re)
p.skip_optional_whitespace()
p.string(">")
p.skip_optional_whitespace()
params = parse_mime_parameters(p, [","])
return _t.cast(str, grp[0]), params
type ParsedLinkHeader = list[tuple[str, Parameters]]
def parse_link_header(value: str) -> ParsedLinkHeader:
"""Parse HTTP `Link` header."""
p = Parser(value)
res = []
p.skip_optional_whitespace()
token = parse_link_value(p)
res.append(token)
while p.at_string(","):
p.advance(1)
p.skip_optional_whitespace()
if p.at_eof() or p.at_string(","):
# empty link value
continue
token = parse_link_value(p)
res.append(token)
return res
def unparse_link_header(links: ParsedLinkHeader) -> str:
return ", ".join(map(lambda v: "<" + v[0] + ">" + unparse_mime_parameters(v[1]), links))
def test_parse_link_header() -> None:
def check(xs: str | list[str], expected: ParsedLinkHeader) -> None:
if isinstance(xs, str):
xs = [xs]
for x in xs:
res = parse_link_header(x)
assert res == expected
check(
[
"<https://example.org>",
" <https://example.org>",
"<https://example.org> ",
" <https://example.org> ",
" < https://example.org > ",
],
[("https://example.org", [])],
)
check(
[
"<https://example.org>;rel=me",
"<https://example.org>; rel=me",
" <https://example.org> ; rel=me",
],
[("https://example.org", [("rel", "me")])],
)
check(
[
"<https://example.org>; rel=preconnect; crossorigin",
"<https://example.org>; rel=preconnect ; crossorigin ",
" <https://example.org> ; rel=preconnect ; crossorigin ,",
" <https://example.org> ; rel=preconnect ; ; crossorigin ; , ,, ",
# " , <https://example.org>; rel=preconnect; crossorigin , ",
],
[("https://example.org", [("rel", "preconnect"), ("crossorigin", "")])],
)
check(
[
'<https://example.org/path/#hash>; rel=canonical; type="text/html"',
],
[("https://example.org/path/#hash", [("rel", "canonical"), ("type", "text/html")])],
)
check(
[
"<https://example.org>; rel=preconnect, "
+ "<https://example.org/index.css>; as=style; rel=preload; crossorigin, "
+ '<https://example.org/index.js>;; as=script; ; rel = "preload" ; ; crossorigin; , '
+ ", "
+ "<https://example.org/main.js>; as=script; rel=preload;;",
],
[
("https://example.org", [("rel", "preconnect")]),
(
"https://example.org/index.css",
[("as", "style"), ("rel", "preload"), ("crossorigin", "")],
),
(
"https://example.org/index.js",
[("as", "script"), ("rel", " preload"), ("crossorigin", "")],
),
("https://example.org/main.js", [("as", "script"), ("rel", "preload")]),
],
)
def test_unparse_link_header() -> None:
def check(x: ParsedLinkHeader, expected: str) -> None:
res = unparse_link_header(x)
assert res == expected
back = parse_link_header(res)
assert back == x
check(
[
("https://example.org", [("rel", "me")]),
],
'<https://example.org>; rel="me"',
)
check(
[
("https://example.org", [("rel", "preconnect")]),
(
"https://example.org/index.css",
[("as", "style"), ("rel", "preload"), ("crossorigin", "")],
),
(
"https://example.org/index.js",
[("as", "script"), ("rel", " preload"), ("crossorigin", "")],
),
("https://example.org/main.js", [("as", "script"), ("rel", "preload")]),
],
'<https://example.org>; rel="preconnect", <https://example.org/index.css>; as="style"; rel="preload"; crossorigin="", <https://example.org/index.js>; as="script"; rel=" preload"; crossorigin="", <https://example.org/main.js>; as="script"; rel="preload"',
)
def parse_refresh_header(value: str) -> tuple[int, str]:
"""Parse HTTP `Refresh` header."""
p = Parser(value)
p.skip_optional_whitespace()
ngrp = p.regex(natural_re)
p.skip_optional_whitespace()
p.string(";")
p.skip_optional_whitespace()
p.string("url=")
ugrp = p.regex(url_re)
return int(_t.cast(str, ngrp[0])), _t.cast(str, ugrp[0])
def unparse_refresh_header(secs: int, url: str) -> str:
return f"{secs}; url={url}"
def test_parse_refresh_header() -> None:
def check(xs: str | list[str], expected: tuple[int, str]) -> None:
if isinstance(xs, str):
xs = [xs]
for x in xs:
res = parse_refresh_header(x)
assert res == expected
check(
[
"10;url=https://example.org/",
"10; url=https://example.org/",
"10 ;url=https://example.org/",
" 10;url=https://example.org/",
"10 ; url=https://example.org/",
],
(10, "https://example.org/"),
)
### HTML attribute parsing
type ParsedSrcsetAttr = list[tuple[str, str | None]]
srcset_condition_re = _re.compile(r"\s+([0-9]+(?:\.[0-9]+)?[xw])")
srcset_sep_re = _re.compile(r"\s*,")
def parse_srcset_attr(value: str) -> ParsedSrcsetAttr:
"""Parse HTML5 srcset attribute"""
res = []
p = Parser(value)
p.skip_optional_whitespace()
while not p.at_eof():
grp1 = p.regex(url_re)
url = _t.cast(str, grp1[1])
if url.endswith(","):
p.advance(-1)
url = url[:-1]
grp2 = p.optional_regex(srcset_condition_re)
p.skip_optional_regex(srcset_sep_re)
p.skip_optional_whitespace()
if len(url) == 0:
# ignore it
continue
res.append((url, None if grp2 is None else _t.cast(str, grp2[0])))
p.eof()
return res
def unparse_srcset_attr(value: ParsedSrcsetAttr) -> str:
"""Unparse HTML5 srcset attribute"""
return ", ".join([(url if cond is None else f"{url} {cond}") for url, cond in value])
def test_parse_srcset_attr() -> None:
def check(xs: str | list[str], expected: ParsedSrcsetAttr) -> None:
if isinstance(xs, str):
xs = [xs]
for x in xs:
res = parse_srcset_attr(x)
assert res == expected
check(
[
"https://example.org",
" https://example.org",
"https://example.org ",
" https://example.org ",
", https://example.org",
" , https://example.org",
" , https://example.org,",
" , https://example.org, ",
" , https://example.org , ",
],
[
("https://example.org", None),
],
)
check(
[
"https://example.org/1.jpg, https://example.org/2.jpg",
"https://example.org/1.jpg , https://example.org/2.jpg",
" https://example.org/1.jpg , https://example.org/2.jpg",
" https://example.org/1.jpg , https://example.org/2.jpg ",
],
[
("https://example.org/1.jpg", None),
("https://example.org/2.jpg", None),
],
)
check(
[
"https://example.org/1.jpg 2.5x, https://example.org/2.jpg",
"""
https://example.org/1.jpg 2.5x
,
https://example.org/2.jpg
""",
],
[
("https://example.org/1.jpg", "2.5x"),
("https://example.org/2.jpg", None),
],
)
def test_unparse_srcset_attr() -> None:
def check(x: ParsedSrcsetAttr, expected: str) -> None:
res = unparse_srcset_attr(x)
assert res == expected
back = parse_srcset_attr(res)
assert back == x
check(
[
("https://example.org/1.jpg", None),
],
"https://example.org/1.jpg",
)
check(
[
("https://example.org/1.jpg", None),
("https://example.org/2.jpg", None),
],
"https://example.org/1.jpg, https://example.org/2.jpg",
)
check(
[
("https://example.org/1.jpg", "5x"),
("https://example.org/2.jpg", "2.5x"),
("https://example.org/3.jpg", None),
],
"https://example.org/1.jpg 5x, https://example.org/2.jpg 2.5x, https://example.org/3.jpg",
)