Authored-By: @architect <architect@fritzlab.net>
This commit is contained in:
+36
-7
@@ -1,5 +1,6 @@
|
|||||||
"""Shared utilities for the site-publish action."""
|
"""Shared utilities for the site-publish action."""
|
||||||
|
|
||||||
|
import ipaddress
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import shutil
|
import shutil
|
||||||
@@ -105,22 +106,47 @@ def _strings(value, label):
|
|||||||
|
|
||||||
|
|
||||||
def _cors_origins(value, label):
|
def _cors_origins(value, label):
|
||||||
origins = _strings(value or [], label)
|
origins = _list(value, label)
|
||||||
if len(origins) != len(set(origins)):
|
if any(not isinstance(item, str) or not item for item in origins):
|
||||||
raise ConfigError(f"{label} must not contain duplicates")
|
raise ConfigError(f"{label} must be a list of non-empty strings")
|
||||||
|
canonical = []
|
||||||
for origin in origins:
|
for origin in origins:
|
||||||
if origin == "*":
|
if origin == "*":
|
||||||
|
canonical.append(origin)
|
||||||
continue
|
continue
|
||||||
parsed = urlparse(origin)
|
parsed = urlparse(origin)
|
||||||
try:
|
try:
|
||||||
port = parsed.port
|
port = parsed.port
|
||||||
except ValueError:
|
except ValueError:
|
||||||
port = None
|
raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins") from None
|
||||||
if parsed.scheme != "https" or not parsed.hostname or parsed.path or parsed.params or (
|
if parsed.scheme != "https" or not parsed.hostname or parsed.path or parsed.params or (
|
||||||
parsed.query or parsed.fragment or parsed.username or parsed.password
|
parsed.query or parsed.fragment or parsed.username or parsed.password
|
||||||
or (":" in parsed.netloc and port is None and not parsed.netloc.endswith("]"))
|
|
||||||
):
|
):
|
||||||
raise ConfigError(f"{label} must contain '*' or HTTPS origins")
|
raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins")
|
||||||
|
try:
|
||||||
|
address = ipaddress.ip_address(parsed.hostname)
|
||||||
|
except ValueError:
|
||||||
|
try:
|
||||||
|
hostname = parsed.hostname.encode("idna").decode("ascii")
|
||||||
|
except UnicodeError:
|
||||||
|
raise ConfigError(
|
||||||
|
f"{label} must contain '*' or canonical HTTPS origins"
|
||||||
|
) from None
|
||||||
|
try:
|
||||||
|
_hostname(hostname, f"{label} hostname")
|
||||||
|
except ConfigError:
|
||||||
|
raise ConfigError(
|
||||||
|
f"{label} must contain '*' or canonical HTTPS origins"
|
||||||
|
) from None
|
||||||
|
else:
|
||||||
|
hostname = f"[{address.compressed}]" if address.version == 6 else address.compressed
|
||||||
|
canonical.append(f"https://{hostname}{f':{port}' if port not in (None, 443) else ''}")
|
||||||
|
if len(canonical) != len(set(canonical)):
|
||||||
|
raise ConfigError(f"{label} must not contain duplicate canonical origins")
|
||||||
|
if "*" in canonical and len(canonical) != 1:
|
||||||
|
raise ConfigError(f"{label} wildcard must be the only origin")
|
||||||
|
if origins != canonical:
|
||||||
|
raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins")
|
||||||
return origins
|
return origins
|
||||||
|
|
||||||
|
|
||||||
@@ -328,7 +354,10 @@ def _artifact(item, index):
|
|||||||
"website_authority": authority,
|
"website_authority": authority,
|
||||||
"credentials": normalized_credentials,
|
"credentials": normalized_credentials,
|
||||||
"cache_rules": cache_rules,
|
"cache_rules": cache_rules,
|
||||||
"cors_origins": _cors_origins(item.get("cors_origins"), f"{label}.cors_origins"),
|
"cors_origins": (
|
||||||
|
_cors_origins(item["cors_origins"], f"{label}.cors_origins")
|
||||||
|
if "cors_origins" in item else []
|
||||||
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+40
-3
@@ -123,14 +123,51 @@ class ConfigContractTests(unittest.TestCase):
|
|||||||
lambda raw: raw["artifacts"][1].__setitem__(
|
lambda raw: raw["artifacts"][1].__setitem__(
|
||||||
"cors_origins", ["http://consumer.example"]
|
"cors_origins", ["http://consumer.example"]
|
||||||
),
|
),
|
||||||
"must contain '\\*' or HTTPS origins",
|
"must contain '\\*' or canonical HTTPS origins",
|
||||||
|
)
|
||||||
|
for origin in (
|
||||||
|
"https://consumer example",
|
||||||
|
"https://Consumer.example",
|
||||||
|
"https://consumer.example:443",
|
||||||
|
"https://consumer.example/",
|
||||||
|
"https://consumer.example:invalid",
|
||||||
|
):
|
||||||
|
with self.subTest(origin=origin):
|
||||||
|
self.assert_invalid(
|
||||||
|
lambda raw, origin=origin: raw["artifacts"][1].__setitem__(
|
||||||
|
"cors_origins", [origin]
|
||||||
|
),
|
||||||
|
"must contain '\\*' or canonical HTTPS origins",
|
||||||
|
)
|
||||||
|
self.assert_invalid(
|
||||||
|
lambda raw: raw["artifacts"][1].__setitem__(
|
||||||
|
"cors_origins", ["https://consumer.example", "https://consumer.example:443"]
|
||||||
|
),
|
||||||
|
"must not contain duplicate canonical origins",
|
||||||
)
|
)
|
||||||
self.assert_invalid(
|
self.assert_invalid(
|
||||||
lambda raw: raw["artifacts"][1].__setitem__(
|
lambda raw: raw["artifacts"][1].__setitem__(
|
||||||
"cors_origins", ["https://consumer.example", "https://consumer.example"]
|
"cors_origins", ["*", "https://consumer.example"]
|
||||||
),
|
),
|
||||||
"must not contain duplicates",
|
"wildcard must be the only origin",
|
||||||
)
|
)
|
||||||
|
raw = copy.deepcopy(self.raw)
|
||||||
|
raw["artifacts"][1]["cors_origins"] = ["https://[2602:817:3000::1]:8443"]
|
||||||
|
cfg = normalize_site_config(raw, "baseline.fritzlab.net")
|
||||||
|
self.assertEqual(
|
||||||
|
["https://[2602:817:3000::1]:8443"],
|
||||||
|
next(a for a in cfg["artifacts"] if a["name"] == "distributions")["cors_origins"],
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_declared_cors_origins_must_be_a_list(self):
|
||||||
|
for value in (None, False, 0, "", {}):
|
||||||
|
with self.subTest(value=value):
|
||||||
|
self.assert_invalid(
|
||||||
|
lambda raw, value=value: raw["artifacts"][1].__setitem__(
|
||||||
|
"cors_origins", value
|
||||||
|
),
|
||||||
|
"cors_origins must be a list",
|
||||||
|
)
|
||||||
|
|
||||||
def test_protected_route_rejects_wildcard_cors(self):
|
def test_protected_route_rejects_wildcard_cors(self):
|
||||||
self.assert_invalid(
|
self.assert_invalid(
|
||||||
|
|||||||
Reference in New Issue
Block a user