fix: harden credential redaction in logged SQL

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Fixes
 
+- Redact all `credential` and `encryption` clauses in logged SQL, regardless of keyword case (thanks @SreeramaYeshwanthGowd!) ([#1610](https://github.com/databricks/dbt-databricks/pull/1610) resolves [#1609](https://github.com/databricks/dbt-databricks/issues/1609))
 - Escape single quotes in relation comments so materialized views and streaming tables with an apostrophe in the description can be created (thanks @SreeramaYeshwanthGowd!) ([#1613](https://github.com/databricks/dbt-databricks/pull/1613) resolves [#1251](https://github.com/databricks/dbt-databricks/issues/1251))
 
 ### Under the Hood

dbt/adapters/databricks/utils.py

@@ -17,26 +17,38 @@
 A = TypeVar("A", bound=BaseAdapter)
 
 
-CREDENTIAL_IN_COPY_INTO_REGEX = re.compile(
-    r"(?<=credential)\s*?\((\s*?'\w*?'\s*?=\s*?'.*?'\s*?(?:,\s*?'\w*?'\s*?=\s*?'.*?'\s*?)*?)\)"
+_SECRET_OPTION_KEY = r"'[^']+'"
+# The alternatives are deliberately non-overlapping to keep matching linear on malformed input.
+_SECRET_OPTION_VALUE = r"'(?:\\.|''|[^'\\]|'(?!'|\s*[,)]))*'"
+_SECRET_OPTION = _SECRET_OPTION_KEY + r"\s*=\s*" + _SECRET_OPTION_VALUE
+
+SECRET_CLAUSE_IN_COPY_INTO_REGEX = re.compile(
+    r"(credential|encryption)\s*\(\s*"
+    r"(" + _SECRET_OPTION + r"(?:\s*,\s*" + _SECRET_OPTION + r")*)"
+    r"\s*\)",
+    re.IGNORECASE,
 )
+SECRET_OPTION_KEY_REGEX = re.compile("(" + _SECRET_OPTION_KEY + r")\s*=\s*" + _SECRET_OPTION_VALUE)
 
 
 def redact_credentials(sql: str) -> str:
-    redacted = _redact_credentials_in_copy_into(sql)
-    return redacted
+    try:
+        return _redact_credentials_in_copy_into(sql)
+    except Exception:
+        return sql
+
+
+def _redact_secret_clause(match: "re.Match[str]") -> str:
+    keys = SECRET_OPTION_KEY_REGEX.findall(match.group(2))
+    return f"{match.group(1)} (" + ", ".join(f"{key} = '[REDACTED]'" for key in keys) + ")"
 
 
 def _redact_credentials_in_copy_into(sql: str) -> str:
-    m = CREDENTIAL_IN_COPY_INTO_REGEX.search(sql, re.MULTILINE)
-    if m:
-        redacted = ", ".join(
-            f"{key.strip()} = '[REDACTED]'"
-            for key, _ in (pair.strip().split("=", 1) for pair in m.group(1).split(","))
-        )
-        return f"{sql[: m.start()]} ({redacted}){sql[m.end() :]}"
-    else:
+    # Cheap substring test first; the case-insensitive scan is much slower on large statements.
+    lowered = sql.lower()
+    if "credential" not in lowered and "encryption" not in lowered:
         return sql
+    return SECRET_CLAUSE_IN_COPY_INTO_REGEX.sub(_redact_secret_clause, sql)
 
 
 def remove_undefined(v: Any) -> Any:

tests/unit/test_utils.py

@@ -1,3 +1,4 @@
+import dbt.adapters.databricks.utils as databricks_utils
 from dbt.adapters.databricks.utils import (
     is_cluster_http_path,
     quote,
@@ -51,6 +52,139 @@ def test_redact_credentials__multiple_credentials(self):
         )
         assert redact_credentials(sql) == expected
 
+    def test_redact_credentials__uppercase_credential(self):
+        sql = "copy into target_table\nfrom source_table\n  WITH (CREDENTIAL ('KEY' = 'VALUE'))"
+        expected = (
+            "copy into target_table\nfrom source_table\n  WITH (CREDENTIAL ('KEY' = '[REDACTED]'))"
+        )
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__encryption(self):
+        sql = (
+            "copy into target_table\n"
+            "from source_table\n"
+            "  WITH (encryption ('TYPE' = 'AWS_SSE_C', 'MASTER_KEY' = 'VALUE'))"
+        )
+        expected = (
+            "copy into target_table\n"
+            "from source_table\n"
+            "  WITH (encryption ('TYPE' = '[REDACTED]', 'MASTER_KEY' = '[REDACTED]'))"
+        )
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__credential_and_encryption(self):
+        sql = (
+            "copy into target_table\n"
+            "from source_table\n"
+            "  WITH (credential ('KEY' = 'VALUE') encryption ('MASTER_KEY' = 'VALUE'))"
+        )
+        expected = (
+            "copy into target_table\n"
+            "from source_table\n"
+            "  WITH (credential ('KEY' = '[REDACTED]') encryption ('MASTER_KEY' = '[REDACTED]'))"
+        )
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__value_with_comma(self):
+        sql = "copy into target_table\n  WITH (credential ('KEY' = 'VALUE,WITH,COMMAS'))"
+        expected = "copy into target_table\n  WITH (credential ('KEY' = '[REDACTED]'))"
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__value_with_newline(self):
+        sql = "copy into target_table\n  WITH (credential ('KEY' = 'VALUE\nCONTINUED'))"
+        expected = "copy into target_table\n  WITH (credential ('KEY' = '[REDACTED]'))"
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__value_with_quote(self):
+        sql = "copy into target_table\n  WITH (credential ('KEY' = 'VALUE'WITH'QUOTES'))"
+        expected = "copy into target_table\n  WITH (credential ('KEY' = '[REDACTED]'))"
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__value_with_escaped_quote(self):
+        sql = "copy into target_table\n  WITH (credential ('KEY' = 'VALUE\\'ESCAPED'))"
+        expected = "copy into target_table\n  WITH (credential ('KEY' = '[REDACTED]'))"
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__escaped_quote_before_delimiter(self):
+        cases = [
+            "copy into target_table\n  WITH (credential ('KEY' = 'PREFIX'',SUFFIX'))",
+            "copy into target_table\n  WITH (credential ('KEY' = 'PREFIX'')SUFFIX'))",
+            "copy into target_table\n  WITH (credential ('KEY' = 'PREFIX\\',SUFFIX'))",
+            "copy into target_table\n  WITH (credential ('KEY' = 'PREFIX\\')SUFFIX'))",
+        ]
+        expected = "copy into target_table\n  WITH (credential ('KEY' = '[REDACTED]'))"
+
+        for sql in cases:
+            redacted = redact_credentials(sql)
+            assert redacted == expected
+            assert "PREFIX" not in redacted
+            assert "SUFFIX" not in redacted
+
+    def test_redact_credentials__malformed_secret_clause_is_unchanged(self):
+        sql = "copy into target_table\n  WITH (credential ('KEY' = 'PREFIX',SUFFIX')) trailing SQL"
+
+        assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__secretless_clause_is_unchanged(self):
+        cases = [
+            "copy into target_table WITH (credential ())",
+            "select credential('public literal') as x, 42 as y",
+            "select my_encryption('public literal') as x",
+        ]
+
+        for sql in cases:
+            assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__unquoted_key_is_unchanged(self):
+        sql = "copy into target_table WITH (credential (KEY = 'SECRET')) trailing SQL"
+
+        assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__large_unterminated_clause(self):
+        sql = "credential (" + ", ".join("'KEY' = 'VALUE'" for _ in range(1_000))
+
+        assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__large_ordinary_statement_uses_fast_path(self, monkeypatch):
+        class UnexpectedRegex:
+            def sub(self, replacement, sql):
+                raise AssertionError("ordinary SQL should bypass the secret-clause regex")
+
+        monkeypatch.setattr(databricks_utils, "SECRET_CLAUSE_IN_COPY_INTO_REGEX", UnexpectedRegex())
+        sql = "select 1 -- " + "x" * 1_000_000
+
+        assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__internal_error_fails_open(self, monkeypatch):
+        sql = "copy into target_table WITH (credential ('KEY' = 'SYNTHETIC_SECRET'))"
+
+        def raise_internal_error(sql: str) -> str:
+            raise RuntimeError("synthetic redactor failure")
+
+        monkeypatch.setattr(
+            databricks_utils,
+            "_redact_credentials_in_copy_into",
+            raise_internal_error,
+        )
+
+        assert redact_credentials(sql) == sql
+
+    def test_redact_credentials__key_with_dots(self):
+        sql = "copy into target_table\n  WITH (credential ('fs.azure.account.key' = 'VALUE'))"
+        expected = (
+            "copy into target_table\n  WITH (credential ('fs.azure.account.key' = '[REDACTED]'))"
+        )
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__prefixed_keyword(self):
+        sql = "copy into target_table\n  WITH (storage_credential ('KEY' = 'VALUE'))"
+        expected = "copy into target_table\n  WITH (storage_credential ('KEY' = '[REDACTED]'))"
+        assert redact_credentials(sql) == expected
+
+    def test_redact_credentials__non_option_clause(self):
+        sql = "select * from target_table where credential_id = 1"
+        assert redact_credentials(sql) == sql
+
     def test_remove_ansi(self):
         test_string = """Python model failed with traceback as:
   ---------------------------------------------------------------------------