fix: preserve # in pg_service.conf passwords

pgcli/main.py

@@ -1,23 +1,21 @@
-from zoneinfo import ZoneInfoNotFoundError
-from configobj import ConfigObj, ParseError
-from pgspecial.namedqueries import NamedQueries
-from .config import skip_initial_comment
-
 import atexit
+import configparser
+import datetime as dt
+import functools
+import io
+import itertools
 import os
 import re
 import sys
 import traceback
 import logging
 import threading
 import shutil
-import functools
-import datetime as dt
-import itertools
 import pathlib
 import platform
 from time import time, sleep
 from typing import Optional
+from zoneinfo import ZoneInfoNotFoundError
 
 from cli_helpers.tabular_output import TabularOutputFormatter
 from cli_helpers.tabular_output.preprocessors import (
@@ -29,6 +27,7 @@
 from .explain_output_formatter import ExplainOutputFormatter
 import click
 import tzlocal
+from pgspecial.namedqueries import NamedQueries
 
 try:
     import setproctitle
@@ -68,6 +67,7 @@
     get_config,
     get_config_filename,
 )
+from .config import skip_initial_comment
 from .key_bindings import pgcli_bindings
 from .packages.formatter.sqlformatter import register_new_formatter
 from .packages.prompt_utils import confirm, confirm_destructive_query
@@ -1973,14 +1973,11 @@ def parse_service_info(service):
         return None, service_file
     with open(service_file, newline="") as f:
         skipped_lines = skip_initial_comment(f)
-        try:
-            service_file_config = ConfigObj(f)
-        except ParseError as err:
-            err.line_number += skipped_lines
-            raise err
+        service_file_config = configparser.ConfigParser(interpolation=None)
+        service_file_config.read_file(io.StringIO("\n" * skipped_lines + f.read()), source=service_file)
     if service not in service_file_config:
         return None, service_file
-    service_conf = service_file_config.get(service)
+    service_conf = dict(service_file_config[service])
     return service_conf, service_file
 
 

tests/test_main.py

@@ -504,6 +504,35 @@ def test_pg_service_file(tmpdir):
     del os.environ["PGSERVICEFILE"]
 
 
+def test_pg_service_file_password_with_hash(tmpdir):
+    with mock.patch.object(PGCli, "connect") as mock_connect:
+        cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))
+        with open(tmpdir.join(".pg_service.conf").strpath, "w") as service_conf:
+            service_conf.write(
+                """File begins with a comment
+            that is not a comment
+
+            [myservice]
+            host=a_host
+            user=a_user
+            port=5433
+            password=abc#123
+            dbname=a_dbname
+            """
+            )
+        os.environ["PGSERVICEFILE"] = tmpdir.join(".pg_service.conf").strpath
+        cli.connect_service("myservice", None)
+        mock_connect.assert_called_with(
+            database="a_dbname",
+            host="a_host",
+            user="a_user",
+            port="5433",
+            passwd="abc#123",
+        )
+
+    del os.environ["PGSERVICEFILE"]
+
+
 def test_ssl_db_uri(tmpdir):
     with mock.patch.object(PGCli, "connect") as mock_connect:
         cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))