Skip to content

Commit 695d7ee

Browse files
committed
fix: bound physical line reads and sanitize blocked markers
1 parent 6dbb93e commit 695d7ee

5 files changed

Lines changed: 248 additions & 20 deletions

File tree

‎assert‎

Whitespace-only changes.

‎src/logprivacy/files/operations.py‎

Lines changed: 53 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,21 @@
55
import os
66
import stat
77
import tempfile
8-
from collections.abc import Callable
8+
from collections.abc import Callable, Iterator
99
from contextlib import suppress
1010
from pathlib import Path
11-
from typing import cast
11+
from typing import IO, cast
1212

1313
from logprivacy.audit import AuditReport
1414
from logprivacy.cleaner import Cleaner
15+
from logprivacy.exceptions import InputLimitExceededError
1516
from logprivacy.internal.audit_location import format_file_location
1617
from logprivacy.internal.traversal import LIMIT_MAX_FINDINGS
1718
from logprivacy.policy import CleanerPolicy
1819
from logprivacy.result import Finding
1920

21+
_MAX_PHYSICAL_LINE_CHARS = 1_000_000
22+
_LIMIT_MAX_LINE_CHARS = "max_line_chars"
2023
_TEMP_PREFIX = ".logprivacy-"
2124
_TEMP_SUFFIX = ".tmp"
2225
_FCHMOD = cast(Callable[[int, int], None] | None, getattr(os, "fchmod", None))
@@ -34,10 +37,19 @@ def scan_file(
3437
limitations: list[str] = []
3538
file_path = Path(path)
3639
safe_name = cleaner._sanitize_location_text(file_path.name)
37-
line_number = 0
38-
with file_path.open("r", encoding=encoding, errors="replace") as stream:
39-
for line in stream:
40-
line_number += 1
40+
with file_path.open(
41+
"r",
42+
encoding=encoding,
43+
errors="replace",
44+
newline="",
45+
) as stream:
46+
for line_number, line, line_error in _iter_bounded_text_lines(stream):
47+
if line_error is not None:
48+
if line_error not in limitations:
49+
limitations.append(line_error)
50+
continue
51+
52+
assert line is not None
4153
report = cleaner.audit(line)
4254
for limitation in report.limitations:
4355
if limitation not in limitations:
@@ -206,7 +218,13 @@ def _write_cleaned_file(
206218
newline="",
207219
) as source,
208220
):
209-
for line in source:
221+
for _, line, line_error in _iter_bounded_text_lines(source):
222+
if line_error is not None:
223+
raise InputLimitExceededError(
224+
limit=line_error,
225+
maximum=_MAX_PHYSICAL_LINE_CHARS,
226+
)
227+
assert line is not None
210228
target.write(cleaner.clean_text(line))
211229

212230
if _FCHMOD is not None:
@@ -219,6 +237,34 @@ def _write_cleaned_file(
219237
os.chmod(temporary_path, target_mode)
220238

221239

240+
def _iter_bounded_text_lines(
241+
stream: IO[str],
242+
) -> Iterator[tuple[int, str | None, str | None]]:
243+
"""Yield physical lines without allowing one line to allocate unbounded memory."""
244+
line_number = 0
245+
while True:
246+
line = stream.readline(_MAX_PHYSICAL_LINE_CHARS + 1)
247+
if line == "":
248+
return
249+
250+
line_number += 1
251+
if len(line) > _MAX_PHYSICAL_LINE_CHARS:
252+
if not line.endswith("\n"):
253+
_discard_text_line_remainder(stream)
254+
yield line_number, None, _LIMIT_MAX_LINE_CHARS
255+
continue
256+
257+
yield line_number, line, None
258+
259+
260+
def _discard_text_line_remainder(stream: IO[str]) -> None:
261+
"""Consume an oversized line in bounded chunks so later lines stay readable."""
262+
while True:
263+
chunk = stream.readline(_MAX_PHYSICAL_LINE_CHARS + 1)
264+
if chunk == "" or chunk.endswith("\n"):
265+
return
266+
267+
222268
def _remove_temporary_file(path: Path) -> None:
223269
"""Remove an unfinished temporary file without hiding the original failure."""
224270
with suppress(OSError):

‎src/logprivacy/integrations/logging_filter.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -202,10 +202,11 @@ def filter(self, record: logging.LogRecord) -> bool:
202202
if self.drop_blocked is False:
203203
raise
204204
categories = ",".join(exc.categories)
205-
_replace_record_with_marker(
206-
record,
205+
marker = sanitize_output_text(
207206
f"[LOGPRIVACY BLOCKED categories={categories}]",
207+
max_chars=DEFAULT_MAX_RENDER_CHARS,
208208
)
209+
_replace_record_with_marker(record, marker)
209210
return True
210211
except InputLimitExceededError as exc:
211212
_replace_record_with_marker(

‎src/logprivacy/jsonl/streaming.py‎

Lines changed: 37 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
import tempfile
99
from collections.abc import Iterator
1010
from pathlib import Path
11-
from typing import IO, Literal, TextIO, cast
11+
from typing import IO, Literal, TextIO
1212

1313
from logprivacy.adapters import AdapterRegistry
1414
from logprivacy.cleaner import Cleaner
@@ -32,6 +32,8 @@
3232

3333
_LIMIT_INVALID_JSON = "invalid_json"
3434
_LIMIT_INVALID_UTF8 = "invalid_utf8"
35+
_LIMIT_MAX_LINE_BYTES = "max_line_bytes"
36+
_MAX_PHYSICAL_LINE_BYTES = 1_000_000
3537

3638

3739
def safe_jsonl_write(
@@ -314,23 +316,32 @@ def _iter_source_lines(source: _SourceType) -> Iterator[_SourceLine]:
314316

315317

316318
def _iter_lines(stream: _SourceStream) -> Iterator[_SourceLine]:
317-
"""Yield line number, decoded text, and an optional safe error reason."""
318-
iterator = iter(stream)
319+
"""Yield source lines while bounding each physical read."""
319320
line_number = 0
320321

321322
while True:
322323
try:
323-
raw_line = next(iterator)
324-
except StopIteration:
325-
return
324+
raw_line = stream.readline(_MAX_PHYSICAL_LINE_BYTES + 1)
326325
except UnicodeDecodeError:
327326
# A caller-provided text stream may fail inside its decoder. Its
328327
# state is not guaranteed to be recoverable, so report one error
329328
# record and stop without exposing decoder input.
330329
yield line_number + 1, None, _LIMIT_INVALID_UTF8
331330
return
332331

332+
if raw_line == b"" or raw_line == "":
333+
return
334+
333335
line_number += 1
336+
if len(raw_line) > _MAX_PHYSICAL_LINE_BYTES:
337+
if not _ends_with_newline(raw_line):
338+
try:
339+
_discard_line_remainder(stream)
340+
except UnicodeDecodeError:
341+
yield line_number, None, _LIMIT_MAX_LINE_BYTES
342+
return
343+
yield line_number, None, _LIMIT_MAX_LINE_BYTES
344+
continue
334345

335346
if isinstance(raw_line, bytes):
336347
try:
@@ -339,17 +350,32 @@ def _iter_lines(stream: _SourceStream) -> Iterator[_SourceLine]:
339350
yield line_number, None, _LIMIT_INVALID_UTF8
340351
continue
341352
else:
342-
# ``stream`` is a union of text and binary streams. Mypy cannot
343-
# always narrow the result of ``next(iterator)`` to ``str`` here,
344-
# even though the bytes branch was handled above.
345-
decoded = cast(str, raw_line)
353+
decoded = raw_line
346354

347355
yield line_number, decoded, None
348356

349357

358+
def _ends_with_newline(value: str | bytes) -> bool:
359+
"""Return whether a bounded chunk reaches the end of its physical line."""
360+
return value.endswith(b"\n") if isinstance(value, bytes) else value.endswith("\n")
361+
362+
363+
def _discard_line_remainder(stream: _SourceStream) -> None:
364+
"""Consume an oversized physical line in bounded chunks."""
365+
while True:
366+
chunk = stream.readline(_MAX_PHYSICAL_LINE_BYTES + 1)
367+
if chunk == b"" or chunk == "" or _ends_with_newline(chunk):
368+
return
369+
370+
350371
def _line_processing_error(line_number: int, reason: str) -> JSONLProcessingError:
351372
"""Build a safe public exception for one invalid source line."""
352-
description = "invalid UTF-8" if reason == _LIMIT_INVALID_UTF8 else "invalid JSON"
373+
if reason == _LIMIT_INVALID_UTF8:
374+
description = "invalid UTF-8"
375+
elif reason == _LIMIT_MAX_LINE_BYTES:
376+
description = "physical line exceeds the safety limit"
377+
else:
378+
description = "invalid JSON"
353379
return JSONLProcessingError(
354380
f"{description} at line {line_number}",
355381
line_number=line_number,
Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
from __future__ import annotations
2+
3+
import logging
4+
from pathlib import Path
5+
6+
import pytest
7+
8+
from logprivacy.cleaner import Cleaner
9+
from logprivacy.exceptions import InputLimitExceededError
10+
from logprivacy.files.operations import clean_file, scan_file
11+
from logprivacy.integrations.logging_filter import LogPrivacyFilter
12+
from logprivacy.internal.matches import _DetectedMatch
13+
from logprivacy.jsonl.streaming import _iter_lines, iter_safe_jsonl
14+
from logprivacy.policy import CleanerPolicy
15+
16+
_MAX_LINE = 1_000_000
17+
18+
19+
class _GuardedBinaryStream:
20+
def __init__(self, data: bytes) -> None:
21+
self.data = data
22+
self.position = 0
23+
self.requested_sizes: list[int] = []
24+
25+
def __iter__(self):
26+
return self
27+
28+
def __next__(self):
29+
raise AssertionError("physical lines must not be read through unbounded iteration")
30+
31+
def readline(self, size: int = -1) -> bytes:
32+
if size < 0:
33+
raise AssertionError("readline() must always receive a finite size")
34+
self.requested_sizes.append(size)
35+
if self.position >= len(self.data):
36+
return b""
37+
end = min(self.position + size, len(self.data))
38+
newline = self.data.find(b"\n", self.position, end)
39+
if newline >= 0:
40+
end = newline + 1
41+
chunk = self.data[self.position : end]
42+
self.position = end
43+
return chunk
44+
45+
46+
class _BlockingRule:
47+
name = "custom_block"
48+
category = "credential\nERROR forged-event"
49+
50+
def find(self, text: str) -> tuple[_DetectedMatch, ...]:
51+
if "blocked-value" not in text:
52+
return ()
53+
start = text.index("blocked-value")
54+
return (
55+
_DetectedMatch(
56+
rule_name=self.name,
57+
category=self.category,
58+
start=start,
59+
end=start + len("blocked-value"),
60+
matched="blocked-value",
61+
reason="test blocked value",
62+
),
63+
)
64+
65+
def replacement_for(self, match: _DetectedMatch, masking: object) -> str:
66+
return "[BLOCKED]"
67+
68+
69+
def _record(message: str) -> logging.LogRecord:
70+
return logging.LogRecord(
71+
name="security.marker",
72+
level=logging.WARNING,
73+
pathname=__file__,
74+
lineno=1,
75+
msg=message,
76+
args=(),
77+
exc_info=None,
78+
)
79+
80+
81+
def test_jsonl_reader_uses_bounded_readline_and_recovers_after_oversized_line() -> None:
82+
stream = _GuardedBinaryStream(b"a" * (_MAX_LINE + 1) + b"\n" + b'{"status":"ok"}\n')
83+
84+
lines = list(_iter_lines(stream))
85+
86+
assert lines[0] == (1, None, "max_line_bytes")
87+
assert lines[1] == (2, '{"status":"ok"}\n', None)
88+
assert stream.requested_sizes
89+
assert max(stream.requested_sizes) <= _MAX_LINE + 1
90+
91+
92+
def test_iter_safe_jsonl_can_replace_an_oversized_line_and_continue() -> None:
93+
stream = _GuardedBinaryStream(b"a" * (_MAX_LINE + 1) + b"\n" + b'{"status":"ok"}\n')
94+
95+
records = list(iter_safe_jsonl(stream, on_error="placeholder"))
96+
97+
assert records[0].line_number == 1
98+
assert records[0].result.cleaned["_logprivacy_error"] == "max_line_bytes"
99+
assert records[1].line_number == 2
100+
assert records[1].result.cleaned == {"status": "ok"}
101+
102+
103+
def test_clean_file_rejects_oversized_line_before_replacing_destination(
104+
tmp_path: Path,
105+
) -> None:
106+
source = tmp_path / "source.log"
107+
destination = tmp_path / "clean.log"
108+
source.write_bytes(b"a" * (_MAX_LINE + 1) + b"\n")
109+
destination.write_text("existing-safe-output", encoding="utf-8")
110+
111+
with pytest.raises(InputLimitExceededError) as caught:
112+
clean_file(source, output=destination)
113+
114+
assert caught.value.limit == "max_line_chars"
115+
assert destination.read_text(encoding="utf-8") == "existing-safe-output"
116+
117+
118+
def test_scan_file_marks_oversized_physical_line_incomplete(tmp_path: Path) -> None:
119+
source = tmp_path / "source.log"
120+
source.write_bytes(b"a" * (_MAX_LINE + 1) + b"\nnormal line\n")
121+
122+
report = scan_file(source)
123+
124+
assert report.complete is False
125+
assert report.limitations == ("max_line_chars",)
126+
127+
128+
def test_blocked_marker_escapes_control_characters_from_category() -> None:
129+
rule = _BlockingRule()
130+
policy = CleanerPolicy(
131+
rules=(rule,),
132+
block_categories=(rule.category,),
133+
)
134+
record = _record("blocked-value")
135+
136+
assert LogPrivacyFilter(cleaner=Cleaner(policy=policy)).filter(record) is True
137+
138+
rendered = record.getMessage()
139+
assert rendered.splitlines() == [rendered]
140+
assert "\n" not in rendered
141+
assert r"\x0aERROR forged-event" in rendered
142+
assert "blocked-value" not in rendered
143+
144+
145+
def test_clean_file_preserves_configured_multibyte_encoding(tmp_path: Path) -> None:
146+
source = tmp_path / "source-utf16.log"
147+
destination = tmp_path / "clean-utf16.log"
148+
source.write_text("password=secret123\nstatus=ok\n", encoding="utf-16")
149+
150+
clean_file(source, output=destination, encoding="utf-16")
151+
152+
cleaned = destination.read_text(encoding="utf-16")
153+
assert "secret123" not in cleaned
154+
assert "password=[SECRET]" in cleaned
155+
assert "status=ok" in cleaned

0 commit comments

Comments
 (0)