fix: Redact auth from DB url in status check (#431)

At present, the status check logs the DB URL, including the password.

This PR uses a helper function from pip to redact the password from the
URLs, making them safe to log.

After this change, this is what the log messages look like:
```
ERROR:cool_seq_tool.resources.status:Encountered error instantiating UTA at URI postgresql://uta_admin:****@localhost:5432/uta/uta_20241220
```

Closes #429

---------

Co-authored-by: Kori Kuzma <[email protected]>
Co-authored-by: jarbesfeld <[email protected]>
Co-authored-by: Katie Stahl <[email protected]>

Author: 4 people

src/cool_seq_tool/resources/status.py

@@ -3,6 +3,7 @@
 import logging
 from collections import namedtuple
 from pathlib import Path
+from urllib.parse import urlparse
 
 from agct._core import ChainfileError
 from asyncpg import InvalidCatalogNameError, UndefinedTableError
@@ -11,7 +12,7 @@
 from cool_seq_tool.handlers.seqrepo_access import SEQREPO_ROOT_DIR, SeqRepoAccess
 from cool_seq_tool.mappers.liftover import LiftOver
 from cool_seq_tool.resources.data_files import DataFile, get_data_file
-from cool_seq_tool.sources.uta_database import UTA_DB_URL, UtaDatabase
+from cool_seq_tool.sources.uta_database import UTA_DB_URL, ParseResult, UtaDatabase
 
 _logger = logging.getLogger(__name__)
 
@@ -119,14 +120,20 @@ async def check_status(
     else:
         status["liftover"] = True
 
+    parsed_result = ParseResult(urlparse(db_url))
+    sanitized_url = parsed_result.sanitized_url
     try:
         await UtaDatabase.create(db_url)
+    except ValueError:
+        _logger.exception("Database URL is not valid")
     except (OSError, InvalidCatalogNameError, UndefinedTableError):
-        _logger.exception("Encountered error instantiating UTA at URI %s", UTA_DB_URL)
+        _logger.exception(
+            "Encountered error instantiating UTA at URI %s", sanitized_url
+        )
     except Exception as e:
         _logger.critical(
             "Encountered unexpected error instantiating UTA from URI %s: %s",
-            UTA_DB_URL,
+            sanitized_url,
             e,
         )
     else:

src/cool_seq_tool/sources/uta_database.py

@@ -5,7 +5,7 @@
 from os import environ
 from typing import Any, Literal, TypeVar
 from urllib.parse import ParseResult as UrlLibParseResult
-from urllib.parse import unquote, urlparse
+from urllib.parse import unquote, urlparse, urlunparse
 
 import asyncpg
 import boto3
@@ -953,3 +953,28 @@ def schema(self) -> str | None:
         """Create schema property."""
         path_elems = self.path.split("/")
         return path_elems[2] if len(path_elems) > 2 else None
+
+    @property
+    def sanitized_url(self) -> str:
+        """Sanitized DB URL with the password masked"""
+        netloc = ""
+        if self.username:
+            netloc += self.username
+            if self.password is not None and self.password != "":
+                netloc += ":***"
+            netloc += "@"
+        if self.hostname:
+            netloc += f"{self.hostname}"
+        if self.port:
+            netloc += f":{self.port}"
+
+        return urlunparse(
+            (
+                self.scheme,
+                netloc,
+                self.path,
+                self.params,
+                self.query,
+                self.fragment,
+            )
+        )

tests/sources/test_uta_database.py

@@ -1,11 +1,14 @@
 """Test UTA data source."""
 
+from urllib.parse import urlparse
+
 import pytest
 
 from cool_seq_tool.schemas import Strand
 from cool_seq_tool.sources.uta_database import (
     GenomicTxData,
     GenomicTxMetadata,
+    ParseResult,
     TxExonAlnData,
 )
 
@@ -360,3 +363,98 @@ async def test_get_mane_transcripts_from_genomic_pos(test_db):
     # invalid ac
     resp = await test_db.get_transcripts_from_genomic_pos("NC_000007.14232", 140753336)
     assert resp == []
+
+
+@pytest.mark.parametrize(
+    ("raw_url", "expected"),
+    [
+        # Username + password
+        (
+            "postgresql://user:pass@localhost:5432/dbname",
+            {
+                "scheme": "postgresql",
+                "username": "user",
+                "password": "pass",
+                "hostname": "localhost",
+                "port": 5432,
+                "database": "dbname",
+                "sanitized_url": "postgresql://user:***@localhost:5432/dbname",
+            },
+        ),
+        # Username with null password
+        (
+            "postgresql://user@localhost/dbname",
+            {
+                "scheme": "postgresql",
+                "username": "user",
+                "password": None,
+                "hostname": "localhost",
+                "port": None,
+                "database": "dbname",
+                "sanitized_url": "postgresql://user@localhost/dbname",
+            },
+        ),
+        # Password is "0"
+        (
+            "postgresql://user:0@localhost/dbname",
+            {
+                "scheme": "postgresql",
+                "username": "user",
+                "password": "0",
+                "hostname": "localhost",
+                "port": None,
+                "database": "dbname",
+                "sanitized_url": "postgresql://user:***@localhost/dbname",
+            },
+        ),
+        # Empty password
+        (
+            "postgresql://user:@localhost/dbname",
+            {
+                "scheme": "postgresql",
+                "username": "user",
+                "password": "",
+                "hostname": "localhost",
+                "port": None,
+                "database": "dbname",
+                "sanitized_url": "postgresql://user@localhost/dbname",
+            },
+        ),
+        # No username
+        (
+            "postgresql://localhost:5432/dbname",
+            {
+                "scheme": "postgresql",
+                "username": None,
+                "password": None,
+                "hostname": "localhost",
+                "port": 5432,
+                "database": "dbname",
+                "sanitized_url": "postgresql://localhost:5432/dbname",
+            },
+        ),
+        # With query params
+        (
+            "postgresql://user:secret@localhost/dbname?query#fragment",
+            {
+                "scheme": "postgresql",
+                "username": "user",
+                "password": "secret",
+                "hostname": "localhost",
+                "port": None,
+                "database": "dbname",
+                "sanitized_url": "postgresql://user:***@localhost/dbname?query#fragment",
+            },
+        ),
+    ],
+)
+async def test_parsed_url(raw_url, expected):
+    parsed_result = ParseResult(urlparse(raw_url))
+
+    assert parsed_result.scheme == expected["scheme"]
+    assert parsed_result.username == expected["username"]
+    assert parsed_result.password == expected["password"]
+    assert parsed_result.hostname == expected["hostname"]
+    assert parsed_result.port == expected["port"]
+    assert parsed_result.database == expected["database"]
+    assert parsed_result.sanitized_url == expected["sanitized_url"]