Release Core v0.1.33 with redirect-sensitive headers
This commit is contained in:
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "govoplan-core"
|
||||
version = "0.1.32"
|
||||
version = "0.1.33"
|
||||
description = "Reusable GovOPlaN platform core, access, tenancy, and RBAC components."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
@@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from typing import Mapping
|
||||
|
||||
@@ -13,6 +14,9 @@ from govoplan_core.security.outbound_http import (
|
||||
|
||||
|
||||
MAX_OUTBOUND_HTTP_REQUEST_BODY_BYTES = 1_000_000
|
||||
_STANDARD_REDIRECT_SENSITIVE_HEADERS = frozenset(
|
||||
{"authorization", "proxy-authorization", "cookie", "cookie2"}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
@@ -51,6 +55,7 @@ def fetch_http(
|
||||
headers: Mapping[str, str] | None = None,
|
||||
body: bytes | None = None,
|
||||
max_bytes: int | None = None,
|
||||
redirect_sensitive_headers: Iterable[str] = (),
|
||||
) -> HttpFetchResponse:
|
||||
if body is not None and len(body) > MAX_OUTBOUND_HTTP_REQUEST_BODY_BYTES:
|
||||
raise ValueError(
|
||||
@@ -63,7 +68,12 @@ def fetch_http(
|
||||
headers=dict(headers or {}),
|
||||
method=method,
|
||||
)
|
||||
opener = build_outbound_http_opener(_PolicyRedirectHandler(label=label))
|
||||
opener = build_outbound_http_opener(
|
||||
_PolicyRedirectHandler(
|
||||
label=label,
|
||||
sensitive_headers=redirect_sensitive_headers,
|
||||
)
|
||||
)
|
||||
with opener.open(request, timeout=timeout) as response: # noqa: S310 - URL and every redirect are policy-validated. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
|
||||
response_headers = dict(response.headers.items())
|
||||
return HttpFetchResponse(
|
||||
@@ -88,6 +98,7 @@ def fetch_http_text(
|
||||
body: bytes | None = None,
|
||||
encoding: str = "utf-8",
|
||||
max_bytes: int | None = None,
|
||||
redirect_sensitive_headers: Iterable[str] = (),
|
||||
) -> str:
|
||||
return fetch_http(
|
||||
url,
|
||||
@@ -97,13 +108,22 @@ def fetch_http_text(
|
||||
headers=headers,
|
||||
body=body,
|
||||
max_bytes=max_bytes,
|
||||
redirect_sensitive_headers=redirect_sensitive_headers,
|
||||
).text(encoding)
|
||||
|
||||
|
||||
class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
def __init__(self, *, label: str) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
label: str,
|
||||
sensitive_headers: Iterable[str] = (),
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self._label = label
|
||||
self._sensitive_headers = _STANDARD_REDIRECT_SENSITIVE_HEADERS | {
|
||||
value.strip().lower() for value in sensitive_headers if value.strip()
|
||||
}
|
||||
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def]
|
||||
candidate = validate_outbound_http_url(newurl, label=f"{self._label} redirect")
|
||||
@@ -113,8 +133,9 @@ class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
return None
|
||||
new_request = super().redirect_request(req, fp, code, msg, headers, candidate)
|
||||
if new_request is not None and _http_origin(previous) != _http_origin(redirected):
|
||||
for header in ("Authorization", "Proxy-Authorization", "Cookie", "Cookie2"):
|
||||
new_request.remove_header(header)
|
||||
for header in tuple(new_request.headers) + tuple(new_request.unredirected_hdrs):
|
||||
if header.lower() in self._sensitive_headers:
|
||||
new_request.remove_header(header)
|
||||
return new_request
|
||||
|
||||
|
||||
|
||||
@@ -239,9 +239,17 @@ class HttpFetchTests(unittest.TestCase):
|
||||
|
||||
request = urllib.request.Request(
|
||||
"https://catalog.example.test/releases",
|
||||
headers={"Authorization": "Bearer secret", "X-Request-ID": "request-1"},
|
||||
headers={
|
||||
"Authorization": "Bearer secret",
|
||||
"Cookie": "session=secret",
|
||||
"X-OTRS-Header-Password": "secret",
|
||||
"X-Request-ID": "request-1",
|
||||
},
|
||||
)
|
||||
handler = _PolicyRedirectHandler(
|
||||
label="Catalog URL",
|
||||
sensitive_headers=("X-OTRS-Header-Password",),
|
||||
)
|
||||
handler = _PolicyRedirectHandler(label="Catalog URL")
|
||||
with patch.dict("os.environ", {"APP_ENV": "test"}), patch(
|
||||
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
||||
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
||||
@@ -265,9 +273,38 @@ class HttpFetchTests(unittest.TestCase):
|
||||
|
||||
self.assertIsNotNone(redirected)
|
||||
self.assertIsNone(redirected.get_header("Authorization"))
|
||||
self.assertIsNone(redirected.get_header("Cookie"))
|
||||
self.assertIsNone(redirected.get_header("X-otrs-header-password"))
|
||||
self.assertEqual("request-1", redirected.get_header("X-request-id"))
|
||||
self.assertIsNone(downgrade)
|
||||
|
||||
def test_core_redirects_preserve_caller_sensitive_headers_on_the_same_origin(self) -> None:
|
||||
import urllib.request
|
||||
|
||||
request = urllib.request.Request(
|
||||
"https://desk.example.test/original",
|
||||
headers={"X-OTRS-Header-SessionID": "secret"},
|
||||
)
|
||||
handler = _PolicyRedirectHandler(
|
||||
label="Service-desk URL",
|
||||
sensitive_headers=("X-OTRS-Header-SessionID",),
|
||||
)
|
||||
with patch.dict("os.environ", {"APP_ENV": "test"}), patch(
|
||||
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
||||
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
||||
):
|
||||
redirected = handler.redirect_request(
|
||||
request,
|
||||
None,
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
"https://desk.example.test/final",
|
||||
)
|
||||
|
||||
self.assertIsNotNone(redirected)
|
||||
self.assertEqual("secret", redirected.get_header("X-otrs-header-sessionid"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Generated
+3
-3
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"dependencies": {
|
||||
"@govoplan/access-webui": "file:../../govoplan-access/webui",
|
||||
"@govoplan/addresses-webui": "file:../../govoplan-addresses/webui",
|
||||
@@ -243,7 +243,7 @@
|
||||
},
|
||||
"../../govoplan-connectors/webui": {
|
||||
"name": "@govoplan/connectors-webui",
|
||||
"version": "0.1.21",
|
||||
"version": "0.1.22",
|
||||
"peerDependencies": {
|
||||
"@govoplan/core-webui": "^0.1.18",
|
||||
"react": ">=19.2.7 <20",
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"dependencies": {
|
||||
"@govoplan/access-webui": "git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-access.git#v0.1.19",
|
||||
"@govoplan/admin-webui": "git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-admin.git#v0.1.18",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "src/index.ts",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@govoplan/core-webui",
|
||||
"version": "0.1.32",
|
||||
"version": "0.1.33",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "src/index.ts",
|
||||
|
||||
Reference in New Issue
Block a user