refactor: update password handling to use SecretStr

bfabric/src/bfabric/bfabric.py

@@ -118,7 +118,7 @@ def from_token(
         """
         config, _ = get_system_auth(config_env=config_env, config_path=config_path)
         token_data = get_token_data(client_config=config, token=token)
-        auth = BfabricAuth(login=token_data.user, password=token_data.user_ws_password.get_secret_value())
+        auth = BfabricAuth(login=token_data.user, password=token_data.user_ws_password)
         return cls(config, auth, engine=engine)
 
     @property

bfabric/src/bfabric/config/bfabric_auth.py

@@ -2,16 +2,11 @@
 
 from typing import Annotated
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, SecretStr
 
 
 class BfabricAuth(BaseModel):
     """Holds the authentication data for the B-Fabric client."""
 
     login: Annotated[str, Field(min_length=3)]
-    password: Annotated[str, Field(min_length=32, max_length=32)]
-
-    def __repr__(self) -> str:
-        return f"BfabricAuth(login={repr(self.login)}, password=...)"
-
-    __str__ = __repr__
+    password: Annotated[SecretStr, Field(min_length=32, max_length=32)]

bfabric/src/bfabric/engine/engine_suds.py

@@ -49,7 +49,7 @@ def read(
         full_query = dict(
             login=auth.login,
             page=page,
-            password=auth.password,
+            password=auth.password.get_secret_value(),
             query=query,
             idonly=return_id_only,
         )
@@ -65,7 +65,7 @@ def save(self, endpoint: str, obj: dict, auth: BfabricAuth, method: str = "save"
         :param method: the method to use for saving, generally "save", but in some cases e.g. "checkandinsert" is more
             appropriate to be used instead.
         """
-        query = {"login": auth.login, "password": auth.password, endpoint: obj}
+        query = {"login": auth.login, "password": auth.password.get_secret_value(), endpoint: obj}
         service = self._get_suds_service(endpoint)
         try:
             response = getattr(service, method)(query)
@@ -84,7 +84,7 @@ def delete(self, endpoint: str, id: int | list[int], auth: BfabricAuth) -> Resul
             # TODO maybe use error here (and make sure it's consistent)
             return ResultContainer([], total_pages_api=0)
 
-        query = {"login": auth.login, "password": auth.password, "id": id}
+        query = {"login": auth.login, "password": auth.password.get_secret_value(), "id": id}
         service = self._get_suds_service(endpoint)
         response = service.delete(query)
         return self._convert_results(response=response, endpoint=endpoint)

bfabric/src/bfabric/engine/engine_zeep.py

@@ -59,7 +59,7 @@ def read(
         full_query = dict(
             login=auth.login,
             page=page,
-            password=auth.password,
+            password=auth.password.get_secret_value(),
             query=query,
             idonly=return_id_only,
         )
@@ -84,7 +84,7 @@ def save(self, endpoint: str, obj: dict, auth: BfabricAuth, method: str = "save"
             excl_keys = ["name", "sampleid", "storageid", "workunitid", "relativepath"]
             _zeep_query_append_skipped(query, excl_keys, inplace=True, overwrite=False)
 
-        full_query = {"login": auth.login, "password": auth.password, endpoint: query}
+        full_query = {"login": auth.login, "password": auth.password.get_secret_value(), endpoint: query}
 
         client = self._get_client(endpoint)
 
@@ -108,7 +108,7 @@ def delete(self, endpoint: str, id: int | list[int], auth: BfabricAuth) -> Resul
             # TODO maybe use error here (and make sure it's consistent)
             return ResultContainer([], total_pages_api=0)
 
-        query = {"login": auth.login, "password": auth.password, "id": id}
+        query = {"login": auth.login, "password": auth.password.get_secret_value(), "id": id}
 
         client = self._get_client(endpoint)
         response = client.service.delete(query)

bfabric/src/bfabric/examples/compare_zeep_suds_query.py

@@ -52,7 +52,7 @@ def full_query(auth: BfabricAuth, query: dict, includedeletableupdateable: bool
     thisQuery = deepcopy(query)
     thisQuery["includedeletableupdateable"] = includedeletableupdateable
 
-    return {"login": auth.login, "password": auth.password, "query": thisQuery}
+    return {"login": auth.login, "password": auth.password.get_secret_value(), "query": thisQuery}
 
 
 def calc_both(

bfabric/src/bfabric/examples/zeep_debug.py

@@ -22,7 +22,7 @@ def full_query(auth: BfabricAuth, query: dict, includedeletableupdateable: bool
     thisQuery = deepcopy(query)
     thisQuery["includedeletableupdateable"] = includedeletableupdateable
 
-    return {"login": auth.login, "password": auth.password, "query": thisQuery}
+    return {"login": auth.login, "password": auth.password.get_secret_value(), "query": thisQuery}
 
 
 def read_zeep(wsdl, fullQuery, raw=True):

docs/changelog.md

@@ -16,6 +16,10 @@ Versioning currently follows `X.Y.Z` where
 - `Bfabric.from_token` to create a `Bfabric` instance from a token
 - `bfabric.rest.token_data` to get token data from the REST API, low-level functionality
 
+### Changed
+
+- Internally, the user password is now in a `pydantic.SecretStr` until we construct the API call. This should prevent some logging related accidents.
+
 ## \[1.13.20\] - 2025-02-10
 
 ### Breaking

tests/bfabric/config/test_bfabric_auth.py

@@ -11,11 +11,14 @@ def example_config_path() -> Path:
 
 
 def test_bfabric_auth_repr() -> None:
-    assert repr(BfabricAuth(login="login", password="x" * 32)) == "BfabricAuth(login='login', password=...)"
+    assert (
+        repr(BfabricAuth(login="login", password="x" * 32))
+        == "BfabricAuth(login='login', password=SecretStr('**********'))"
+    )
 
 
 def test_bfabric_auth_str() -> None:
-    assert str(BfabricAuth(login="login", password="x" * 32)) == "BfabricAuth(login='login', password=...)"
+    assert str(BfabricAuth(login="login", password="x" * 32)) == "login='login' password=SecretStr('**********')"
 
 
 if __name__ == "__main__":

tests/bfabric/config/test_bfabric_client_config.py

@@ -70,7 +70,7 @@ def test_bfabric_config_read_yml_bypath_default(mocker: MockerFixture, example_c
 
     config, auth = read_config(example_config_path)
     assert auth.login == "my_epic_production_login"
-    assert auth.password == "01234567890123456789012345678901"
+    assert auth.password.get_secret_value() == "01234567890123456789012345678901"
     assert config.base_url == "https://mega-production-server.uzh.ch/myprod"
 
     logot.assert_logged(logged.debug(f"Reading configuration from: {str(example_config_path.absolute())}"))
@@ -85,7 +85,7 @@ def test_bfabric_config_read_yml_bypath_environment_variable(
 
     config, auth = read_config(example_config_path)
     assert auth.login == "my_epic_test_login"
-    assert auth.password == "012345678901234567890123456789ff"
+    assert auth.password.get_secret_value() == "012345678901234567890123456789ff"
     assert config.base_url == "https://mega-test-server.uzh.ch/mytest"
 
     logot.assert_logged(logged.debug(f"Reading configuration from: {str(example_config_path.absolute())}"))

tests/bfabric/config/test_config_file.py

@@ -54,7 +54,7 @@ def test_environment_config_when_auth(data_with_auth):
     config = EnvironmentConfig.model_validate(data_with_auth["PRODUCTION"])
     assert config.config.base_url == "https://example.com/"
     assert config.auth.login == "test-dummy"
-    assert config.auth.password == "00000000001111111111222222222233"
+    assert config.auth.password.get_secret_value() == "00000000001111111111222222222233"
 
 
 def test_environment_config_when_no_auth(data_no_auth):
@@ -69,7 +69,7 @@ def test_config_file_when_auth(data_with_auth):
     assert len(config.environments) == 1
     assert config.environments["PRODUCTION"].config.base_url == "https://example.com/"
     assert config.environments["PRODUCTION"].auth.login == "test-dummy"
-    assert config.environments["PRODUCTION"].auth.password == "00000000001111111111222222222233"
+    assert config.environments["PRODUCTION"].auth.password.get_secret_value() == "00000000001111111111222222222233"
 
 
 def test_config_file_when_no_auth(data_no_auth):
@@ -88,7 +88,7 @@ def test_config_file_when_multiple(data_multiple):
     assert config.environments["PRODUCTION"].auth is None
     assert config.environments["TEST"].config.base_url == "https://test.example.com/"
     assert config.environments["TEST"].auth.login == "test-dummy"
-    assert config.environments["TEST"].auth.password == "00000000001111111111222222222233"
+    assert config.environments["TEST"].auth.password.get_secret_value() == "00000000001111111111222222222233"
 
 
 def test_config_file_when_non_existent_default(data_no_auth):

tests/bfabric/engine/test_engine_suds.py

@@ -1,6 +1,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from pydantic import SecretStr
 from suds import MethodNotFound
 from suds.client import Client
 
@@ -16,7 +17,7 @@ def engine_suds():
 
 @pytest.fixture
 def mock_auth():
-    return MagicMock(login="test_user", password="test_pass")
+    return MagicMock(login="test_user", password=SecretStr("test_pass"))
 
 
 @pytest.fixture

tests/bfabric/engine/test_engine_zeep.py

@@ -2,6 +2,7 @@
 
 import pytest
 import zeep
+from pydantic import SecretStr
 
 from bfabric.engine.engine_zeep import EngineZeep, _zeep_query_append_skipped
 from bfabric.errors import BfabricRequestError
@@ -15,7 +16,7 @@ def engine_zeep():
 
 @pytest.fixture
 def mock_auth():
-    return MagicMock(login="test_user", password="test_pass")
+    return MagicMock(login="test_user", password=SecretStr("test_pass"))
 
 
 @pytest.fixture

tests/bfabric/test_bfabric_config.py

@@ -22,7 +22,7 @@ def test_read_yml_bypath_all_fields(example_config_path: Path) -> None:
     job_notification_emails_ground_truth = "[email protected] [email protected]"
 
     assert auth.login == "my_epic_test_login"
-    assert auth.password == "012345678901234567890123456789ff"
+    assert auth.password.get_secret_value() == "012345678901234567890123456789ff"
     assert config.base_url == "https://mega-test-server.uzh.ch/mytest"
     assert config.application_ids == applications_dict_ground_truth
     assert config.job_notification_emails == job_notification_emails_ground_truth