Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/logging_utils/scrubbing.py: 93%

1from opal_common.http_utils import redact_url_in_text 

2 

3 

4class CredentialScrubbingStream: 

5 """A writable-stream wrapper that scrubs embedded URL credentials from 

6 every fully-formatted log record before it reaches the underlying stream. 

7 

8 This is the catch-all that the model-layer redaction (``RedactedReprMixin``) 

9 cannot provide: third-party exceptions - most notably aiohttp 

10 ``ClientResponseError`` - render the full request URL 

11 (``https://user:pw@host/path?token=...``) inside the traceback that loguru 

12 formats into the ``{exception}`` slot. The leaking object is not an OPAL 

13 model, so masking ``repr()`` does nothing for it. 

14 

15 Because loguru hands a stream sink the *final* formatted text (message + 

16 traceback, and the JSON blob when ``serialize=True``), scrubbing on 

17 ``write`` covers every record shape on this sink. ``redact_url_in_text`` is 

18 idempotent, so the explicit call-site redaction upstream is unaffected. 

19 

20 Note: this wraps a *stream* sink. loguru's path-based file sink (with 

21 rotation/retention) opens the file itself and cannot be wrapped this way - 

22 message-level scrubbing for that sink is handled in ``Formatter.format``. 

23 """ 

24 

25 def __init__(self, stream): 

26 self._stream = stream 

27 

28 def write(self, message: str): 

29 return self._stream.write(redact_url_in_text(message)) 

30 

31 def flush(self): 

32 # loguru flushes the sink after each record when possible. 

33 flush = getattr(self._stream, "flush", None) 

34 if callable(flush): 34 ↛ exitline 34 didn't return from function 'flush' because the condition on line 34 was always true

35 flush() 

36 

37 def __getattr__(self, name): 

38 # Delegate everything else (isatty/fileno/encoding/...) to the wrapped 

39 # stream so loguru treats this exactly like the underlying stream. 

40 return getattr(self._stream, name)