diff --git a/src/govoplan_files/backend/storage/connector_profile_store.py b/src/govoplan_files/backend/storage/connector_profile_store.py index 054dbfc..a9f9f9a 100644 --- a/src/govoplan_files/backend/storage/connector_profile_store.py +++ b/src/govoplan_files/backend/storage/connector_profile_store.py @@ -188,18 +188,80 @@ def update_connector_profile_row( clear_password: bool = False, clear_token: bool = False, ) -> FileConnectorProfile: + _validate_profile_update_references( + row, + password_env=password_env, + token_env=token_env, + secret_ref=secret_ref, + metadata=metadata, + ) + _update_profile_connection_fields( + row, + label=label, + provider=provider, + endpoint_url=endpoint_url, + base_path=base_path, + enabled=enabled, + credential_profile_id=credential_profile_id, + credential_mode=credential_mode, + ) + _update_profile_credential_fields( + row, + username=username, + password=password, + token=token, + password_env=password_env, + token_env=token_env, + secret_ref=secret_ref, + clear_password=clear_password, + clear_token=clear_token, + ) + _update_profile_governance_fields( + row, + capabilities=capabilities, + policy=policy, + metadata=metadata, + ) + row.updated_by_user_id = user_id + session.add(row) + session.flush() + return row + + +def _validate_profile_update_references( + row: FileConnectorProfile, + *, + password_env: str | None, + token_env: str | None, + secret_ref: str | None, + metadata: Mapping[str, Any] | None, +) -> None: reject_api_controlled_deployment_references( password_env=password_env, token_env=token_env, secret_ref=secret_ref, metadata=metadata, ) - if secret_ref is not None and _clean(secret_ref) != _clean(row.secret_ref): - if _clean(row.secret_ref): - raise FileStorageError( - "An existing external secret reference cannot be replaced or cleared until Files can prove " - "provider ownership and confirm provider-side deletion" - ) + if secret_ref is None or _clean(secret_ref) == _clean(row.secret_ref): + return + if _clean(row.secret_ref): + raise FileStorageError( + "An existing external secret reference cannot be replaced or cleared until Files can prove " + "provider ownership and confirm provider-side deletion" + ) + + +def _update_profile_connection_fields( + row: FileConnectorProfile, + *, + label: str | None, + provider: str | None, + endpoint_url: str | None, + base_path: str | None, + enabled: bool | None, + credential_profile_id: str | None, + credential_mode: str | None, +) -> None: if label is not None: row.label = _normalize_label(label) if provider is not None: @@ -214,6 +276,20 @@ def update_connector_profile_row( row.credential_profile_id = _clean(credential_profile_id) if credential_mode is not None: row.credential_mode = _normalize_credential_mode(credential_mode) + + +def _update_profile_credential_fields( + row: FileConnectorProfile, + *, + username: str | None, + password: str | None, + token: str | None, + password_env: str | None, + token_env: str | None, + secret_ref: str | None, + clear_password: bool, + clear_token: bool, +) -> None: if username is not None: row.username = _clean(username) if password is not None: @@ -230,16 +306,21 @@ def update_connector_profile_row( row.token_env = _clean(token_env) if secret_ref is not None: row.secret_ref = _clean(secret_ref) + + +def _update_profile_governance_fields( + row: FileConnectorProfile, + *, + capabilities: list[str] | None, + policy: Mapping[str, Any] | None, + metadata: Mapping[str, Any] | None, +) -> None: if capabilities is not None: row.capabilities = _string_list(capabilities) if policy is not None: row.policy = dict(policy) if metadata is not None: row.metadata_ = dict(metadata) - row.updated_by_user_id = user_id - session.add(row) - session.flush() - return row def _normalize_scope(*, tenant_id: str, scope_type: str, scope_id: str | None) -> tuple[str, str | None, str | None]: diff --git a/tests/test_connector_profile_store.py b/tests/test_connector_profile_store.py new file mode 100644 index 0000000..ecb352c --- /dev/null +++ b/tests/test_connector_profile_store.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_access.backend.db import models as access_models # noqa: F401 - resolve Files user foreign keys +from govoplan_core.db.base import Base +from govoplan_core.security.secrets import decrypt_secret, encrypt_secret +from govoplan_files.backend.db.models import FileConnectorProfile +from govoplan_files.backend.storage.common import FileStorageError +from govoplan_files.backend.storage.connector_profile_store import update_connector_profile_row + + +class ConnectorProfileStoreUpdateTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine, tables=[FileConnectorProfile.__table__]) + self.session = sessionmaker(bind=self.engine)() + self.row = FileConnectorProfile( + id="profile-1", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + label="Original profile", + provider="webdav", + endpoint_url="https://dav.example.test/original", + base_path="original", + enabled=True, + credential_mode="basic", + username="original-user", + password_encrypted=encrypt_secret("original-password"), + token_encrypted=encrypt_secret("original-token"), + capabilities=["browse"], + policy={"allow": {"providers": ["webdav"]}}, + metadata_={"original": True}, + created_by_user_id=None, + updated_by_user_id=None, + ) + self.session.add(self.row) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all(bind=self.engine) + self.engine.dispose() + + def test_update_preserves_normalization_and_flush_only_transaction_boundary(self) -> None: + updated = update_connector_profile_row( + self.session, + self.row, + user_id="user-2", + label=" Updated profile ", + provider="NEXTCLOUD", + endpoint_url=" https://cloud.example.test/dav ", + base_path=" shared/reports ", + enabled=False, + credential_profile_id=" credential-2 ", + credential_mode="TOKEN", + username=" updated-user ", + password="updated-password", + token="updated-token", + capabilities=["browse", " import ", ""], + policy={"deny": {"external_paths": ["private"]}}, + metadata={"department": "reports"}, + ) + + self.assertIs(self.row, updated) + self.assertEqual("Updated profile", updated.label) + self.assertEqual("nextcloud", updated.provider) + self.assertEqual("https://cloud.example.test/dav", updated.endpoint_url) + self.assertEqual("shared/reports", updated.base_path) + self.assertFalse(updated.enabled) + self.assertEqual("credential-2", updated.credential_profile_id) + self.assertEqual("token", updated.credential_mode) + self.assertEqual("updated-user", updated.username) + self.assertEqual("updated-password", decrypt_secret(updated.password_encrypted)) + self.assertEqual("updated-token", decrypt_secret(updated.token_encrypted)) + self.assertEqual(["browse", "import"], updated.capabilities) + self.assertEqual({"deny": {"external_paths": ["private"]}}, updated.policy) + self.assertEqual({"department": "reports"}, updated.metadata_) + self.assertEqual("user-2", updated.updated_by_user_id) + + self.session.rollback() + persisted = self.session.get(FileConnectorProfile, self.row.id) + assert persisted is not None + self.assertEqual("Original profile", persisted.label) + self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted)) + self.assertTrue(persisted.enabled) + + def test_secret_replacement_wins_over_clear_while_clear_removes_an_omitted_token(self) -> None: + update_connector_profile_row( + self.session, + self.row, + user_id="user-2", + password="replacement-password", + clear_password=True, + clear_token=True, + ) + + self.assertEqual("replacement-password", decrypt_secret(self.row.password_encrypted)) + self.assertIsNone(self.row.token_encrypted) + self.assertEqual("https://dav.example.test/original", self.row.endpoint_url) + self.assertEqual("original-user", self.row.username) + + def test_legacy_external_secret_reference_cannot_be_cleared_or_partially_mutate_the_row(self) -> None: + self.row.secret_ref = "vault:tenant-1:files:profile" + self.session.commit() + + with self.assertRaisesRegex(FileStorageError, "provider-side deletion"): + update_connector_profile_row( + self.session, + self.row, + user_id="user-2", + label="Must not be applied", + secret_ref="", + ) + + self.assertEqual("Original profile", self.row.label) + self.assertEqual("vault:tenant-1:files:profile", self.row.secret_ref) + + def test_flush_failure_remains_rollback_safe(self) -> None: + with patch.object(self.session, "flush", side_effect=RuntimeError("database unavailable")), self.assertRaisesRegex( + RuntimeError, + "database unavailable", + ): + update_connector_profile_row( + self.session, + self.row, + user_id="user-2", + label="Uncommitted profile", + password="uncommitted-password", + ) + + self.session.rollback() + persisted = self.session.get(FileConnectorProfile, self.row.id) + assert persisted is not None + self.assertEqual("Original profile", persisted.label) + self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted)) + + +if __name__ == "__main__": + unittest.main()