From 1444ba80a0da5fc4d8f5f4f8f541cacc1124a097 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 18:52:33 +0200 Subject: [PATCH] refactor(files): centralize connector visibility --- src/govoplan_files/backend/router.py | 91 ++----- .../backend/storage/connector_visibility.py | 180 +++++++++++++ tests/test_connector_visibility.py | 240 ++++++++++++++++++ 3 files changed, 446 insertions(+), 65 deletions(-) create mode 100644 src/govoplan_files/backend/storage/connector_visibility.py create mode 100644 tests/test_connector_visibility.py diff --git a/src/govoplan_files/backend/router.py b/src/govoplan_files/backend/router.py index ea6b27a..ff02ead 100644 --- a/src/govoplan_files/backend/router.py +++ b/src/govoplan_files/backend/router.py @@ -3,7 +3,6 @@ from __future__ import annotations import json import os import tempfile -from dataclasses import replace from datetime import datetime from io import BytesIO from typing import Any, Literal @@ -131,11 +130,11 @@ from govoplan_files.backend.storage.connector_profile_store import ( connector_profile_from_row, create_connector_profile_row, get_connector_profile_row, - list_database_connector_profiles, update_connector_profile_row, ) -from govoplan_files.backend.storage.connector_profiles import ConnectorProfile, connector_profiles_from_settings +from govoplan_files.backend.storage.connector_profiles import ConnectorProfile from govoplan_files.backend.storage.connector_providers import connector_provider_descriptors +from govoplan_files.backend.storage.connector_visibility import visible_connector_profiles_for_actor from govoplan_files.backend.storage.connector_spaces import ( connector_space_owner_id, create_connector_space, @@ -455,35 +454,6 @@ def _enforce_connector_policy(source_provenance_json: str | None, connector_poli ensure_connector_policy_allows(request, sources) -def _connector_profile_visible( - session: Session, - principal: ApiPrincipal, - profile: ConnectorProfile, - *, - campaign_id: str | None, - group_ids: set[str], -) -> bool: - if profile.scope_type == "system": - return True - if profile.scope_type == "tenant": - return profile.scope_id == principal.tenant_id - if profile.scope_type == "user": - return profile.scope_id == principal.user.id - if profile.scope_type == "group": - return bool(profile.scope_id and profile.scope_id in group_ids) - if profile.scope_type == "campaign": - if not profile.scope_id: - return False - if campaign_id and profile.scope_id != campaign_id: - return False - try: - _ensure_campaign_file_access(session, principal, profile.scope_id) - except HTTPException: - return False - return True - return False - - def _visible_connector_profiles( session: Session, principal: ApiPrincipal, @@ -494,30 +464,33 @@ def _visible_connector_profiles( include_admin_scopes: bool = False, include_effective_policy: bool = True, ) -> list[ConnectorProfile]: - provider_norm = provider.strip().casefold() if provider else None - group_ids = set(user_group_ids(session, tenant_id=principal.tenant_id, user_id=principal.user.id, include_admin_groups=_is_admin(principal))) - database_profiles = list_database_connector_profiles( + group_ids = user_group_ids( session, tenant_id=principal.tenant_id, - include_disabled=include_disabled, + user_id=principal.user.id, + include_admin_groups=_is_admin(principal), + ) + + def campaign_visible(profile_campaign_id: str) -> bool: + try: + _ensure_campaign_file_access(session, principal, profile_campaign_id) + except HTTPException: + return False + return True + + return visible_connector_profiles_for_actor( + session, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + group_ids=group_ids, + settings=settings, + provider=provider, + campaign_id=campaign_id, + campaign_visible=campaign_visible, + include_disabled=include_disabled, + include_admin_scopes=include_admin_scopes, + include_effective_policy=include_effective_policy, ) - env_profiles = connector_profiles_from_settings(settings) - profiles_by_id: dict[str, ConnectorProfile] = {} - for profile in [*database_profiles, *env_profiles]: - if profile.id not in profiles_by_id: - profiles_by_id[profile.id] = profile - profiles = list(profiles_by_id.values()) - visible: list[ConnectorProfile] = [] - for profile in profiles: - if not (profile.enabled or include_disabled): - continue - if provider_norm is not None and profile.provider != provider_norm: - continue - admin_scope_visible = include_admin_scopes and profile.scope_type in {"user", "group", "campaign"} - if not admin_scope_visible and not _connector_profile_visible(session, principal, profile, campaign_id=campaign_id, group_ids=group_ids): - continue - visible.append(_with_effective_connector_policy(session, principal, profile) if include_effective_policy else profile) - return visible def _webdav_discovery_candidates(payload: FileConnectorDiscoveryRequest) -> list[str]: @@ -641,18 +614,6 @@ def _visible_connector_profile( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connector profile not found") -def _with_effective_connector_policy(session: Session, principal: ApiPrincipal, profile: ConnectorProfile) -> ConnectorProfile: - sources = effective_connector_policy_sources( - session, - tenant_id=principal.tenant_id, - scope_type=profile.scope_type, - scope_id=profile.scope_id, - ) - if not sources: - return profile - return replace(profile, policy_sources=tuple([*sources, *profile.policy_sources])) - - def _require_connector_profile_write(principal: ApiPrincipal, scope_type: str) -> None: clean_scope = scope_type.strip().casefold() if clean_scope == "system": diff --git a/src/govoplan_files/backend/storage/connector_visibility.py b/src/govoplan_files/backend/storage/connector_visibility.py new file mode 100644 index 0000000..78a534f --- /dev/null +++ b/src/govoplan_files/backend/storage/connector_visibility.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +from collections.abc import Callable, Iterable +from dataclasses import replace + +from sqlalchemy.orm import Session + +from govoplan_files.backend.storage.connector_deployment import ( + connector_effective_endpoint_url, +) +from govoplan_files.backend.storage.connector_profile_store import ( + list_database_connector_profiles, +) +from govoplan_files.backend.storage.connector_profiles import ( + ConnectorProfile, + connector_profiles_from_settings, +) +from govoplan_files.backend.storage.connector_policy import ( + ConnectorAccessRequest, + connector_policy_decision, +) +from govoplan_files.backend.storage.connector_policy_store import ( + effective_connector_policy_sources, +) +from govoplan_files.backend.storage.connector_providers import ( + connector_provider_descriptors, +) + + +CampaignVisibility = Callable[[str], bool] + + +def visible_connector_profiles_for_actor( + session: Session, + *, + tenant_id: str, + user_id: str, + group_ids: Iterable[str], + settings: object | None, + provider: str | None = None, + campaign_id: str | None = None, + campaign_visible: CampaignVisibility | None = None, + include_disabled: bool = False, + include_admin_scopes: bool = False, + include_effective_policy: bool = True, +) -> list[ConnectorProfile]: + """Return the same actor-filtered connector set used by API and docs surfaces.""" + + provider_norm = provider.strip().casefold() if provider else None + actor_group_ids = {str(group_id) for group_id in group_ids if str(group_id)} + database_profiles = list_database_connector_profiles( + session, + tenant_id=tenant_id, + include_disabled=include_disabled, + ) + configured_profiles = connector_profiles_from_settings(settings) + profiles_by_id: dict[str, ConnectorProfile] = {} + for profile in [*database_profiles, *configured_profiles]: + if profile.id not in profiles_by_id: + profiles_by_id[profile.id] = profile + + visible: list[ConnectorProfile] = [] + for profile in profiles_by_id.values(): + if not (profile.enabled or include_disabled): + continue + if provider_norm is not None and profile.provider != provider_norm: + continue + admin_scope_visible = include_admin_scopes and profile.scope_type in { + "user", + "group", + "campaign", + } + if not admin_scope_visible and not _profile_visible_to_actor( + profile, + tenant_id=tenant_id, + user_id=user_id, + group_ids=actor_group_ids, + campaign_id=campaign_id, + campaign_visible=campaign_visible, + ): + continue + visible.append( + _with_effective_connector_policy( + session, tenant_id=tenant_id, profile=profile + ) + if include_effective_policy + else profile + ) + return visible + + +def connector_profile_usable_for_import(profile: ConnectorProfile) -> bool: + """Whether a visible profile can safely offer the current live browse/import task.""" + + if ( + not profile.enabled + or not profile.endpoint_url + or not profile.credentials_configured + ): + return False + descriptor = next( + ( + item + for item in connector_provider_descriptors() + if item.provider == profile.provider + ), + None, + ) + if descriptor is None: + return False + if not ( + descriptor.implemented + and descriptor.installed + and descriptor.browse_supported + and descriptor.import_supported + ): + return False + # These SDK paths intentionally fail closed until every connection peer and + # SDK-managed redirect/referral can be pinned and revalidated. + if profile.provider in {"s3", "smb"}: + return False + + # Prove that the initial root browse performed by the current Files UI is + # policy-allowed. A later selected remote path/item is checked again. + return connector_policy_decision( + ConnectorAccessRequest( + connector_id=profile.id, + credential_id=profile.credential_profile_id, + provider=profile.provider, + external_path="", + external_url=connector_effective_endpoint_url( + provider=profile.provider, + endpoint_url=profile.endpoint_url, + metadata=profile.metadata, + ), + operation="import", + ), + profile.policy_sources, + ).allowed + + +def _profile_visible_to_actor( + profile: ConnectorProfile, + *, + tenant_id: str, + user_id: str, + group_ids: set[str], + campaign_id: str | None, + campaign_visible: CampaignVisibility | None, +) -> bool: + if profile.scope_type == "system": + return True + if profile.scope_type == "tenant": + return profile.scope_id == tenant_id + if profile.scope_type == "user": + return profile.scope_id == user_id + if profile.scope_type == "group": + return bool(profile.scope_id and profile.scope_id in group_ids) + if profile.scope_type != "campaign" or not profile.scope_id: + return False + if campaign_id and profile.scope_id != campaign_id: + return False + return bool(campaign_visible and campaign_visible(profile.scope_id)) + + +def _with_effective_connector_policy( + session: Session, + *, + tenant_id: str, + profile: ConnectorProfile, +) -> ConnectorProfile: + sources = effective_connector_policy_sources( + session, + tenant_id=tenant_id, + scope_type=profile.scope_type, + scope_id=profile.scope_id, + ) + if not sources: + return profile + return replace(profile, policy_sources=tuple([*sources, *profile.policy_sources])) diff --git a/tests/test_connector_visibility.py b/tests/test_connector_visibility.py new file mode 100644 index 0000000..e044ad6 --- /dev/null +++ b/tests/test_connector_visibility.py @@ -0,0 +1,240 @@ +from __future__ import annotations + +import unittest +from dataclasses import replace +from unittest.mock import patch + +from govoplan_files.backend.storage.connector_policy import ConnectorPolicySource +from govoplan_files.backend.storage.connector_profiles import ConnectorProfile +from govoplan_files.backend.storage.connector_visibility import ( + connector_profile_usable_for_import, + visible_connector_profiles_for_actor, +) + + +def _profile( + profile_id: str, + *, + scope_type: str = "system", + scope_id: str | None = None, + provider: str = "webdav", + enabled: bool = True, +) -> ConnectorProfile: + return ConnectorProfile( + id=profile_id, + label=profile_id, + provider=provider, + scope_type=scope_type, + scope_id=scope_id, + endpoint_url=f"https://{profile_id}.example.invalid", + enabled=enabled, + credential_mode="anonymous", + ) + + +class ConnectorVisibilityTests(unittest.TestCase): + @patch( + "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources", + return_value=[], + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings" + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles" + ) + def test_profiles_are_filtered_to_actor_scopes_and_database_definition_wins( + self, + database_profiles, + configured_profiles, + _policy_sources, + ) -> None: + database_profiles.return_value = [ + _profile("system"), + _profile("tenant", scope_type="tenant", scope_id="tenant-1"), + _profile("other-tenant", scope_type="tenant", scope_id="tenant-2"), + _profile("user", scope_type="user", scope_id="user-1"), + _profile("other-user", scope_type="user", scope_id="user-2"), + _profile("group", scope_type="group", scope_id="group-1"), + _profile("campaign", scope_type="campaign", scope_id="campaign-1"), + _profile("disabled", enabled=False), + _profile("duplicate", provider="webdav"), + ] + configured_profiles.return_value = [_profile("duplicate", provider="nextcloud")] + + visible = visible_connector_profiles_for_actor( + object(), # type: ignore[arg-type] + tenant_id="tenant-1", + user_id="user-1", + group_ids={"group-1"}, + settings=object(), + campaign_visible=lambda campaign_id: campaign_id == "campaign-1", + ) + + self.assertEqual( + {"system", "tenant", "user", "group", "campaign", "duplicate"}, + {profile.id for profile in visible}, + ) + duplicate = next(profile for profile in visible if profile.id == "duplicate") + self.assertEqual("webdav", duplicate.provider) + + @patch( + "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources", + return_value=[], + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings", + return_value=[], + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles" + ) + def test_campaign_and_admin_visibility_remain_explicit( + self, + database_profiles, + _configured_profiles, + _policy_sources, + ) -> None: + database_profiles.return_value = [ + _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"), + ] + + campaign_only = visible_connector_profiles_for_actor( + object(), # type: ignore[arg-type] + tenant_id="tenant-1", + user_id="user-1", + group_ids=(), + settings=None, + campaign_id="campaign-a", + campaign_visible=lambda _campaign_id: True, + ) + self.assertEqual(["campaign-a"], [profile.id for profile in campaign_only]) + + admin = visible_connector_profiles_for_actor( + object(), # type: ignore[arg-type] + tenant_id="tenant-1", + user_id="user-1", + group_ids=(), + settings=None, + campaign_visible=None, + include_admin_scopes=True, + ) + self.assertEqual( + {"campaign-a", "campaign-b", "other-user"}, + {profile.id for profile in admin}, + ) + + @patch( + "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings", + return_value=[], + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.list_database_connector_profiles" + ) + @patch( + "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources" + ) + def test_effective_policy_is_attached_without_mutating_profile( + self, + policy_sources, + database_profiles, + _configured_profiles, + ) -> None: + profile = _profile("profile") + source = ConnectorPolicySource( + scope_type="system", + label="System", + policy={"allow": {"providers": ["webdav"]}}, + ) + database_profiles.return_value = [profile] + policy_sources.return_value = [source] + + visible = visible_connector_profiles_for_actor( + object(), # type: ignore[arg-type] + tenant_id="tenant-1", + user_id="user-1", + group_ids=(), + settings=None, + ) + + self.assertEqual((), profile.policy_sources) + self.assertEqual((source,), visible[0].policy_sources) + + def test_import_usability_requires_safe_provider_credentials_endpoint_and_identity_policy( + self, + ) -> None: + self.assertTrue(connector_profile_usable_for_import(_profile("webdav"))) + self.assertFalse( + connector_profile_usable_for_import(_profile("s3", provider="s3")) + ) + self.assertFalse( + connector_profile_usable_for_import(_profile("smb", provider="smb")) + ) + self.assertFalse( + connector_profile_usable_for_import( + _profile("sharepoint", provider="sharepoint") + ) + ) + self.assertFalse( + connector_profile_usable_for_import( + ConnectorProfile( + id="no-endpoint", + label="No endpoint", + provider="webdav", + credential_mode="anonymous", + ) + ) + ) + self.assertFalse( + connector_profile_usable_for_import( + ConnectorProfile( + id="no-credentials", + label="No credentials", + provider="webdav", + endpoint_url="https://files.example.invalid", + credential_mode="basic", + ) + ) + ) + + denied = replace( + _profile("denied"), + policy_sources=( + ConnectorPolicySource( + scope_type="system", + label="System", + policy={"deny": {"providers": ["webdav"]}}, + ), + ), + ) + self.assertFalse(connector_profile_usable_for_import(denied)) + + path_limited = replace( + _profile("path-limited"), + policy_sources=( + ConnectorPolicySource( + scope_type="system", + label="System", + policy={"allow": {"external_paths": ["/approved/*"]}}, + ), + ), + ) + self.assertFalse(connector_profile_usable_for_import(path_limited)) + + endpoint_allowed = replace( + _profile("endpoint-allowed"), + policy_sources=( + ConnectorPolicySource( + scope_type="system", + label="System", + policy={"allow": {"external_urls": ["https://*.example.invalid"]}}, + ), + ), + ) + self.assertTrue(connector_profile_usable_for_import(endpoint_allowed)) + + +if __name__ == "__main__": + unittest.main()