Close as obsolete
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -1029,7 +1029,7 @@ class TestConnectionTestEndpoint:
|
||||
assert "bucket" in data["message"].lower()
|
||||
|
||||
def test_test_s3_blocks_private_ip(self, int_client):
|
||||
"""S3 test blocks requests to private/internal IPs via endpoint_url (SSRF protection)."""
|
||||
"""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"},
|
||||
@@ -1039,59 +1039,7 @@ class TestConnectionTestEndpoint:
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is False
|
||||
assert "internal" in data["message"].lower() or "private" in data["message"].lower()
|
||||
|
||||
def test_test_s3_blocks_localhost(self, int_client):
|
||||
"""S3 test blocks requests to localhost via endpoint_url."""
|
||||
payload = {
|
||||
"integration_type": "S3",
|
||||
"config": {"bucket": "my-bucket", "endpoint_url": "http://localhost/s3"},
|
||||
"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 "internal" in data["message"].lower() or "private" in data["message"].lower()
|
||||
|
||||
def test_test_s3_blocks_file_scheme(self, int_client):
|
||||
"""S3 test blocks file:// scheme via endpoint_url."""
|
||||
payload = {
|
||||
"integration_type": "S3",
|
||||
"config": {"bucket": "my-bucket", "endpoint_url": "file:///etc/passwd"},
|
||||
"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 "scheme" in data["message"].lower()
|
||||
|
||||
def test_test_s3_blocks_non_string_endpoint_url(self, int_client):
|
||||
"""S3 test rejects non-string endpoint_url values gracefully (no 500)."""
|
||||
payload = {
|
||||
"integration_type": "S3",
|
||||
"config": {"bucket": "my-bucket", "endpoint_url": 12345},
|
||||
"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 "string" in data["message"].lower()
|
||||
|
||||
def test_test_s3_blocks_empty_hostname_endpoint_url(self, int_client):
|
||||
"""S3 test rejects endpoint_url with no hostname (e.g. malformed URL)."""
|
||||
payload = {
|
||||
"integration_type": "S3",
|
||||
"config": {"bucket": "my-bucket", "endpoint_url": "https://"},
|
||||
"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 "hostname" in data["message"].lower()
|
||||
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."""
|
||||
|
||||
Reference in New Issue
Block a user