mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-05 13:41:36 +02:00
maziggy
e325948dcc
Subprocess output and user-supplied URLs are scrubbed of credentials before they reach the application log. Adds a shared redaction helper in core/logging_filters and routes the existing support-bundle sanitizer through the same pattern.
113 lines
5.0 KiB
Python
113 lines
5.0 KiB
Python
"""Credentials must never reach bambuddy.log.
|
|
|
|
Subprocesses echo their input URL back at us: ffmpeg prints the RTSP input in
|
|
its ``Input #0`` line, so logging its stderr verbatim published the printer
|
|
access code (or an external camera's password) into the log file — which users
|
|
routinely attach to public GitHub issues.
|
|
|
|
These cover the shared helper plus the two funnels that carry subprocess output
|
|
into the log.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from backend.app.api.routes.camera import _read_ffmpeg_stderr, _summarize_ffmpeg_stderr
|
|
from backend.app.core.logging_filters import redact_url_credentials
|
|
from backend.app.services.log_reader import sanitize_log_content
|
|
|
|
# What ffmpeg actually prints for the camera's local TLS-proxy input. The
|
|
# access code sits in the userinfo of the URL it quotes back.
|
|
FFMPEG_INPUT_LINE = "Input #0, rtsp, from 'rtsp://bblp:38A4KQ2P@127.0.0.1:48521/streaming/live/1':"
|
|
|
|
|
|
class TestRedactUrlCredentials:
|
|
def test_masks_the_printer_access_code(self):
|
|
result = redact_url_credentials(FFMPEG_INPUT_LINE)
|
|
assert "38A4KQ2P" not in result
|
|
assert result == "Input #0, rtsp, from 'rtsp://bblp:[REDACTED]@127.0.0.1:48521/streaming/live/1':"
|
|
|
|
def test_keeps_everything_that_is_not_the_secret(self):
|
|
"""Host, port, path and username stay — the line has to remain diagnosable."""
|
|
result = redact_url_credentials("rtsp://admin:hunter2@192.168.1.50:554/stream1")
|
|
assert result == "rtsp://admin:[REDACTED]@192.168.1.50:554/stream1"
|
|
|
|
def test_masks_every_scheme_not_just_the_ones_we_use_today(self):
|
|
for url, expected in (
|
|
("http://user:pw@cam.local/snapshot", "http://user:[REDACTED]@cam.local/snapshot"),
|
|
("https://user:pw@cam.local/snapshot", "https://user:[REDACTED]@cam.local/snapshot"),
|
|
("rtsps://bblp:code@printer:322/streaming/live/1", "rtsps://bblp:[REDACTED]@printer:322/streaming/live/1"),
|
|
("ftp://bblp:code@printer:990/", "ftp://bblp:[REDACTED]@printer:990/"),
|
|
):
|
|
assert redact_url_credentials(url) == expected
|
|
|
|
def test_masks_a_password_containing_an_at_sign(self):
|
|
"""The userinfo ends at the LAST @ before the path — no tail may survive."""
|
|
result = redact_url_credentials("rtsp://admin:p@ssw0rd@192.168.1.50/stream")
|
|
assert result == "rtsp://admin:[REDACTED]@192.168.1.50/stream"
|
|
assert "ssw0rd" not in result
|
|
|
|
def test_masks_several_urls_in_one_blob(self):
|
|
text = "first rtsp://bblp:AAAAAAAA@10.0.0.1/live then rtsp://bblp:BBBBBBBB@10.0.0.2/live"
|
|
result = redact_url_credentials(text)
|
|
assert "AAAAAAAA" not in result
|
|
assert "BBBBBBBB" not in result
|
|
assert result.count("[REDACTED]") == 2
|
|
|
|
def test_never_runs_past_the_authority_into_the_path(self):
|
|
"""A later @ in the path must not drag the host into the mask."""
|
|
result = redact_url_credentials("rtsp://bblp:code@10.0.0.1/live/user@example")
|
|
assert result == "rtsp://bblp:[REDACTED]@10.0.0.1/live/user@example"
|
|
|
|
def test_leaves_credential_free_text_alone(self):
|
|
for untouched in (
|
|
"Connection refused",
|
|
"rtsp://10.0.0.1:554/stream1",
|
|
"mailto and user@example.com in prose",
|
|
"Starting USB camera stream from /dev/video0 at 10 fps",
|
|
):
|
|
assert redact_url_credentials(untouched) == untouched
|
|
|
|
def test_tolerates_empty_and_none(self):
|
|
assert redact_url_credentials("") == ""
|
|
assert redact_url_credentials(None) is None
|
|
|
|
|
|
class TestFfmpegStderrFunnel:
|
|
"""`_summarize_ffmpeg_stderr` is the one funnel every stderr log in the
|
|
camera route passes through, so redaction lands there."""
|
|
|
|
def test_summary_strips_the_access_code(self):
|
|
stderr = f"{FFMPEG_INPUT_LINE}\n[rtsp @ 0x5] Could not find codec parameters\n"
|
|
result = _summarize_ffmpeg_stderr(stderr)
|
|
assert "38A4KQ2P" not in result
|
|
assert "[REDACTED]" in result
|
|
# The actionable error is untouched.
|
|
assert "Could not find codec parameters" in result
|
|
|
|
def test_incremental_reader_strips_the_access_code(self):
|
|
async def run():
|
|
reader = asyncio.StreamReader()
|
|
reader.feed_data(f"{FFMPEG_INPUT_LINE}\nError opening input: Connection refused\n".encode())
|
|
reader.feed_eof()
|
|
|
|
class _FakeProcess:
|
|
stderr = reader
|
|
|
|
return await _read_ffmpeg_stderr(_FakeProcess())
|
|
|
|
result = asyncio.run(run())
|
|
assert result is not None
|
|
assert "38A4KQ2P" not in result
|
|
assert "Connection refused" in result
|
|
|
|
|
|
class TestSupportBundleSanitizerUnchanged:
|
|
"""The bundle sanitizer shares the pattern but keeps its own, stricter
|
|
replacement — it drops the username too. Guard against drift."""
|
|
|
|
def test_bundle_still_drops_the_whole_userinfo(self):
|
|
result = sanitize_log_content("rtsp://bblp:38A4KQ2P@10.0.0.1/live")
|
|
assert "38A4KQ2P" not in result
|
|
assert "bblp" not in result
|
|
assert "[CREDENTIALS]@" in result
|