from __future__ import annotations import os import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from unittest.mock import MagicMock from fastapi import HTTPException from govoplan_files.backend.router import discover_connector_endpoint from govoplan_files.backend.schemas import FileConnectorDiscoveryRequest from govoplan_files.backend.storage.connector_browse import ConnectorBrowseError, _profile_password, _s3_verify from govoplan_files.backend.storage.connector_deployment import ( ConnectorDeploymentConfigurationError, connector_effective_endpoint_url, connector_secret_env_value, reject_api_controlled_deployment_references, ) from govoplan_files.backend.storage.connector_profiles import ConnectorProfile from govoplan_files.backend.storage.connector_profile_store import create_connector_profile_row class ConnectorDeploymentBoundaryTests(unittest.TestCase): def test_database_profile_cannot_read_arbitrary_process_environment(self) -> None: profile = ConnectorProfile( id="tenant-webdav", label="Tenant WebDAV", provider="webdav", password_env="MASTER_KEY_B64", source_kind="database", ) with patch.dict( os.environ, { "MASTER_KEY_B64": "must-not-leave-process", "GOVOPLAN_CONNECTOR_SECRET_ENV_ALLOWLIST": "MASTER_KEY_B64", }, clear=False, ), self.assertRaisesRegex(ConnectorBrowseError, "deployment-owned"): _profile_password(profile) def test_deployment_profile_requires_exact_secret_allowlist(self) -> None: with patch.dict( os.environ, { "GOVOPLAN_FILES_WEBDAV_PASSWORD": "deployment-secret", "GOVOPLAN_CONNECTOR_SECRET_ENV_ALLOWLIST": "GOVOPLAN_FILES_WEBDAV_PASSWORD", }, clear=False, ): self.assertEqual( "deployment-secret", connector_secret_env_value( "GOVOPLAN_FILES_WEBDAV_PASSWORD", source_kind="settings", ), ) with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "not listed"): connector_secret_env_value("MASTER_KEY_B64", source_kind="settings") def test_api_profiles_cannot_select_secret_environment_names(self) -> None: with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "deployment-owned"): reject_api_controlled_deployment_references(password_env="DATABASE_URL") with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "deployment-owned"): reject_api_controlled_deployment_references(metadata={"secret_access_key_env": "MASTER_KEY_B64"}) session = MagicMock() with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "deployment-owned"): create_connector_profile_row( session, tenant_id="tenant-1", user_id="user-1", profile_id="unsafe", label="Unsafe", provider="webdav", scope_type="tenant", password_env="DATABASE_URL", ) session.get.assert_not_called() def test_api_profiles_cannot_create_unowned_external_secret_references(self) -> None: with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "provider ownership"): reject_api_controlled_deployment_references(secret_ref="vault:tenant-1:files") session = MagicMock() with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "provider ownership"): create_connector_profile_row( session, tenant_id="tenant-1", user_id="user-1", profile_id="unsafe-secret-ref", label="Unsafe secret reference", provider="webdav", scope_type="tenant", credential_mode="secret_ref", secret_ref="vault:tenant-1:files", ) session.get.assert_not_called() def test_api_profiles_cannot_hide_plaintext_credentials_in_metadata(self) -> None: for metadata in ( {"secret_access_key": "plaintext-secret"}, {"credentials": {"password": "plaintext-secret"}}, {"provider": {"refresh_token": "plaintext-secret"}}, ): with self.subTest(metadata=metadata), self.assertRaisesRegex( ConnectorDeploymentConfigurationError, "dedicated encrypted credential fields", ): reject_api_controlled_deployment_references(metadata=metadata) def test_profile_responses_hide_environment_and_local_ca_references(self) -> None: profile = ConnectorProfile( id="deployment-s3", label="Deployment S3", provider="s3", metadata={ "bucket": "documents", "secret_access_key_env": "GOVOPLAN_FILES_S3_SECRET", "ca_bundle": "/etc/govoplan/connector-ca.pem", }, ) self.assertEqual({"bucket": "documents"}, profile.to_response()["metadata"]) def test_ca_bundle_must_be_an_exact_deployment_allowlisted_file(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: allowed = Path(temp_dir, "connector-ca.pem") other = Path(temp_dir, "other-ca.pem") allowed.write_text("test CA", encoding="utf-8") other.write_text("other CA", encoding="utf-8") with patch.dict( os.environ, { "APP_ENV": "production", "GOVOPLAN_CONNECTOR_CA_BUNDLE_ALLOWLIST": str(allowed), }, clear=False, ): profile = ConnectorProfile( id="s3", label="S3", provider="s3", metadata={"ca_bundle": str(allowed)}, ) self.assertEqual(str(allowed.resolve()), _s3_verify(profile)) with self.assertRaisesRegex(ConnectorDeploymentConfigurationError, "not listed"): reject_api_controlled_deployment_references(metadata={"ca_bundle": str(other)}) def test_tls_verification_can_only_be_disabled_in_development(self) -> None: with patch.dict(os.environ, {"APP_ENV": "production"}, clear=False), self.assertRaisesRegex( ConnectorDeploymentConfigurationError, "dev/test", ): reject_api_controlled_deployment_references(metadata={"verify_tls": False}) with patch.dict(os.environ, {"APP_ENV": "test"}, clear=False): reject_api_controlled_deployment_references(metadata={"verify_tls": False}) def test_effective_endpoint_uses_webdav_override(self) -> None: self.assertEqual( "https://dav.example.test/root", connector_effective_endpoint_url( provider="seafile", endpoint_url="https://seafile.example.test", metadata={"webdav_endpoint_url": "https://dav.example.test/root"}, ), ) class ConnectorDiscoveryBoundaryTests(unittest.TestCase): @staticmethod def _principal() -> SimpleNamespace: return SimpleNamespace(tenant_id="tenant-1", user=SimpleNamespace(id="user-1"), api_key=None) def test_discovery_rejects_environment_credentials_before_io(self) -> None: payload = FileConnectorDiscoveryRequest( provider="webdav", endpoint_url="https://dav.example.test", credential_mode="basic", credentials={"username": "admin", "password_env": "MASTER_KEY_B64"}, ) with patch("govoplan_files.backend.router.browse_connector_profile") as browse, self.assertRaises( HTTPException ) as raised: discover_connector_endpoint(payload, session=object(), principal=self._principal()) # type: ignore[arg-type] self.assertEqual(400, raised.exception.status_code) browse.assert_not_called() def test_discovery_applies_policy_and_audit_before_each_io_candidate(self) -> None: payload = FileConnectorDiscoveryRequest( provider="webdav", endpoint_url="https://dav.example.test/root", metadata={ "webdav_endpoint_url": "https://bypass.example.test", "static_listing": {"": []}, }, ) events: list[str] = [] def ensure(*_args: object, **kwargs: object) -> None: self.assertEqual("https://dav.example.test/root/", kwargs["endpoint_url"]) events.append("policy") def audit(*_args: object, **_kwargs: object) -> None: events.append("audit") def browse(profile: ConnectorProfile, **_kwargs: object) -> list[object]: self.assertEqual({}, profile.metadata) events.append("io") return [] with patch("govoplan_files.backend.router._ensure_connector_configuration_allowed", side_effect=ensure), patch( "govoplan_files.backend.router._audit_connector_discovery_attempt", side_effect=audit, ), patch("govoplan_files.backend.router.browse_connector_profile", side_effect=browse): response = discover_connector_endpoint(payload, session=object(), principal=self._principal()) # type: ignore[arg-type] self.assertEqual("usable", response.status) self.assertEqual(["policy", "audit", "io"], events) self.assertEqual({"discovered_by": "webdav-propfind"}, response.metadata) if __name__ == "__main__": unittest.main()