Fix PromQL injection, connection leaks, and missing timeouts

Addresses critical findings from quality framework audit:

- Add _escape_promql_label() and _promql_filter() to reporter to prevent
  PromQL injection via cluster/node/database/index names (FM-4)
- Apply escaping to H001 base_filter, H002 idx_scan query, and H004
  redundant index queries (the most dangerous sites where DB metadata
  like index_name/table_name is interpolated)
- Add escape_promql_label() to Flask backend and apply to filter building
- Fix connection leak in CLI: 2 locations in postgres-ai.ts where
  Client.connect() had no finally block (mon targets add, interactive add)
- Add connectionTimeoutMillis: 10000 to all Client() instances in CLI
- Add connect_timeout=10 to all psycopg2.connect() calls (reporter + Flask)
- Add 11 unit tests for PromQL escaping covering injection attempts,
  backslash/quote handling, and normal PostgreSQL identifiers

https://claude.ai/code/session_01TKKnEc2Yn2zM64bwCJ2UaX

Author: claude

cli/bin/postgres-ai.ts

@@ -2452,16 +2452,20 @@ mon
 
         // Test connection
         console.log("Testing connection to the added instance...");
-        try {
-          const client = new Client({ connectionString: connStr });
-          await client.connect();
-          const result = await client.query("select version();");
-          console.log("✓ Connection successful");
-          console.log(`${result.rows[0].version}\n`);
-          await client.end();
-        } catch (error) {
-          const message = error instanceof Error ? error.message : String(error);
-          console.error(`✗ Connection failed: ${message}\n`);
+        {
+          let testClient: InstanceType<typeof Client> | null = null;
+          try {
+            testClient = new Client({ connectionString: connStr, connectionTimeoutMillis: 10000 });
+            await testClient.connect();
+            const result = await testClient.query("select version();");
+            console.log("✓ Connection successful");
+            console.log(`${result.rows[0].version}\n`);
+          } catch (error) {
+            const message = error instanceof Error ? error.message : String(error);
+            console.error(`✗ Connection failed: ${message}\n`);
+          } finally {
+            if (testClient) await testClient.end();
+          }
         }
       } else if (opts.yes) {
         // Auto-yes mode without database URL - skip database setup
@@ -2496,16 +2500,20 @@ mon
 
               // Test connection
               console.log("Testing connection to the added instance...");
-              try {
-                const client = new Client({ connectionString: connStr });
-                await client.connect();
-                const result = await client.query("select version();");
-                console.log("✓ Connection successful");
-                console.log(`${result.rows[0].version}\n`);
-                await client.end();
-              } catch (error) {
-                const message = error instanceof Error ? error.message : String(error);
-                console.error(`✗ Connection failed: ${message}\n`);
+              {
+                let testClient: InstanceType<typeof Client> | null = null;
+                try {
+                  testClient = new Client({ connectionString: connStr, connectionTimeoutMillis: 10000 });
+                  await testClient.connect();
+                  const result = await testClient.query("select version();");
+                  console.log("✓ Connection successful");
+                  console.log(`${result.rows[0].version}\n`);
+                } catch (error) {
+                  const message = error instanceof Error ? error.message : String(error);
+                  console.error(`✗ Connection failed: ${message}\n`);
+                } finally {
+                  if (testClient) await testClient.end();
+                }
               }
             }
           } else {
@@ -3292,7 +3300,7 @@ targets
       console.log(`Testing connection to monitoring target '${name}'...`);
 
       // Use native pg client instead of requiring psql to be installed
-      const client = new Client({ connectionString: instance.conn_str });
+      const client = new Client({ connectionString: instance.conn_str, connectionTimeoutMillis: 10000 });
 
       try {
         await client.connect();

monitoring_flask_backend/app.py

@@ -16,6 +16,11 @@
 logger = logging.getLogger(__name__)
 
 
+def escape_promql_label(value: str) -> str:
+    """Escape a value for safe use inside PromQL label matchers (double-quoted strings)."""
+    return value.replace("\\", "\\\\").replace('"', '\\"')
+
+
 def smart_truncate_query(query: str, max_length: int = 40) -> str:
     """
     Smart SQL query truncation for display names.
@@ -250,7 +255,7 @@ def get_query_texts_from_sink(db_name: str = None, truncation_mode: str = 'smart
 
     conn = None
     try:
-        conn = psycopg2.connect(POSTGRES_SINK_URL)
+        conn = psycopg2.connect(POSTGRES_SINK_URL, connect_timeout=10)
         with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
             # Skip db_name filter if it's empty, "All", or contains special chars
             use_db_filter = db_name and db_name.lower() not in ('all', '') and not db_name.startswith('$')
@@ -384,14 +389,14 @@ def get_pgss_metrics_csv():
         # Build the base query for pg_stat_statements metrics
         base_query = 'pgwatch_pg_stat_statements_calls'
 
-        # Add filters if provided
+        # Add filters if provided (escape values to prevent PromQL injection)
         filters = []
         if cluster_name:
-            filters.append(f'cluster="{cluster_name}"')
+            filters.append(f'cluster="{escape_promql_label(cluster_name)}"')
         if node_name:
-            filters.append(f'instance=~".*{node_name}.*"')
+            filters.append(f'instance=~".*{escape_promql_label(node_name)}.*"')
         if db_name:
-            filters.append(f'datname="{db_name}"')
+            filters.append(f'datname="{escape_promql_label(db_name)}"')
 
         if filters:
             base_query += '{' + ','.join(filters) + '}'
@@ -1176,7 +1181,7 @@ def get_query_texts():
 
         conn = None
         try:
-            conn = psycopg2.connect(POSTGRES_SINK_URL)
+            conn = psycopg2.connect(POSTGRES_SINK_URL, connect_timeout=10)
             with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
                 # Skip db_name filter if it's empty, "All", or contains special chars
                 use_db_filter = db_name and db_name.lower() not in ('all', '') and not db_name.startswith('$')
@@ -1287,7 +1292,7 @@ def get_query_info_metrics():
 
         conn = None
         try:
-            conn = psycopg2.connect(POSTGRES_SINK_URL)
+            conn = psycopg2.connect(POSTGRES_SINK_URL, connect_timeout=10)
             with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
                 # Skip db_name filter if it's empty, "All", or contains special chars
                 use_db_filter = db_name and db_name.lower() not in ('all', '') and not db_name.startswith('$')

reporter/postgres_reports.py

@@ -190,6 +190,29 @@ def _load_build_metadata(self) -> Dict[str, Optional[str]]:
             "build_ts": self._read_text_file(build_ts_path),
         }
 
+    @staticmethod
+    def _escape_promql_label(value: str) -> str:
+        """Escape a value for use inside PromQL label matchers.
+
+        PromQL label values are enclosed in double quotes and only need
+        backslash and double-quote characters escaped (like Go strings).
+        This prevents PromQL injection when cluster names, database names,
+        or other user-controlled strings contain special characters.
+        """
+        return value.replace("\\", "\\\\").replace('"', '\\"')
+
+    def _promql_filter(self, **labels: str) -> str:
+        """Build a PromQL label filter string from keyword arguments.
+
+        Example:
+            self._promql_filter(cluster="my-cluster", node_name="node-01")
+            # returns: '{cluster="my-cluster", node_name="node-01"}'
+        """
+        parts = []
+        for key, val in labels.items():
+            parts.append(f'{key}="{self._escape_promql_label(val)}"')
+        return "{" + ", ".join(parts) + "}"
+
     def test_connection(self) -> bool:
         """Test connection to Prometheus."""
         try:
@@ -205,9 +228,9 @@ def connect_postgres_sink(self) -> bool:
             return False
         if psycopg2 is None:
             raise RuntimeError("psycopg2 is required for postgres sink access but is not installed")
-        
+
         try:
-            self.pg_conn = psycopg2.connect(self.postgres_sink_url)
+            self.pg_conn = psycopg2.connect(self.postgres_sink_url, connect_timeout=10)
             return True
         except Exception as e:
             logger.error(f"Postgres sink connection failed: {e}")
@@ -670,7 +693,8 @@ def generate_h001_invalid_indexes_report(self, cluster: str = "local", node_name
 
             # Query all invalid indexes metrics and merge by index key
             # Each field is a separate metric in pgwatch prometheus export
-            base_filter = f'cluster="{cluster}", node_name="{node_name}", datname="{db_name}"'
+            _esc = self._escape_promql_label
+            base_filter = f'cluster="{_esc(cluster)}", node_name="{_esc(node_name)}", datname="{_esc(db_name)}"'
 
             # Query primary metric (index_size_bytes) - this determines which indexes exist
             size_query = f'last_over_time(pgwatch_pg_invalid_indexes_index_size_bytes{{{base_filter}}}[3h])'
@@ -831,7 +855,8 @@ def generate_h002_unused_indexes_report(self, cluster: str = "local", node_name:
                     index_size_bytes = float(item['value'][1]) if item.get('value') else 0
 
                     # Query other related metrics for this index
-                    idx_scan_query = f'last_over_time(pgwatch_unused_indexes_idx_scan{{cluster="{cluster}", node_name="{node_name}", datname="{db_name}", schema_name="{schema_name}", table_name="{table_name}", index_name="{index_name}"}}[3h])'
+                    _e = self._escape_promql_label
+                    idx_scan_query = f'last_over_time(pgwatch_unused_indexes_idx_scan{{cluster="{_e(cluster)}", node_name="{_e(node_name)}", datname="{_e(db_name)}", schema_name="{_e(schema_name)}", table_name="{_e(table_name)}", index_name="{_e(index_name)}"}}[3h])'
                     idx_scan_result = self.query_instant(idx_scan_query)
                     idx_scan = float(idx_scan_result['data']['result'][0]['value'][1]) if idx_scan_result.get('data',
                                                                                                               {}).get(
@@ -941,18 +966,20 @@ def generate_h004_redundant_indexes_report(self, cluster: str = "local", node_na
                     index_size_bytes = float(item['value'][1]) if item.get('value') else 0
 
                     # Query other related metrics for this index
-                    table_size_query = f'last_over_time(pgwatch_redundant_indexes_table_size_bytes{{cluster="{cluster}", node_name="{node_name}", dbname="{db_name}", schema_name="{schema_name}", table_name="{table_name}", index_name="{index_name}"}}[3h])'
+                    _e = self._escape_promql_label
+                    _idx_filter = f'cluster="{_e(cluster)}", node_name="{_e(node_name)}", dbname="{_e(db_name)}", schema_name="{_e(schema_name)}", table_name="{_e(table_name)}", index_name="{_e(index_name)}"'
+                    table_size_query = f'last_over_time(pgwatch_redundant_indexes_table_size_bytes{{{_idx_filter}}}[3h])'
                     table_size_result = self.query_instant(table_size_query)
                     table_size_bytes = float(
                         table_size_result['data']['result'][0]['value'][1]) if table_size_result.get('data', {}).get(
                         'result') else 0
 
-                    index_usage_query = f'last_over_time(pgwatch_redundant_indexes_index_usage{{cluster="{cluster}", node_name="{node_name}", dbname="{db_name}", schema_name="{schema_name}", table_name="{table_name}", index_name="{index_name}"}}[3h])'
+                    index_usage_query = f'last_over_time(pgwatch_redundant_indexes_index_usage{{{_idx_filter}}}[3h])'
                     index_usage_result = self.query_instant(index_usage_query)
                     index_usage = float(index_usage_result['data']['result'][0]['value'][1]) if index_usage_result.get(
                         'data', {}).get('result') else 0
 
-                    supports_fk_query = f'last_over_time(pgwatch_redundant_indexes_supports_fk{{cluster="{cluster}", node_name="{node_name}", dbname="{db_name}", schema_name="{schema_name}", table_name="{table_name}", index_name="{index_name}"}}[3h])'
+                    supports_fk_query = f'last_over_time(pgwatch_redundant_indexes_supports_fk{{{_idx_filter}}}[3h])'
                     supports_fk_result = self.query_instant(supports_fk_query)
                     supports_fk = bool(
                         int(supports_fk_result['data']['result'][0]['value'][1])) if supports_fk_result.get('data',

tests/reporter/test_promql_escaping.py

@@ -0,0 +1,92 @@
+"""Tests for PromQL label escaping and query filter building.
+
+These tests verify that user-controlled values (cluster names, database names,
+index names, etc.) are properly escaped when interpolated into PromQL queries,
+preventing PromQL injection attacks.
+"""
+import pytest
+
+from reporter.postgres_reports import PostgresReportGenerator
+
+
+@pytest.fixture
+def generator():
+    """Create a generator instance for testing."""
+    return PostgresReportGenerator(
+        prometheus_url="http://prom.test",
+        postgres_sink_url="",
+    )
+
+
+class TestEscapePromqlLabel:
+    """Tests for _escape_promql_label static method."""
+
+    @pytest.mark.unit
+    def test_plain_string_unchanged(self):
+        assert PostgresReportGenerator._escape_promql_label("my-cluster") == "my-cluster"
+
+    @pytest.mark.unit
+    def test_escapes_double_quotes(self):
+        assert PostgresReportGenerator._escape_promql_label('db"name') == 'db\\"name'
+
+    @pytest.mark.unit
+    def test_escapes_backslashes(self):
+        assert PostgresReportGenerator._escape_promql_label("path\\to") == "path\\\\to"
+
+    @pytest.mark.unit
+    def test_escapes_backslash_before_quote(self):
+        """Backslash must be escaped first, then quote."""
+        result = PostgresReportGenerator._escape_promql_label('a\\"b')
+        assert result == 'a\\\\\\"b'
+
+    @pytest.mark.unit
+    def test_empty_string(self):
+        assert PostgresReportGenerator._escape_promql_label("") == ""
+
+    @pytest.mark.unit
+    def test_injection_attempt_closing_brace(self):
+        """A value like: db"}} OR vector(1) should not break out of the label."""
+        result = PostgresReportGenerator._escape_promql_label('db"}} OR vector(1)')
+        assert '"' not in result or result.count('\\"') == result.count('"')
+        assert result == 'db\\"}} OR vector(1)'
+
+    @pytest.mark.unit
+    def test_normal_postgres_identifiers(self):
+        """Common PostgreSQL identifiers should pass through unchanged."""
+        identifiers = [
+            "public",
+            "my_table",
+            "idx_users_email",
+            "pg_stat_statements",
+            "node-01",
+            "cluster.local",
+        ]
+        for ident in identifiers:
+            assert PostgresReportGenerator._escape_promql_label(ident) == ident
+
+
+class TestPromqlFilter:
+    """Tests for _promql_filter method."""
+
+    @pytest.mark.unit
+    def test_single_label(self, generator):
+        result = generator._promql_filter(cluster="local")
+        assert result == '{cluster="local"}'
+
+    @pytest.mark.unit
+    def test_multiple_labels(self, generator):
+        result = generator._promql_filter(cluster="local", node_name="node-01")
+        assert 'cluster="local"' in result
+        assert 'node_name="node-01"' in result
+        assert result.startswith("{")
+        assert result.endswith("}")
+
+    @pytest.mark.unit
+    def test_escapes_values(self, generator):
+        result = generator._promql_filter(cluster='my"cluster')
+        assert result == '{cluster="my\\"cluster"}'
+
+    @pytest.mark.unit
+    def test_empty_labels(self, generator):
+        result = generator._promql_filter()
+        assert result == "{}"