didericis-codex
b791fe4b84
test / unit (pull_request) Waiting to run
test / image-input-builds (pull_request) Waiting to run
test / coverage (pull_request) Blocked by required conditions
test / integration-docker (pull_request) Blocked by required conditions
tracker-policy-pr / check-pr (pull_request) Failing after 12m49s
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""Outbound DLP request scanning and redaction for the egress pipeline."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import ItemsView, Mapping, Protocol
|
|
|
|
from .dlp import (
|
|
build_outbound_scan_text,
|
|
outbound_scan_headers,
|
|
scan_outbound,
|
|
)
|
|
from .dlp_detectors import redact_tokens, strip_crlf
|
|
from .types import Route, ScanResult
|
|
|
|
|
|
class MutableHeaders(Protocol):
|
|
def items(self) -> ItemsView[str, str]: ...
|
|
def __getitem__(self, name: str, /) -> str: ...
|
|
def __setitem__(self, name: str, value: str, /) -> None: ...
|
|
|
|
|
|
class MutableRequest(Protocol):
|
|
pretty_host: str
|
|
path: str
|
|
headers: MutableHeaders
|
|
text: str
|
|
|
|
def get_text(self, strict: bool = False) -> str | None: ...
|
|
|
|
|
|
def scan_request(
|
|
request: MutableRequest,
|
|
route: Route,
|
|
env: Mapping[str, str],
|
|
*,
|
|
safe_tokens: set[str] | None = None,
|
|
) -> ScanResult | None:
|
|
"""Scan all mutable outbound request surfaces in their canonical order."""
|
|
request_path, _, query = request.path.partition("?")
|
|
headers = outbound_scan_headers(route, dict(request.headers.items()))
|
|
body = request.get_text(strict=False) or ""
|
|
scan_text = build_outbound_scan_text(
|
|
request.pretty_host, request_path, query, headers, body,
|
|
)
|
|
# Bodies cannot alter HTTP framing, so CRLF detection is deliberately
|
|
# restricted to the request line and headers.
|
|
crlf_text = build_outbound_scan_text(
|
|
request.pretty_host, request_path, query, headers, "",
|
|
)
|
|
return scan_outbound(
|
|
route,
|
|
scan_text,
|
|
env,
|
|
safe_tokens=safe_tokens,
|
|
crlf_text=crlf_text,
|
|
)
|
|
|
|
|
|
def redact_request(
|
|
request: MutableRequest,
|
|
route: Route,
|
|
env: Mapping[str, str],
|
|
) -> bool:
|
|
"""Redact mutable request surfaces and return whether the result is clean."""
|
|
body = request.get_text(strict=False)
|
|
if body:
|
|
redacted_body = redact_tokens(body, env=env)
|
|
if redacted_body != body:
|
|
request.text = redacted_body
|
|
for name, value in list(request.headers.items()):
|
|
if name.lower() == "host":
|
|
continue
|
|
redacted = strip_crlf(redact_tokens(value, env=env))
|
|
if redacted != value:
|
|
request.headers[name] = redacted
|
|
redacted_path = strip_crlf(redact_tokens(request.path, env=env))
|
|
if redacted_path != request.path:
|
|
request.path = redacted_path
|
|
result = scan_request(request, route, env)
|
|
return result is None or result.severity != "block"
|
|
|
|
|
|
__all__ = ["MutableRequest", "redact_request", "scan_request"]
|