refactor(observability): redact url at the _emit_event chokepoint

Move URL redaction from per-emit-site _observed_url() into _emit_event
itself, so the single emission boundary sanitizes the `url` attribute for
both the log record and the OTel span event. A future emit site can no
longer reintroduce the leak by passing a raw url, and the four middleware
modules revert to their original `str(request.url)` (net-unchanged vs main).

Adds direct _emit_event redaction tests for both channels.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lesnik512

src/httpware/_internal/observability.py

@@ -12,17 +12,10 @@
 import logging
 import typing
 
-import httpx2
-
 from httpware._internal import import_checker
 from httpware._internal.redaction import redact_url
 
 
-def _observed_url(request: httpx2.Request) -> str:
-    """Return the request URL safe for emission (userinfo + sensitive query masked)."""
-    return redact_url(str(request.url))
-
-
 def _emit_event(
     logger: logging.Logger,
     event_name: str,
@@ -33,6 +26,12 @@ def _emit_event(
 ) -> None:
     """Emit one observability event to both channels.
 
+    The ``url`` attribute, when present, is run through
+    ``redaction.redact_url`` here — at the single emission boundary — so a
+    request URL's userinfo and known-sensitive query/fragment secrets never
+    reach a log record or span event, regardless of how a caller built the
+    attributes dict.
+
     1. Always emits a structured log record at ``level`` with ``extra=attributes``
        (so log aggregators that index ``extra`` see structured fields).
     2. If ``opentelemetry-api`` is installed, calls
@@ -47,7 +46,9 @@ def _emit_event(
     the optional-extras isolation invariant: ``import httpware`` must not pull
     ``opentelemetry`` into ``sys.modules`` when the extra is absent.
     """
-    logger.log(level, message, extra={**attributes, "event": event_name})
+    raw_url = attributes.get("url")
+    safe_attributes = {**attributes, "url": redact_url(raw_url)} if isinstance(raw_url, str) else attributes
+    logger.log(level, message, extra={**safe_attributes, "event": event_name})
     if import_checker.is_otel_installed:
         try:
             from opentelemetry import trace  # noqa: PLC0415 — lazy by design (optional-extras isolation)
@@ -60,4 +61,4 @@ def _emit_event(
         # The structured log record above has already fired; CancelledError/KeyboardInterrupt
         # are not Exception subclasses and will still propagate.
         with contextlib.suppress(Exception):
-            trace.get_current_span().add_event(event_name, attributes=attributes)
+            trace.get_current_span().add_event(event_name, attributes=safe_attributes)

src/httpware/middleware/resilience/bulkhead.py

@@ -24,7 +24,7 @@
 
 import httpx2
 
-from httpware._internal.observability import _emit_event, _observed_url
+from httpware._internal.observability import _emit_event
 from httpware.errors import BulkheadFullError
 from httpware.middleware import AsyncNext, Next
 
@@ -114,7 +114,7 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                     "max_concurrent": self._max_concurrent,
                     "acquire_timeout": self._acquire_timeout,
                     "method": request.method,
-                    "url": _observed_url(request),
+                    "url": str(request.url),
                 },
             )
             raise BulkheadFullError(
@@ -170,7 +170,7 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
                     "max_concurrent": self._max_concurrent,
                     "acquire_timeout": self._acquire_timeout,
                     "method": request.method,
-                    "url": _observed_url(request),
+                    "url": str(request.url),
                 },
             )
             raise BulkheadFullError(

src/httpware/middleware/resilience/circuit_breaker.py

@@ -34,7 +34,7 @@
 
 import httpx2
 
-from httpware._internal.observability import _emit_event, _observed_url
+from httpware._internal.observability import _emit_event
 from httpware.errors import CircuitOpenError, NetworkError, StatusError, TimeoutError  # noqa: A004
 from httpware.middleware import AsyncNext, Next
 
@@ -190,7 +190,7 @@ def _emit(
             event_name,
             level=level,
             message=message,
-            attributes={**attributes, "method": request.method, "url": _observed_url(request)},
+            attributes={**attributes, "method": request.method, "url": str(request.url)},
         )
 
 

src/httpware/middleware/resilience/retry.py

@@ -17,7 +17,7 @@
 
 import httpx2
 
-from httpware._internal.observability import _emit_event, _observed_url
+from httpware._internal.observability import _emit_event
 from httpware._internal.status import STREAMING_BODY_MARKER
 from httpware.errors import NetworkError, RetryBudgetExhaustedError, StatusError, TimeoutError  # noqa: A004
 from httpware.middleware import AsyncNext, Next
@@ -133,7 +133,7 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                     message="retry refused — request body is a stream that cannot replay",
                     attributes={
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_exception_type": type(last_exc).__qualname__,
                     },
                 )
@@ -152,7 +152,7 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                     attributes={
                         "attempts": attempt + 1,
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_status": last_response.status_code if last_response is not None else None,
                         "last_exception_type": type(last_exc).__qualname__,
                     },
@@ -168,7 +168,7 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                     attributes={
                         "attempts": attempt + 1,
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_status": last_response.status_code if last_response is not None else None,
                     },
                 )
@@ -271,7 +271,7 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
                     message="retry refused — request body is a stream that cannot replay",
                     attributes={
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_exception_type": type(last_exc).__qualname__,
                     },
                 )
@@ -290,7 +290,7 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
                     attributes={
                         "attempts": attempt + 1,
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_status": last_response.status_code if last_response is not None else None,
                         "last_exception_type": type(last_exc).__qualname__,
                     },
@@ -306,7 +306,7 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
                     attributes={
                         "attempts": attempt + 1,
                         "method": request.method,
-                        "url": _observed_url(request),
+                        "url": str(request.url),
                         "last_status": last_response.status_code if last_response is not None else None,
                     },
                 )

src/httpware/middleware/resilience/timeout.py

@@ -18,7 +18,7 @@
 
 import httpx2
 
-from httpware._internal.observability import _emit_event, _observed_url
+from httpware._internal.observability import _emit_event
 from httpware.errors import TimeoutError as HttpwareTimeoutError
 from httpware.middleware import AsyncNext
 
@@ -69,7 +69,7 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                 attributes={
                     "timeout": self._timeout,
                     "method": request.method,
-                    "url": _observed_url(request),
+                    "url": str(request.url),
                 },
             )
             msg = f"overall timeout of {self._timeout}s exceeded"

tests/test_observability.py

@@ -32,6 +32,43 @@ def test_emit_event_logs_at_warning_with_extra_fields(caplog: pytest.LogCaptureF
     assert record.event == "test.event"  # ty: ignore[unresolved-attribute]
 
 
+def test_emit_event_redacts_url_secret_in_log_record(caplog: pytest.LogCaptureFixture) -> None:
+    """The `url` attribute is redacted at the emission boundary, before the log record fires."""
+    with caplog.at_level(logging.WARNING, logger="httpware.test.observability"):
+        _emit_event(
+            _TEST_LOGGER,
+            "test.event",
+            level=logging.WARNING,
+            message="leaky",
+            attributes={"url": "https://u:p@example.test/x?api_key=topsecret"},
+        )
+
+    record = caplog.records[0]
+    assert "topsecret" not in record.url  # ty: ignore[unresolved-attribute]
+    assert "api_key=REDACTED" in record.url  # ty: ignore[unresolved-attribute]
+    assert "u:p@" not in record.url  # ty: ignore[unresolved-attribute]
+
+
+def test_emit_event_redacts_url_secret_in_otel_event() -> None:
+    """The OTel span event receives the redacted `url`, not the raw secret."""
+    mock_span = MagicMock(name="MockSpan")
+    with (
+        patch("httpware._internal.import_checker.is_otel_installed", True),
+        patch("opentelemetry.trace.get_current_span", return_value=mock_span),
+    ):
+        _emit_event(
+            _TEST_LOGGER,
+            "test.event",
+            level=logging.WARNING,
+            message="leaky",
+            attributes={"url": "https://example.test/x?token=topsecret"},
+        )
+
+    _, kwargs = mock_span.add_event.call_args
+    assert "topsecret" not in kwargs["attributes"]["url"]
+    assert "token=REDACTED" in kwargs["attributes"]["url"]
+
+
 def test_emit_event_respects_level_parameter(caplog: pytest.LogCaptureFixture) -> None:
     """When level=DEBUG is passed, the record is at DEBUG."""
     with caplog.at_level(logging.DEBUG, logger="httpware.test.observability"):