Redact password column from CSV import logging and error output

CHANGELOG.md

@@ -1,6 +1,16 @@
 
 ## Unreleased
 
+* Security: `UserItem.CSVImport` no longer logs the password column when
+  validating a user-import CSV file. Previously, `validate_file_for_import`
+  emitted the first four characters of each raw row at INFO (which could
+  include the beginning of the password when the username was short), and the
+  full raw row was pushed into the returned `invalid_lines` list unmasked; the
+  per-column log inside `_validate_import_line_or_throw` also wrote the
+  password value at DEBUG. Row-level logging is now DEBUG-only and logs the
+  username instead of a raw slice, and the password column is replaced with
+  `***` before any line reaches a log handler or the invalid-lines list.
+  Fixes #1829.
 * Added Workload Identity Federation (WIF) support to the connection update
   endpoints. `ConnectionItem` has two new string properties,
   `workgroup_auth_mode` and `auth_federation_service_account`, that are

tableauserverclient/models/user_item.py

@@ -1,3 +1,4 @@
+import csv
 import io
 import warnings
 import xml.etree.ElementTree as ET
@@ -471,7 +472,7 @@ class ColumnType(IntEnum):
         def create_user_from_line(line: str):
             if line is None or line is False or line == "\n" or line == "":
                 return None
-            values: list[str] = list(map(str.strip, line.strip().split(",")))
+            values = UserItem.CSVImport._parse_csv_line(line)
             if len(values) > UserItem.CSVImport.COLUMN_COUNT:
                 raise ValueError("Too many attributes for user import")
             username = values[UserItem.CSVImport.ColumnType.USERNAME]
@@ -521,6 +522,13 @@ def create_user_from_line(line: str):
                 )
             return user
 
+        @staticmethod
+        def _parse_csv_line(line: str) -> list[str]:
+            values = next(csv.reader([line], strict=True, skipinitialspace=True), [])
+            if not values:
+                values = [""]
+            return list(map(str.strip, values))
+
         # Read through an entire CSV file meant for user import
         # Return the number of valid lines and a list of all the invalid lines
         @staticmethod
@@ -530,17 +538,42 @@ def validate_file_for_import(csv_file: io.TextIOWrapper, logger) -> tuple[int, l
             csv_file.seek(0)  # set to start of file in case it has been read earlier
             line: str = csv_file.readline()
             while line and line != "":
+                # Log only the username (column 0); the rest of the line contains the password (column 1) and other PII.
+                username = line.partition(",")[0].strip()
                 try:
-                    # do not print passwords
-                    logger.info(f"Reading user {line[:4]}")
+                    logger.debug(f"Reading user {username}")
                     UserItem.CSVImport._validate_import_line_or_throw(line, logger)
                     num_valid_lines += 1
                 except Exception as exc:
-                    logger.info(f"Error parsing {line[:4]}: {exc}")
-                    invalid_lines.append(line)
+                    logger.debug(f"Error parsing user {username}: {exc}")
+                    invalid_lines.append(UserItem.CSVImport._redact_password_column(line))
                 line = csv_file.readline()
             return num_valid_lines, invalid_lines
 
+        # Return a copy of a raw CSV line with the password column replaced by "***".
+        # Callers that log or expose invalid rows will not disclose the credential.
+        # Preserves the original line ending (\r\n or \n) so log output and
+        # returned rows stay byte-identical apart from the redaction.
+        @staticmethod
+        def _redact_password_column(line: str) -> str:
+            if line.endswith("\r\n"):
+                body, ending = line[:-2], "\r\n"
+            elif line.endswith("\n"):
+                body, ending = line[:-1], "\n"
+            else:
+                body, ending = line, ""
+            try:
+                fields = UserItem.CSVImport._parse_csv_line(body)
+            except csv.Error:
+                return "<invalid CSV row redacted>" + ending
+            pass_index = UserItem.CSVImport.ColumnType.PASS.value
+            if len(fields) > pass_index:
+                fields[pass_index] = "***"
+                output = io.StringIO()
+                csv.writer(output, lineterminator="").writerow(fields)
+                return output.getvalue() + ending
+            return line
+
         # Some fields in the import file are restricted to specific values
         # Iterate through each field and validate the given value against hardcoded constraints
         @staticmethod
@@ -558,13 +591,14 @@ def _validate_import_line_or_throw(incoming, logger) -> None:
                 list(UserItem.CSVImport._AUTH_CANONICAL.values()),  # auth - normalized before comparison
             ]
 
-            line = list(map(str.strip, incoming.split(",")))
+            line = UserItem.CSVImport._parse_csv_line(incoming)
             if len(line) > UserItem.CSVImport.COLUMN_COUNT:
                 raise ValueError("Too many attributes for user import")
             username = line[UserItem.CSVImport.ColumnType.USERNAME.value]
             logger.debug(f"> details - {username}")
             UserItem.validate_username_or_throw(username)
             for i in range(1, len(line)):
+                column = UserItem.CSVImport.ColumnType(i)
                 value = line[i]
                 valid = _valid_attributes[i]
                 column = UserItem.CSVImport.ColumnType(i)
@@ -591,8 +625,16 @@ def _validate_import_line_or_throw(incoming, logger) -> None:
                             skip_validation = True
                     else:
                         value = value.lower()
-                logger.debug(f"column {column.name}: {value}")
-                if not skip_validation:
+                # A malformed row can shift password fragments into later columns;
+                # keep every field after the username out of the logs.
+                safe_value = value if column.value < UserItem.CSVImport.ColumnType.PASS.value else "***"
+                logger.debug(f"column {column.name}: {safe_value}")
+                # Never call _validate_attribute_value for PASS: its ValueError
+                # embeds the raw value, and a future PR that adds password-format
+                # checks to PASS's allowlist would leak the password through the
+                # exception message. Belt-and-braces even though PASS's allowlist
+                # is currently empty.
+                if not skip_validation and column != UserItem.CSVImport.ColumnType.PASS:
                     UserItem.CSVImport._validate_attribute_value(value, valid, column)
 
         # Given a restricted set of possible values, confirm the item is in that set

test/test_user_model.py

@@ -171,6 +171,125 @@ def test_validate_usernames_file() -> None:
     assert valid == 5, f"Exactly 5 of the lines were valid, counted {valid + len(invalid)}"
 
 
+def _mask_present(records: list) -> bool:
+    combined = "\n".join(record.getMessage() for record in records)
+    return "PASS" in combined and "***" in combined
+
+
+def test_password_not_logged_at_debug(caplog: pytest.LogCaptureFixture) -> None:
+    """Regression test for #1829: passwords must not appear in DEBUG logs."""
+    secret = "hunter2SUPERSECRET"
+    line = f"jsmith,{secret},John Smith,creator,site,yes,jsmith@example.com"
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        TSC.UserItem.CSVImport._validate_import_line_or_throw(line, logger)
+    combined = "\n".join(record.getMessage() for record in caplog.records)
+    assert secret not in combined, f"Password leaked into logs: {combined!r}"
+    # Positive assertion: something references the PASS column and something is
+    # masked as ***, so a "fix" that only removed the log line would not pass.
+    assert _mask_present(caplog.records), f"Expected masked PASS log line; got: {combined!r}"
+
+
+def test_password_not_logged_when_line_invalid(caplog: pytest.LogCaptureFixture) -> None:
+    """Regression test for #1829: passwords must not appear when a row fails to validate."""
+    secret = "hunter2SUPERSECRET"
+    line = f"jsmith,{secret},John Smith,not-a-real-license,site,yes,jsmith@example.com"
+    test_data = _mock_file_content([line])
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        valid, invalid = TSC.UserItem.CSVImport.validate_file_for_import(test_data, logger)
+    assert valid == 0
+    assert len(invalid) == 1
+    assert secret not in invalid[0], f"Password leaked into returned invalid_lines: {invalid[0]!r}"
+    combined = "\n".join(record.getMessage() for record in caplog.records)
+    assert secret not in combined, f"Password leaked into logs on invalid row: {combined!r}"
+
+
+def test_password_with_comma_is_fully_redacted(caplog: pytest.LogCaptureFixture) -> None:
+    """A quoted comma in a password must not leak from invalid row output or logs."""
+    line = 'jsmith,"hunter2,SECRETTAIL",John Smith,not-a-real-license,site,yes,jsmith@example.com'
+    test_data = _mock_file_content([line])
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        valid, invalid = TSC.UserItem.CSVImport.validate_file_for_import(test_data, logger)
+    assert valid == 0
+    assert len(invalid) == 1
+    assert "hunter2" not in invalid[0]
+    assert "SECRETTAIL" not in invalid[0]
+    combined = "\n".join(record.getMessage() for record in caplog.records)
+    assert "hunter2" not in combined
+    assert "SECRETTAIL" not in combined
+    assert _mask_present(caplog.records)
+
+
+def test_malformed_csv_row_is_not_returned_unredacted(caplog: pytest.LogCaptureFixture) -> None:
+    secret = "SECRETTAIL"
+    test_data = _mock_file_content([f'jsmith,"{secret}'])
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        valid, invalid = TSC.UserItem.CSVImport.validate_file_for_import(test_data, logger)
+    assert valid == 0
+    assert invalid == ["<invalid CSV row redacted>"]
+    combined = "\n".join(record.getMessage() for record in caplog.records)
+    assert secret not in combined
+
+
+def test_password_not_in_validate_attribute_value_error(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Regression test: _validate_import_line_or_throw must never call
+    _validate_attribute_value for the PASS column. Today PASS has an empty
+    allowlist so the validator returns early, but if a future PR adds
+    password-format checks (length, complexity, banned chars) the raw password
+    would leak through ValueError("Invalid value {item} for {column_type}").
+    Guarantee the invariant at the caller so it doesn't depend on the allowlist
+    staying empty."""
+    secret = "hunter2SUPERSECRET"
+    line = f"jsmith,{secret},John Smith,creator,site,yes,jsmith@example.com"
+
+    # Spy that records every call and simulates a future PR that added a
+    # non-empty allowlist to PASS (any value not in the allowlist raises).
+    calls: list[tuple] = []
+    original = TSC.UserItem.CSVImport._validate_attribute_value
+
+    def spy(item: str, possible_values: list, column_type) -> None:
+        calls.append((item, column_type))
+        if column_type == TSC.UserItem.CSVImport.ColumnType.PASS:
+            # Simulate a hypothetical password-format check.
+            raise ValueError(f"Invalid value {item} for {column_type}")
+        return original(item, possible_values, column_type)
+
+    monkeypatch.setattr(TSC.UserItem.CSVImport, "_validate_attribute_value", spy)
+
+    with caplog.at_level(logging.DEBUG, logger=logger.name):
+        # If PASS were passed to the validator, the spy would raise with the
+        # raw secret embedded in the ValueError. It must not raise.
+        TSC.UserItem.CSVImport._validate_import_line_or_throw(line, logger)
+
+    pass_calls = [c for c in calls if c[1] == TSC.UserItem.CSVImport.ColumnType.PASS]
+    assert not pass_calls, f"_validate_attribute_value was called for PASS: {pass_calls!r}"
+
+    combined = "\n".join(record.getMessage() for record in caplog.records)
+    assert secret not in combined, f"Password leaked into logs: {combined!r}"
+
+
+def test_redact_password_column_helper() -> None:
+    """Unit-level coverage for _redact_password_column across newline and edge cases."""
+    redact = TSC.UserItem.CSVImport._redact_password_column
+    # LF-terminated
+    assert redact("jsmith,hunter2,fname\n") == "jsmith,***,fname\n"
+    # CRLF-terminated (the \r rides with the last field, ending is preserved)
+    assert redact("jsmith,hunter2,fname\r\n") == "jsmith,***,fname\r\n"
+    # CRLF where password IS the last field: the \r must not be silently
+    # dropped when the password value is replaced.
+    assert redact("jsmith,hunter2\r\n") == "jsmith,***\r\n"
+    # No trailing newline
+    assert redact("jsmith,hunter2,fname") == "jsmith,***,fname"
+    # Empty password field: still replaced (unconditional mask)
+    assert redact("jsmith,,fname") == "jsmith,***,fname"
+    # Trailing comma with nothing after: column 1 exists as empty string, gets masked
+    assert redact("jsmith,") == "jsmith,***"
+    # Single column: no password to redact; return line unchanged
+    assert redact("jsmith") == "jsmith"
+    assert redact("jsmith\n") == "jsmith\n"
+
+
 def test_validate_mixed_case_license() -> None:
     # Regression: issue #1809 - 'Viewer' (capital V) was rejected by case-sensitive check
     TSC.UserItem.CSVImport._validate_import_line_or_throw("username, pword, fname, Viewer, None, no, email", logger)
@@ -199,6 +318,12 @@ def test_create_user_with_auth_column() -> None:
     assert user.auth_setting == "SAML", f"Expected SAML, got {user.auth_setting}"
 
 
+def test_create_user_with_comma_in_password() -> None:
+    user = TSC.UserItem.CSVImport.create_user_from_line('username,"pass,word",fname,creator,none,yes,email')
+    assert user is not None
+    assert user.fullname == "fname"
+
+
 def test_too_many_columns_raises() -> None:
     with pytest.raises(ValueError):
         TSC.UserItem.CSVImport.create_user_from_line("u, p, n, creator, none, yes, email, SAML, extra")