129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""Unit tests for the cache client's _redact_cache_url helper."""
|
|
|
|
import pytest
|
|
|
|
from src.cache.client import (
|
|
_redact_cache_url, # pyright: ignore[reportPrivateUsage]
|
|
)
|
|
|
|
|
|
class TestRedactCacheUrl:
|
|
"""Tests for _redact_cache_url — a security-relevant logging helper
|
|
that must never raise and must never leak a password."""
|
|
|
|
# --- Password masking ---
|
|
|
|
def test_password_only_userinfo(self):
|
|
assert (
|
|
_redact_cache_url("redis://:secret@localhost:6379/0")
|
|
== "redis://:***@localhost:6379/0"
|
|
)
|
|
|
|
def test_user_and_password(self):
|
|
result = _redact_cache_url("redis://user:s3cret@10.0.0.1:6380/2")
|
|
assert "***" in result
|
|
assert "s3cret" not in result
|
|
assert "user" in result
|
|
|
|
def test_rediss_protocol(self):
|
|
result = _redact_cache_url("rediss://:secret@redis.example.com:6380")
|
|
assert result.startswith("rediss://")
|
|
assert "***" in result
|
|
assert "secret" not in result
|
|
|
|
def test_complex_password(self):
|
|
result = _redact_cache_url("redis://:p%40ssw0rd!%24@host:6379/0")
|
|
assert "***" in result
|
|
assert "p%40ssw0rd" not in result
|
|
|
|
def test_password_never_leaked(self):
|
|
"""The original password must never appear in the redacted output."""
|
|
for url in [
|
|
"redis://:hunter2@localhost:6379/0",
|
|
"redis://admin:hunter2@localhost:6379/0",
|
|
"rediss://:hunter2@[::1]:6380/1",
|
|
]:
|
|
assert "hunter2" not in _redact_cache_url(url)
|
|
|
|
# --- Secrets in query parameters ---
|
|
# redis-py accepts ?password= (querystring options become client
|
|
# kwargs) and cashews accepts ?secret= (HMAC signing key), so both
|
|
# are real configuration paths that must not reach the logs.
|
|
|
|
@pytest.mark.parametrize("param", ["password", "secret", "PASSWORD"])
|
|
def test_query_param_secret_masked(self, param: str):
|
|
result = _redact_cache_url(f"redis://host:6379/0?{param}=s3cret")
|
|
assert "s3cret" not in result
|
|
assert f"{param}=***" in result
|
|
|
|
def test_query_param_masking_preserves_other_params(self):
|
|
result = _redact_cache_url("redis://host:6379/0?db=1&password=s3cret&ssl=true")
|
|
assert "s3cret" not in result
|
|
assert "db=1" in result
|
|
assert "ssl=true" in result
|
|
|
|
def test_userinfo_and_query_secret_both_masked(self):
|
|
result = _redact_cache_url("redis://:hunter2@host:6379/0?secret=s3cret")
|
|
assert "hunter2" not in result
|
|
assert "s3cret" not in result
|
|
|
|
def test_non_secret_query_params_unchanged(self):
|
|
url = "redis://localhost:6379/0?suppress=true"
|
|
assert _redact_cache_url(url) == url
|
|
|
|
# --- No-password URLs (returned unchanged) ---
|
|
|
|
def test_user_without_password_unchanged(self):
|
|
assert (
|
|
_redact_cache_url("redis://user@localhost:6379/0")
|
|
== "redis://user@localhost:6379/0"
|
|
)
|
|
|
|
def test_in_memory_url_unchanged(self):
|
|
assert _redact_cache_url("mem://") == "mem://"
|
|
|
|
# --- IPv6 ---
|
|
|
|
def test_ipv6_brackets_preserved(self):
|
|
result = _redact_cache_url("rediss://:secret@[::1]:6380/1")
|
|
assert "[::1]" in result
|
|
assert "***" in result
|
|
assert "secret" not in result
|
|
|
|
# --- Malformed URLs (must NOT raise) ---
|
|
|
|
def test_invalid_port_redacts_password(self):
|
|
"""Regression test for two review findings: accessing
|
|
``parsed.port`` on a URL with a non-numeric port raises
|
|
``ValueError`` (must not crash startup inside an except block),
|
|
and the fallback must never echo the raw URL back — the
|
|
password has to be masked even when the port is unparseable.
|
|
"""
|
|
result = _redact_cache_url("redis://:pass@host:notaport/0")
|
|
assert "pass" not in result
|
|
assert "***" in result
|
|
|
|
def test_out_of_range_port_redacts_password(self):
|
|
result = _redact_cache_url("redis://:supersecret@host:99999/0")
|
|
assert "supersecret" not in result
|
|
assert "***" in result
|
|
|
|
def test_unparseable_url_never_echoed(self):
|
|
# Unbalanced IPv6 bracket makes urlparse itself raise; the
|
|
# fallback must return a placeholder, not the raw input.
|
|
result = _redact_cache_url("redis://:secret@[::1:6379/0")
|
|
assert "secret" not in result
|
|
|
|
def test_missing_scheme_never_echoed(self):
|
|
# Without "redis://" urlparse sees no netloc, so the userinfo
|
|
# (and its password) is invisible to .password — the string
|
|
# must not be echoed back.
|
|
result = _redact_cache_url(":hunter2@host:6379/0")
|
|
assert "hunter2" not in result
|
|
|
|
def test_garbage_input_does_not_raise(self):
|
|
assert isinstance(_redact_cache_url("not a url at all"), str)
|
|
|
|
def test_empty_string_does_not_raise(self):
|
|
assert isinstance(_redact_cache_url(""), str)
|