diff --git a/tests/test_api_integrations.py b/tests/test_api_integrations.py index 958a1a86..561ac177 100644 --- a/tests/test_api_integrations.py +++ b/tests/test_api_integrations.py @@ -998,6 +998,23 @@ class TestConnectionTestEndpoint: assert data["success"] is False assert "Missing" in data["message"] + def test_test_imap_blocks_private_ip(self, int_client): + """IMAP test with private IP returns failure (SSRF protection).""" + payload = { + "integration_type": "IMAP", + "config": { + "host": "127.0.0.1", + "port": 993, + "username": "user", + }, + "credentials": {"password": "pass"}, + } + resp = int_client.post("/api/integrations/test", json=payload) + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is False + assert "Invalid hostname or IP address" in data["message"] + def test_test_s3_missing_bucket(self, int_client): """S3 test with missing bucket returns failure.""" payload = { @@ -1011,6 +1028,19 @@ class TestConnectionTestEndpoint: assert data["success"] is False assert "bucket" in data["message"].lower() + def test_test_s3_blocks_private_ip(self, int_client): + """S3 test with private IP endpoint returns failure (SSRF protection).""" + payload = { + "integration_type": "S3", + "config": {"bucket": "my-bucket", "endpoint_url": "http://127.0.0.1:9000"}, + "credentials": {"access_key_id": "AKIA", "secret_access_key": "secret"}, + } + resp = int_client.post("/api/integrations/test", json=payload) + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is False + assert "Invalid endpoint URL or private IP" in data["message"] + def test_test_webdav_missing_url(self, int_client): """WebDAV test with missing URL returns failure.""" payload = {