security(files): filter connectors before secret resolution
This commit is contained in:
@@ -11,7 +11,10 @@ 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
|
||||
from govoplan_files.backend.storage.connector_profile_store import (
|
||||
list_database_connector_profiles,
|
||||
update_connector_profile_row,
|
||||
)
|
||||
|
||||
|
||||
class ConnectorProfileStoreUpdateTests(unittest.TestCase):
|
||||
@@ -140,6 +143,34 @@ class ConnectorProfileStoreUpdateTests(unittest.TestCase):
|
||||
self.assertEqual("Original profile", persisted.label)
|
||||
self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted))
|
||||
|
||||
def test_visibility_filter_runs_before_connector_secrets_are_decrypted(self) -> None:
|
||||
invisible = FileConnectorProfile(
|
||||
id="invisible-profile",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="user",
|
||||
scope_id="another-user",
|
||||
label="Invisible profile",
|
||||
provider="webdav",
|
||||
endpoint_url="https://invisible.example.test",
|
||||
enabled=True,
|
||||
credential_mode="basic",
|
||||
password_encrypted="not-a-valid-encrypted-secret",
|
||||
capabilities=["browse"],
|
||||
policy={},
|
||||
metadata_={},
|
||||
)
|
||||
self.session.add(invisible)
|
||||
self.session.commit()
|
||||
|
||||
profiles = list_database_connector_profiles(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
row_visible=lambda row: row.id == self.row.id,
|
||||
)
|
||||
|
||||
self.assertEqual([self.row.id], [profile.id for profile in profiles])
|
||||
self.assertEqual("original-password", profiles[0].password_value)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -41,7 +41,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
"govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings"
|
||||
)
|
||||
@patch(
|
||||
"govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles"
|
||||
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
||||
)
|
||||
def test_profiles_are_filtered_to_actor_scopes_and_database_definition_wins(
|
||||
self,
|
||||
@@ -49,7 +49,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
configured_profiles,
|
||||
_policy_sources,
|
||||
) -> None:
|
||||
database_profiles.return_value = [
|
||||
profiles = [
|
||||
_profile("system"),
|
||||
_profile("tenant", scope_type="tenant", scope_id="tenant-1"),
|
||||
_profile("other-tenant", scope_type="tenant", scope_id="tenant-2"),
|
||||
@@ -60,6 +60,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
_profile("disabled", enabled=False),
|
||||
_profile("duplicate", provider="webdav"),
|
||||
]
|
||||
database_profiles.return_value = (profiles, {profile.id for profile in profiles})
|
||||
configured_profiles.return_value = [_profile("duplicate", provider="nextcloud")]
|
||||
|
||||
visible = visible_connector_profiles_for_actor(
|
||||
@@ -87,7 +88,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
return_value=[],
|
||||
)
|
||||
@patch(
|
||||
"govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles"
|
||||
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
||||
)
|
||||
def test_campaign_and_admin_visibility_remain_explicit(
|
||||
self,
|
||||
@@ -95,11 +96,12 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
_configured_profiles,
|
||||
_policy_sources,
|
||||
) -> None:
|
||||
database_profiles.return_value = [
|
||||
profiles = [
|
||||
_profile("campaign-a", scope_type="campaign", scope_id="campaign-a"),
|
||||
_profile("campaign-b", scope_type="campaign", scope_id="campaign-b"),
|
||||
_profile("other-user", scope_type="user", scope_id="user-2"),
|
||||
]
|
||||
database_profiles.return_value = (profiles, {profile.id for profile in profiles})
|
||||
|
||||
campaign_only = visible_connector_profiles_for_actor(
|
||||
object(), # type: ignore[arg-type]
|
||||
@@ -131,7 +133,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
return_value=[],
|
||||
)
|
||||
@patch(
|
||||
"govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles"
|
||||
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
||||
)
|
||||
@patch(
|
||||
"govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources"
|
||||
@@ -148,7 +150,7 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
label="System",
|
||||
policy={"allow": {"providers": ["webdav"]}},
|
||||
)
|
||||
database_profiles.return_value = [profile]
|
||||
database_profiles.return_value = ([profile], {profile.id})
|
||||
policy_sources.return_value = [source]
|
||||
|
||||
visible = visible_connector_profiles_for_actor(
|
||||
@@ -198,6 +200,29 @@ class ConnectorVisibilityTests(unittest.TestCase):
|
||||
)
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
connector_profile_usable_for_import(
|
||||
ConnectorProfile(
|
||||
id="metadata-endpoint",
|
||||
label="Metadata endpoint",
|
||||
provider="webdav",
|
||||
credential_mode="anonymous",
|
||||
metadata={"webdav_endpoint_url": "https://files.example.invalid"},
|
||||
)
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
connector_profile_usable_for_import(
|
||||
ConnectorProfile(
|
||||
id="unresolved-secret-reference",
|
||||
label="Unresolved secret reference",
|
||||
provider="webdav",
|
||||
endpoint_url="https://files.example.invalid",
|
||||
credential_mode="basic",
|
||||
secret_ref="vault://files/webdav",
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
denied = replace(
|
||||
_profile("denied"),
|
||||
|
||||
Reference in New Issue
Block a user