fix: keep credentials out of the text doctor borrows from its clients

Sanitising the URL in the message was not enough: the client's own error text
went into the detail too, and both psycopg and redis-py quote the URL they were
given. The reason is kept, the endpoint is kept, and the URL, username and
password are taken out of it.

The Redis test raised a generic error, so it asserted the password was absent
without ever exercising the path that leaked. Both tests now raise what the
clients actually raise.
This commit is contained in:
Alex committed 2026-09-17 12:01:28 +01:00
1 parent 9326c6edb1
commit 18313c7823
2 files changed
+40 -6

No files matched your search

+21 -4
View File
@@ -801,7 +801,7 @@ def _check_postgres(uri: Optional[str]) -> Check:
row = cursor.fetchone()
current = row[0] if row else None
except (psycopg.Error, OSError, ValueError) as exc:
return Check("postgres", "fail", f"cannot connect: {str(exc).strip()}")
return Check("postgres", "fail", f"cannot connect to {_endpoint(uri)}: {_scrub(str(exc).strip(), uri)}")
head = _migration_head()
if not current:
return Check("postgres", "fail", f"PostgreSQL {version}, no schema yet; run `docsgpt migrate`")
@@ -811,8 +811,8 @@ def _check_postgres(uri: Optional[str]) -> Check:
return Check("postgres", "ok", f"PostgreSQL {version}, schema at {current}")
def _redis_endpoint(url: str) -> str:
"""A Redis URL without its credentials: this ends up on a terminal, in CI logs and in issues."""
def _endpoint(url: str) -> str:
"""A URL without its credentials: this ends up on a terminal, in CI logs and in issues."""
try:
parts = urlsplit(url)
except ValueError:
@@ -823,6 +823,21 @@ def _redis_endpoint(url: str) -> str:
return f"{parts.scheme}://{host}{parts.path}" if host else "the configured URL"
def _scrub(text: str, url: Optional[str]) -> str:
"""Client errors quote the URL they were handed, credentials and all, so take them back out."""
if not url:
return text
text = text.replace(url, _endpoint(url))
try:
parts = urlsplit(url)
except ValueError:
return text
for secret in (parts.password, parts.username):
if secret:
text = text.replace(secret, "...")
return text
def _check_redis(urls: Mapping[str, str]) -> Check:
"""Ping every Redis the settings name; they are usually one server, three databases."""
if not urls:
@@ -835,7 +850,9 @@ def _check_redis(urls: Mapping[str, str]) -> Check:
try:
redis.Redis.from_url(url, socket_connect_timeout=3).ping()
except Exception as exc: # noqa: BLE001 - every client error here is the same finding
return Check("redis", "fail", f"{label} ({_redis_endpoint(url)}) does not answer: {str(exc).strip()}")
return Check(
"redis", "fail", f"{label} ({_endpoint(url)}) does not answer: {_scrub(str(exc).strip(), url)}"
)
return Check("redis", "ok", f"answering on {len(urls)} database(s)")
+19 -2
View File
@@ -213,6 +213,22 @@ class TestPostgresCheck:
assert check.level == "fail"
assert "connection refused" in check.detail
def test_a_database_password_is_never_printed(self, monkeypatch):
"""psycopg quotes the connection string it was given, which carries the password."""
import psycopg
uri = "postgresql://docsgpt:hunter2@db.example.com:5432/docsgpt"
def refuse(*args, **kwargs):
raise psycopg.OperationalError(f'connection to "{uri}" failed: timeout expired')
monkeypatch.setattr(psycopg, "connect", refuse)
check = commands._check_postgres(uri)
assert check.level == "fail"
assert "hunter2" not in check.detail
assert "db.example.com:5432" in check.detail, "the endpoint still has to be identifiable"
assert "timeout expired" in check.detail, "and the reason has to survive the scrubbing"
def test_without_a_uri_at_all(self):
check = commands._check_postgres(None)
assert check.level == "fail"
@@ -251,7 +267,8 @@ class TestRedisCheck:
def from_url(cls, url, **kwargs):
class Client:
def ping(self):
raise ConnectionError("connection refused")
# redis-py quotes the URL it was handed, which is how the credentials got out.
raise ConnectionError(f"no route to {url}")
return Client()
@@ -263,7 +280,7 @@ class TestRedisCheck:
assert "redis.example.com:6380/0" in check.detail, "the endpoint still has to be identifiable"
def test_a_malformed_url_is_not_echoed_either(self):
assert commands._redis_endpoint("redis://[::1") == "the configured URL"
assert commands._endpoint("redis://[::1") == "the configured URL"
def test_without_any_redis_configured(self):
assert commands._check_redis({}).level == "fail"