Sanitize DB urls before printing

quetz/cli.py

@@ -29,7 +29,7 @@
     configure_logger,
     create_config,
 )
-from quetz.database import get_session
+from quetz.database import get_session, sanitize_db_url
 from quetz.db_models import (
     ApiKey,
     Channel,
@@ -315,7 +315,7 @@ def _is_deployment(base_dir: Path):
             if not database_exists(config.sqlalchemy_database_url):
                 logger.error(
                     "Cannot verify that specified database exists. "
-                    + config.sqlalchemy_database_url
+                    + sanitize_db_url(config.sqlalchemy_database_url)
                 )
                 return False
             else:

quetz/database.py

@@ -1,11 +1,14 @@
 # Copyright 2020 QuantStack
 # Distributed under the terms of the Modified BSD License.
 import logging
+import re
 from contextlib import contextmanager
 from typing import Callable
 
 from sqlalchemy import create_engine, event
 from sqlalchemy.engine import Engine
+from sqlalchemy.engine.url import make_url
+from sqlalchemy.exc import ArgumentError
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.orm.session import Session
 from sqlalchemy.pool import StaticPool
@@ -83,3 +86,32 @@ def get_db_manager():
         yield db
     finally:
         db.close()
+
+
+def sanitize_db_url(db_url: str) -> str:
+    """
+    Sanitizes the DB url so it is safe to print.
+
+    If the URL is parseable with sqlalchemy's make_url,
+    it is parsed and the password is replaced.
+    If not, we try to replace everything between ":" and "@",
+    if those characters are present.
+
+    If neither method succeeds, we give up and return the
+    full initial URL.
+    """
+
+    # Attempt 1: Actual parsing, this is ideal but may fail
+    try:
+        parsed_url = make_url(db_url)
+        if parsed_url.password:
+            return db_url.replace(parsed_url.password, "***")
+    except ArgumentError:
+        pass
+
+    # Attempt 2: Poor man's parsing: Just replacing everything between ":" and "@"
+    if ":" in db_url and "@" in db_url:
+        return re.sub(":[^:@]*@", ":***@", db_url)
+
+    # Fallback: We don't understand the URL format, so we do nothing
+    return db_url

quetz/tests/test_database.py

@@ -0,0 +1,22 @@
+import pytest
+
+from quetz.database import sanitize_db_url
+
+
+@pytest.mark.parametrize(
+    "input_url,expected_output_url",
+    (
+        (
+            "sqlite:///./quetz.sqlite",
+            "sqlite:///./quetz.sqlite",
+        ),  # No password, no effect
+        (
+            "postgresql+psycopg2://postgres_user:postgres_password@localhost:5432/postgres",  # noqa: E501
+            "postgresql+psycopg2://postgres_user:***@localhost:5432/postgres",
+        ),
+        ("A:B@C:1111/DB", "A:***@C:1111/DB"),
+        ("THISISNOTAURL", "THISISNOTAURL"),
+    ),
+)
+def test_sanitize_db_url(input_url, expected_output_url):
+    assert sanitize_db_url(input_url) == expected_output_url