fix(sql): redact passwords in connection URI errors

src/iceberg/catalog/sql/catalog_store_mysql.cc

@@ -93,7 +93,8 @@ Result<std::shared_ptr<CatalogStore>> MakeMySqlCatalogStore(
     return std::make_shared<MysqlSingleCatalogStore>(options.catalog_name,
                                                      std::move(connection));
   } catch (const std::exception& e) {
-    return IOError("Failed to connect to MySQL '{}': {}", options.uri, e.what());
+    return IOError("Failed to connect to MySQL '{}': {}",
+                   RedactConnectionUri(options.uri), e.what());
   }
 }
 

src/iceberg/catalog/sql/catalog_store_postgresql.cc

@@ -83,7 +83,8 @@ Result<std::shared_ptr<CatalogStore>> MakePostgreSqlCatalogStore(
     return std::make_shared<PostgresqlSingleCatalogStore>(options.catalog_name,
                                                           std::move(connection));
   } catch (const std::exception& e) {
-    return IOError("Failed to connect to PostgreSQL '{}': {}", options.uri, e.what());
+    return IOError("Failed to connect to PostgreSQL '{}': {}",
+                   RedactConnectionUri(options.uri), e.what());
   }
 }
 

src/iceberg/catalog/sql/connection_uri.cc

@@ -50,9 +50,37 @@ Result<uint32_t> ParsePort(std::string_view port_str, std::string_view uri) {
 
 }  // namespace
 
+std::string RedactConnectionUri(std::string_view uri) {
+  // Locate the authority section, mirroring ParseConnectionUri: after
+  // "scheme://" when present, up to the first '/'.
+  size_t authority_begin = 0;
+  if (const auto scheme = uri.find("://"); scheme != std::string_view::npos) {
+    authority_begin = scheme + 3;
+  }
+  std::string_view authority = uri.substr(authority_begin);
+  if (const auto slash = authority.find('/'); slash != std::string_view::npos) {
+    authority = authority.substr(0, slash);
+  }
+  const auto at = authority.rfind('@');
+  if (at == std::string_view::npos) {
+    return std::string(uri);
+  }
+  const std::string_view userinfo = authority.substr(0, at);
+  const auto colon = userinfo.find(':');
+  if (colon == std::string_view::npos) {
+    return std::string(uri);  // no password present
+  }
+  const size_t password_begin = authority_begin + colon + 1;
+  const size_t password_end = authority_begin + at;
+  std::string redacted(uri);
+  redacted.replace(password_begin, password_end - password_begin, "***");
+  return redacted;
+}
+
 Result<ConnectionUri> ParseConnectionUri(std::string_view uri) {
   ConnectionUri result;
-  const std::string original(uri);
+  // Error messages must never echo an embedded password.
+  const std::string redacted = RedactConnectionUri(uri);
 
   const bool has_scheme = uri.find("://") != std::string_view::npos;
   if (const auto scheme = uri.find("://"); scheme != std::string_view::npos) {
@@ -67,7 +95,7 @@ Result<ConnectionUri> ParseConnectionUri(std::string_view uri) {
   }
   if (has_scheme && authority.empty()) {
     return InvalidArgument("Invalid SQL connection URI '{}': authority is empty",
-                           original);
+                           redacted);
   }
 
   // Split userinfo from host at the last '@'.
@@ -87,7 +115,7 @@ Result<ConnectionUri> ParseConnectionUri(std::string_view uri) {
   if (const auto colon = host_port.rfind(':'); colon != std::string_view::npos) {
     result.host = std::string(host_port.substr(0, colon));
     ICEBERG_ASSIGN_OR_RAISE(result.port,
-                            ParsePort(host_port.substr(colon + 1), original));
+                            ParsePort(host_port.substr(colon + 1), redacted));
   } else {
     result.host = std::string(host_port);
   }

src/iceberg/catalog/sql/connection_uri_internal.h

@@ -47,4 +47,12 @@ struct ConnectionUri {
 /// be passed unambiguously to a SQL driver is returned as `InvalidArgument`.
 Result<ConnectionUri> ParseConnectionUri(std::string_view uri);
 
+/// \brief Render a connection URI with any password replaced by "***".
+///
+/// Connection URIs may embed credentials (`scheme://user:password@host`).
+/// Use this rendering whenever a URI is interpolated into an error message
+/// or log line so the password never reaches an audience broader than the
+/// credential's. The input is returned unchanged when no password is present.
+std::string RedactConnectionUri(std::string_view uri);
+
 }  // namespace iceberg::sql

src/iceberg/test/sql_catalog_test.cc

@@ -167,6 +167,28 @@ TEST(ConnectionUriTest, RejectsInvalidPort) {
               IsError(ErrorKind::kInvalidArgument));
 }
 
+TEST(ConnectionUriTest, RedactsPassword) {
+  EXPECT_EQ(RedactConnectionUri("postgresql://alice:secret@db.example:5432/prod"),
+            "postgresql://alice:***@db.example:5432/prod");
+  // No credentials or no password: returned unchanged.
+  EXPECT_EQ(RedactConnectionUri("mysql://db.example:3306/prod"),
+            "mysql://db.example:3306/prod");
+  EXPECT_EQ(RedactConnectionUri("mysql://alice@db.example/prod"),
+            "mysql://alice@db.example/prod");
+  // Password containing '@' (last '@' delimits the host, as in ParseConnectionUri).
+  EXPECT_EQ(RedactConnectionUri("mysql://alice:p@ss@db.example/prod"),
+            "mysql://alice:***@db.example/prod");
+  // '@' in the path must not be mistaken for a userinfo delimiter.
+  EXPECT_EQ(RedactConnectionUri("mysql://db.example/we@rd"), "mysql://db.example/we@rd");
+}
+
+TEST(ConnectionUriTest, ParseErrorsDoNotEchoPassword) {
+  auto result = ParseConnectionUri("mysql://alice:secret@db.example:abc/prod");
+  ASSERT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(result, ::testing::Not(HasErrorMessage("secret")));
+  EXPECT_THAT(result, HasErrorMessage("alice:***@db.example"));
+}
+
 TEST_F(SqlCatalogTest, CatalogName) { EXPECT_EQ(catalog_->name(), "test_catalog"); }
 
 TEST_F(SqlCatalogTest, NamespaceLifecycle) {