sql: redact inline credentials in CREATE CONNECTION options

src/sql-parser/src/ast/defs/ddl.rs

@@ -908,6 +908,26 @@ pub enum ConnectionOptionName {
     Warehouse,
 }
 
+impl ConnectionOptionName {
+    pub(crate) fn value_contains_sensitive_data(&self) -> bool {
+        matches!(
+            self,
+            ConnectionOptionName::AccessKeyId
+                | ConnectionOptionName::Credential
+                | ConnectionOptionName::Password
+                | ConnectionOptionName::SaslPassword
+                | ConnectionOptionName::SaslUsername
+                | ConnectionOptionName::SecretAccessKey
+                | ConnectionOptionName::ServiceAccountKey
+                | ConnectionOptionName::SessionToken
+                | ConnectionOptionName::SslCertificate
+                | ConnectionOptionName::SslCertificateAuthority
+                | ConnectionOptionName::SslKey
+                | ConnectionOptionName::User
+        )
+    }
+}
+
 impl AstDisplay for ConnectionOptionName {
     fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
         f.write_str(match self {
@@ -964,47 +984,11 @@ impl WithOptionName for ConnectionOptionName {
     /// this value could contain sensitive user data. If you're uncertain, err
     /// on the conservative side and return `true`.
     fn redact_value(&self) -> bool {
-        match self {
-            ConnectionOptionName::AccessKeyId
-            | ConnectionOptionName::AvailabilityZones
-            | ConnectionOptionName::AwsConnection
-            | ConnectionOptionName::AwsPrivatelink
-            | ConnectionOptionName::Broker
-            | ConnectionOptionName::Brokers
-            | ConnectionOptionName::Credential
-            | ConnectionOptionName::Database
-            | ConnectionOptionName::Endpoint
-            | ConnectionOptionName::GcpConnection
-            | ConnectionOptionName::Host
-            | ConnectionOptionName::Password
-            | ConnectionOptionName::Port
-            | ConnectionOptionName::ProgressTopic
-            | ConnectionOptionName::ProgressTopicReplicationFactor
-            | ConnectionOptionName::PublicKey1
-            | ConnectionOptionName::PublicKey2
-            | ConnectionOptionName::Region
-            | ConnectionOptionName::Registry
-            | ConnectionOptionName::AssumeRoleArn
-            | ConnectionOptionName::AssumeRoleSessionName
-            | ConnectionOptionName::SaslMechanisms
-            | ConnectionOptionName::SaslPassword
-            | ConnectionOptionName::SaslUsername
-            | ConnectionOptionName::Scope
-            | ConnectionOptionName::SecurityProtocol
-            | ConnectionOptionName::SecretAccessKey
-            | ConnectionOptionName::ServiceAccountKey
-            | ConnectionOptionName::ServiceName
-            | ConnectionOptionName::SshTunnel
-            | ConnectionOptionName::SslCertificate
-            | ConnectionOptionName::SslCertificateAuthority
-            | ConnectionOptionName::SslKey
-            | ConnectionOptionName::SslMode
-            | ConnectionOptionName::SessionToken
-            | ConnectionOptionName::CatalogType
-            | ConnectionOptionName::Url
-            | ConnectionOptionName::User
-            | ConnectionOptionName::Warehouse => false,
-        }
+        // Credential-bearing options keep their values in redacted mode, so an
+        // inline credential literal renders as `'<REDACTED>'`. A `SECRET`
+        // reference is unaffected either way. It renders as its catalog name
+        // via `WithOptionValue::Secret`.
+        self.value_contains_sensitive_data()
     }
 }
 

src/sql-parser/src/ast/defs/statement.rs

@@ -4463,10 +4463,16 @@ impl<T: AstInfo> AstDisplay for WithOptionValue<T> {
                 | WithOptionValue::Expr(_) => {
                     // These are redact-aware.
                 }
-                WithOptionValue::Secret(_) | WithOptionValue::ConnectionKafkaBroker(_) => {
+                WithOptionValue::ConnectionKafkaBroker(_) => {
                     f.write_str("'<REDACTED>'");
                     return;
                 }
+                // A secret reference is a catalog item name, not the secret
+                // value, so it is safe to show in redacted output. An option
+                // that accepts an inline credential is parsed as a `Value`,
+                // which is redacted by that arm together with the option's
+                // `redact_value()`.
+                WithOptionValue::Secret(_) => {}
                 WithOptionValue::DataType(_)
                 | WithOptionValue::Item(_)
                 | WithOptionValue::UnresolvedItemName(_)

src/sql-parser/src/parser.rs

@@ -3068,6 +3068,9 @@ impl<'a> Parser<'a> {
             )),
             ConnectionOptionName::GcpConnection => Some(self.parse_object_option_value()?),
             ConnectionOptionName::SshTunnel => Some(self.parse_object_option_value()?),
+            _ if name.value_contains_sensitive_data() => {
+                self.parse_optional_connection_credential_option_value()?
+            }
             _ => self.parse_optional_option_value()?,
         };
         Ok(ConnectionOption { name, value })
@@ -5523,6 +5526,40 @@ impl<'a> Parser<'a> {
         Ok(WithOptionValue::Item(self.parse_raw_name()?))
     }
 
+    fn parse_optional_connection_credential_option_value(
+        &mut self,
+    ) -> Result<Option<WithOptionValue<Raw>>, ParserError> {
+        match self.peek_token() {
+            Some(Token::RParen) | Some(Token::Comma) | Some(Token::Semicolon) | None => Ok(None),
+            _ => {
+                let _ = self.consume_token(&Token::Eq);
+                Ok(Some(self.parse_connection_credential_option_value()?))
+            }
+        }
+    }
+
+    fn parse_connection_credential_option_value(
+        &mut self,
+    ) -> Result<WithOptionValue<Raw>, ParserError> {
+        if self.parse_keyword(SECRET) {
+            if let Some(secret) = self.maybe_parse(Parser::parse_raw_name) {
+                Ok(WithOptionValue::Secret(secret))
+            } else {
+                Ok(WithOptionValue::Value(Value::String("secret".to_string())))
+            }
+        } else if let Some(value) = self.maybe_parse(Parser::parse_value) {
+            Ok(WithOptionValue::Value(value))
+        } else if let Some(ident) = self.maybe_parse(Parser::parse_identifier) {
+            Ok(WithOptionValue::Value(Value::String(ident.into_string())))
+        } else {
+            self.expected(
+                self.peek_pos(),
+                "connection credential value",
+                self.peek_token(),
+            )
+        }
+    }
+
     fn parse_optional_option_value(&mut self) -> Result<Option<WithOptionValue<Raw>>, ParserError> {
         // The next token might be a value and might not. The only valid things
         // that indicate no value would be `)` for end-of-options , `,` for

src/sql-parser/tests/sqlparser_common.rs

@@ -643,6 +643,160 @@ fn test_partition_by_value_redacted() {
     );
 }
 
+#[mz_ore::test]
+#[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+fn test_connection_option_redaction() {
+    // Credential-bearing connection options must not leak an inline credential
+    // literal (AWS keys/session tokens, SASL usernames, SSL material, Iceberg
+    // credentials) onto the redacted-telemetry channel (`mz_sql_text_redacted`,
+    // catalog `create_sql` redacted round-trips, trace attributes). A `SECRET`
+    // reference is a catalog name rather than a credential, so it stays visible,
+    // as do other non-sensitive options, keeping redacted SQL diagnostic.
+    struct Case {
+        sql: &'static str,
+        // Substrings that must be gone from the redacted output.
+        redacted_absent: &'static [&'static str],
+        // Substrings that must survive redaction: non-sensitive values,
+        // redacted-literal placeholders, and secret-name references.
+        redacted_present: &'static [&'static str],
+    }
+
+    let cases = [
+        Case {
+            sql: "CREATE CONNECTION c TO AWS (ACCESS KEY ID = 'akid_leak', ENDPOINT = 'endpoint_ok', REGION = 'region_ok', SECRET ACCESS KEY = 'sak_leak', SESSION TOKEN = 'token_leak')",
+            redacted_absent: &["akid_leak", "sak_leak", "token_leak"],
+            redacted_present: &[
+                "ACCESS KEY ID = '<REDACTED>'",
+                "SESSION TOKEN = '<REDACTED>'",
+                "ENDPOINT = 'endpoint_ok'",
+                "REGION = 'region_ok'",
+            ],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO AWS (ACCESS KEY ID = AKIAEXAMPLE, REGION = region_ok, SECRET ACCESS KEY = sak_leak, SESSION TOKEN = token_leak)",
+            redacted_absent: &["akiaexample", "sak_leak", "token_leak"],
+            redacted_present: &[
+                "ACCESS KEY ID = '<REDACTED>'",
+                "SECRET ACCESS KEY = '<REDACTED>'",
+                "SESSION TOKEN = '<REDACTED>'",
+                "REGION = region_ok",
+            ],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO POSTGRES (HOST = pghost, PORT = 1234, DATABASE = 'db_ok', SSL MODE = 'mode_ok', SSL CERTIFICATE AUTHORITY = 'ca_leak', PASSWORD = 'pw_leak', SSL CERTIFICATE = 'cert_leak', SSL KEY = 'key_leak', USER = 'user_leak')",
+            redacted_absent: &["ca_leak", "pw_leak", "cert_leak", "key_leak", "user_leak"],
+            redacted_present: &[
+                "PASSWORD = '<REDACTED>'",
+                "USER = '<REDACTED>'",
+                "HOST = pghost",
+                "PORT = 1234",
+                "DATABASE = 'db_ok'",
+                "SSL MODE = 'mode_ok'",
+            ],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO POSTGRES (HOST = pghost, PORT = 1234, DATABASE = db_ok, SSL MODE = mode_ok, SSL CERTIFICATE AUTHORITY = ca_leak, PASSWORD = secretpw123, SSL CERTIFICATE = cert_leak, SSL KEY = key_leak, USER = myuser)",
+            redacted_absent: &["ca_leak", "secretpw123", "cert_leak", "key_leak", "myuser"],
+            redacted_present: &[
+                "SSL CERTIFICATE AUTHORITY = '<REDACTED>'",
+                "PASSWORD = '<REDACTED>'",
+                "SSL CERTIFICATE = '<REDACTED>'",
+                "SSL KEY = '<REDACTED>'",
+                "USER = '<REDACTED>'",
+                "HOST = pghost",
+                "PORT = 1234",
+                "DATABASE = db_ok",
+                "SSL MODE = mode_ok",
+            ],
+        },
+        Case {
+            // `SASL USERNAME` carries an inline literal (redacted); `SASL
+            // PASSWORD` is a secret reference whose catalog name stays visible.
+            sql: "CREATE CONNECTION c TO KAFKA (BROKER = 'broker_ok:9092', SASL MECHANISMS = 'mech_ok', SASL USERNAME = 'sasluser_leak', SASL PASSWORD = SECRET saslpw_ref, SECURITY PROTOCOL = 'proto_ok')",
+            redacted_absent: &["sasluser_leak"],
+            redacted_present: &[
+                "SASL USERNAME = '<REDACTED>'",
+                "SASL PASSWORD = SECRET saslpw_ref",
+                "BROKER = 'broker_ok:9092'",
+                "SASL MECHANISMS = 'mech_ok'",
+                "SECURITY PROTOCOL = 'proto_ok'",
+            ],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO KAFKA (BROKER = 'broker_ok:9092', SASL MECHANISMS = mech_ok, SASL USERNAME = svcacct, SASL PASSWORD = saslpw_leak, SECURITY PROTOCOL = proto_ok)",
+            redacted_absent: &["svcacct", "saslpw_leak"],
+            redacted_present: &[
+                "SASL USERNAME = '<REDACTED>'",
+                "SASL PASSWORD = '<REDACTED>'",
+                "BROKER = 'broker_ok:9092'",
+                "SASL MECHANISMS = mech_ok",
+                "SECURITY PROTOCOL = proto_ok",
+            ],
+        },
+        Case {
+            // A secret-only option carrying just a reference: nothing to redact,
+            // the catalog name shows.
+            sql: "CREATE CONNECTION c TO GCP (SERVICE ACCOUNT KEY = SECRET gcpkey_ref)",
+            redacted_absent: &[],
+            redacted_present: &["SERVICE ACCOUNT KEY = SECRET gcpkey_ref"],
+        },
+        Case {
+            // A `StringOrSecret` option pointing at a secret keeps the name too.
+            sql: "CREATE CONNECTION c TO AWS (ACCESS KEY ID = SECRET akid_ref, REGION = 'region_ok')",
+            redacted_absent: &[],
+            redacted_present: &["ACCESS KEY ID = SECRET akid_ref", "REGION = 'region_ok'"],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO ICEBERG CATALOG (CATALOG TYPE = 'type_ok', CREDENTIAL = 'cred_leak', WAREHOUSE = 'wh_ok')",
+            redacted_absent: &["cred_leak"],
+            redacted_present: &[
+                "CREDENTIAL = '<REDACTED>'",
+                "CATALOG TYPE = 'type_ok'",
+                "WAREHOUSE = 'wh_ok'",
+            ],
+        },
+        Case {
+            sql: "CREATE CONNECTION c TO ICEBERG CATALOG (CATALOG TYPE = type_ok, CREDENTIAL = cred_leak, WAREHOUSE = wh_ok)",
+            redacted_absent: &["cred_leak"],
+            redacted_present: &[
+                "CREDENTIAL = '<REDACTED>'",
+                "CATALOG TYPE = type_ok",
+                "WAREHOUSE = wh_ok",
+            ],
+        },
+    ];
+
+    for case in cases {
+        let ast = parse_statements(case.sql)
+            .unwrap_or_else(|e| panic!("{:?} should parse: {e}", case.sql))
+            .into_iter()
+            .next()
+            .expect("one statement")
+            .ast;
+
+        let redacted = ast.to_ast_string_redacted();
+        let simple = ast.to_ast_string_simple();
+        for needle in case.redacted_absent {
+            assert!(
+                !redacted.contains(needle),
+                "redacted output leaked {needle:?}:\n{redacted}"
+            );
+            // The value must still render in the non-redacted form, proving we
+            // only changed the redacted path.
+            assert!(
+                simple.contains(needle),
+                "non-redacted output unexpectedly missing {needle:?}:\n{simple}"
+            );
+        }
+        for needle in case.redacted_present {
+            assert!(
+                redacted.contains(needle),
+                "redacted output missing expected {needle:?}:\n{redacted}"
+            );
+        }
+    }
+}
+
 #[mz_ore::test]
 #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
 fn test_collate_low_precedence_display_roundtrip() {

src/sql-parser/tests/testdata/source

@@ -22,9 +22,9 @@ CREATE CONNECTION my_connection TO SQL SERVER (
     PASSWORD SECRET sql_server_pass
 )
 ----
-CREATE CONNECTION my_connection TO SQL SERVER (HOST = 'no_such_address.mtrlz.com', PORT = 5433, DATABASE = sql_server, USER = sql_server, PASSWORD = SECRET sql_server_pass)
+CREATE CONNECTION my_connection TO SQL SERVER (HOST = 'no_such_address.mtrlz.com', PORT = 5433, DATABASE = sql_server, USER = 'sql_server', PASSWORD = SECRET sql_server_pass)
 =>
-CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("my_connection")]), connection_type: SqlServer, if_not_exists: false, values: [ConnectionOption { name: Host, value: Some(Value(String("no_such_address.mtrlz.com"))) }, ConnectionOption { name: Port, value: Some(Value(Number("5433"))) }, ConnectionOption { name: Database, value: Some(UnresolvedItemName(UnresolvedItemName([Ident("sql_server")]))) }, ConnectionOption { name: User, value: Some(UnresolvedItemName(UnresolvedItemName([Ident("sql_server")]))) }, ConnectionOption { name: Password, value: Some(Secret(Name(UnresolvedItemName([Ident("sql_server_pass")])))) }], with_options: [] })
+CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("my_connection")]), connection_type: SqlServer, if_not_exists: false, values: [ConnectionOption { name: Host, value: Some(Value(String("no_such_address.mtrlz.com"))) }, ConnectionOption { name: Port, value: Some(Value(Number("5433"))) }, ConnectionOption { name: Database, value: Some(UnresolvedItemName(UnresolvedItemName([Ident("sql_server")]))) }, ConnectionOption { name: User, value: Some(Value(String("sql_server"))) }, ConnectionOption { name: Password, value: Some(Secret(Name(UnresolvedItemName([Ident("sql_server_pass")])))) }], with_options: [] })
 
 parse-statement
 CREATE SOURCE "no_such_table"

test/mysql-cdc-old-syntax/10-create-connection.td

@@ -41,20 +41,25 @@ name       type
 ------------------------------
 mysq   mysql
 
+>[version<14000] SHOW CREATE CONNECTION mysq
+name   create_sql
+---------------------------------
+materialize.public.mysq "CREATE CONNECTION \"materialize\".\"public\".\"mysq\" TO MYSQL (HOST = \"mysql\", PASSWORD = SECRET \"materialize\".\"public\".\"mysqlpass\", USER = \"root\")"
+
 >[14000<=version<2600700] SHOW CREATE CONNECTION mysq
 name   create_sql
 ---------------------------------
 materialize.public.mysq "CREATE CONNECTION materialize.public.mysq TO MYSQL (HOST = mysql, PASSWORD = SECRET materialize.public.mysqlpass, USER = root);"
 
->[version>=2600700] SHOW CREATE CONNECTION mysq
+>[2600700<=version<2603200] SHOW CREATE CONNECTION mysq
 name   create_sql
 ---------------------------------
 materialize.public.mysq "CREATE CONNECTION materialize.public.mysq\nTO MYSQL (HOST = mysql, PASSWORD = SECRET materialize.public.mysqlpass, USER = root);"
 
->[version<14000] SHOW CREATE CONNECTION mysq
+>[version>=2603200] SHOW CREATE CONNECTION mysq
 name   create_sql
 ---------------------------------
-materialize.public.mysq "CREATE CONNECTION \"materialize\".\"public\".\"mysq\" TO MYSQL (HOST = \"mysql\", PASSWORD = SECRET \"materialize\".\"public\".\"mysqlpass\", USER = \"root\")"
+materialize.public.mysq "CREATE CONNECTION materialize.public.mysq\nTO MYSQL (HOST = mysql, PASSWORD = SECRET materialize.public.mysqlpass, USER = 'root');"
 
 #
 # Error checking

test/mysql-cdc/10-create-connection.td

@@ -41,7 +41,12 @@ name       type
 ------------------------------
 mysq   mysql
 
->[version>=2600700] SHOW CREATE CONNECTION mysq
+>[version>=2603200] SHOW CREATE CONNECTION mysq
+name   create_sql
+---------------------------------
+materialize.public.mysq "CREATE CONNECTION materialize.public.mysq\nTO MYSQL (HOST = mysql, PASSWORD = SECRET materialize.public.mysqlpass, USER = 'root');"
+
+>[2600700<=version<2603200] SHOW CREATE CONNECTION mysq
 name   create_sql
 ---------------------------------
 materialize.public.mysq "CREATE CONNECTION materialize.public.mysq\nTO MYSQL (HOST = mysql, PASSWORD = SECRET materialize.public.mysqlpass, USER = root);"

test/mysql-cdc/create-alter-iam-connection.td

@@ -28,9 +28,12 @@ mysql_conn mysql
 >[version<2600700] SHOW CREATE CONNECTION mysql_conn
 materialize.public.mysql_conn "CREATE CONNECTION materialize.public.mysql_conn TO MYSQL (AWS CONNECTION = materialize.public.aws_conn, HOST = mysql, SSL MODE = required, USER = root);"
 
->[version>=2600700] SHOW CREATE CONNECTION mysql_conn
+>[2600700<=version<2603200] SHOW CREATE CONNECTION mysql_conn
 materialize.public.mysql_conn "CREATE CONNECTION materialize.public.mysql_conn\nTO MYSQL\n    (AWS CONNECTION = materialize.public.aws_conn, HOST = mysql, SSL MODE = required, USER = root);"
 
+>[version>=2603200] SHOW CREATE CONNECTION mysql_conn
+materialize.public.mysql_conn "CREATE CONNECTION materialize.public.mysql_conn\nTO MYSQL\n    (AWS CONNECTION = materialize.public.aws_conn, HOST = mysql, SSL MODE = required, USER = 'root');"
+
 
 ! ALTER CONNECTION mysql_conn SET (PASSWORD SECRET mysqlpass);
 contains:IAM authentication is not supported with password