Coverage for dataexcept/_previews.py: 100%

1"""Bounded excerpts of a payload that failed to parse. 

2 

3A parsing failure happens on content the caller did not write and often did not 

4choose: an API response, a queue message, a file from somewhere else. Storing 

5the whole offending payload on the exception drags it into every log line, 

6envelope and telemetry event that touches the failure -- including whatever 

7credentials or personal data the payload happened to contain. 

8 

9The excerpt is the compromise. It is short enough to read in a log and long 

10enough to show what the parser choked on, and because it is an ordinary string 

11attribute it goes through the same URL redaction as every other public 

12attribute on a DataExcept exception. 

13""" 

14 

15from __future__ import annotations 

16 

17__all__ = ["MAX_PREVIEW_LENGTH", "TRUNCATION_MARKER", "bounded_preview"] 

18 

19#: Longest excerpt kept, in characters, counting the truncation marker. A 

20#: caller who wants less can pass less; there is deliberately no way to ask for 

21#: more, because the point of the field is that an untrusted payload cannot 

22#: reach a log at its own length. 

23MAX_PREVIEW_LENGTH = 200 

24 

25#: Appended when anything was cut, and only then. 

26TRUNCATION_MARKER = "..." 

27 

28#: Bytes decoded before the excerpt is measured. Four is the longest a UTF-8 

29#: character gets, so this always yields at least MAX_PREVIEW_LENGTH characters 

30#: when the payload has them, without decoding a payload of any size. 

31_BYTE_WINDOW = MAX_PREVIEW_LENGTH * 4 

32 

33 

34def _as_text(value: str | bytes | bytearray) -> tuple[str, bool]: 

35 """Return *value* as text, and whether reading it already cut something.""" 

36 if isinstance(value, str): 

37 return value, False 

38 head = bytes(value[:_BYTE_WINDOW]) 

39 # backslashreplace, not replace: an excerpt of a malformed payload is a 

40 # best effort by definition and must not fail where the parser did, but it 

41 # should stay printable. U+FFFD is neither readable nor writable on a 

42 # console that is not UTF-8, while `\xff` says which byte was wrong. 

43 return head.decode("utf-8", errors="backslashreplace"), len(value) > len(head) 

44 

45 

46def bounded_preview(value: str | bytes | bytearray | None) -> str | None: 

47 r"""Return at most :data:`MAX_PREVIEW_LENGTH` characters of *value*. 

48 

49 ``None`` gives ``None``: no excerpt was asked for. Bytes are decoded as 

50 UTF-8, with any undecodable byte shown as ``\xNN``. 

51 :data:`TRUNCATION_MARKER` is appended when, and only when, something was 

52 left out, and the result stays within the bound either way. 

53 

54 The bound is on the payload. Redaction runs afterwards, over this value as 

55 over every other public attribute, and a credential shorter than its 

56 replacement makes the stored string a few characters longer. Chasing an 

57 exact count through a pass whose job is to remove secrets rather than 

58 preserve lengths would buy nothing: the excerpt is bounded so that an 

59 untrusted payload cannot reach a log at its own size, and it does that. 

60 """ 

61 if value is None: 

62 return None 

63 if not isinstance(value, (str, bytes, bytearray)): 

64 raise TypeError( 

65 f"preview must be str, bytes, or None, got {type(value).__name__}" 

66 ) 

67 

68 text, truncated = _as_text(value) 

69 if not truncated and len(text) <= MAX_PREVIEW_LENGTH: 

70 return text 

71 # Something was left out, whether by the decode window or by length. Cut 

72 # short enough for the marker to fit inside the bound rather than on top of 

73 # it: a documented maximum the value can exceed is not a maximum. 

74 return text[: MAX_PREVIEW_LENGTH - len(TRUNCATION_MARKER)] + TRUNCATION_MARKER