Harden external file connector boundaries
This commit is contained in:
196
tests/test_connector_deployment.py
Normal file
196
tests/test_connector_deployment.py
Normal file
@@ -0,0 +1,196 @@
|
||||
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_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()
|
||||
Reference in New Issue
Block a user