Security: Replace urllib.request with httpx in WebDAV testing
The `_test_webdav_connection` function previously used `urllib.request.urlopen` to verify connection credentials. This triggers a Bandit B310 warning because `urllib` supports multiple schemes (like file://, ftp://) and implicitly follows redirects. Although scheme checking and a basic `is_private_ip` validation were implemented, using `urllib.request` remains risky because a public URL could return an HTTP redirect to a private IP (e.g., 127.0.0.1) which `urllib` would blindly follow, causing an SSRF (Server-Side Request Forgery) bypass. This commit replaces `urllib.request` with `httpx.request` using explicitly `follow_redirects=False`. This eliminates the B310 vulnerability, ensures requests only hit the specified URL without following potentially malicious redirects, and standardizes the application on `httpx` for safer HTTP connections. In addition to fixing the vulnerability, test coverage is added for the new WebDAV connections logic. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
"""Tests for the per-user integrations API (app/api/integrations.py)."""
|
||||
|
||||
import unittest.mock
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
@@ -1061,6 +1062,69 @@ class TestConnectionTestEndpoint:
|
||||
assert data["success"] is False
|
||||
assert "scheme" in data["message"].lower()
|
||||
|
||||
@unittest.mock.patch("httpx.request")
|
||||
def test_test_webdav_success(self, mock_request, int_client):
|
||||
"""WebDAV test succeeds with valid credentials and a valid status code."""
|
||||
mock_response = unittest.mock.MagicMock()
|
||||
mock_response.status_code = 207 # Typical WebDAV success for PROPFIND
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
payload = {
|
||||
"integration_type": "WEBDAV",
|
||||
"config": {"url": "https://example.com/webdav"},
|
||||
"credentials": {"username": "user1", "password": "password123"},
|
||||
}
|
||||
resp = int_client.post("/api/integrations/test", json=payload)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is True
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"PROPFIND",
|
||||
"https://example.com/webdav",
|
||||
auth=("user1", "password123"),
|
||||
headers={"Depth": "0"},
|
||||
timeout=10.0,
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
@unittest.mock.patch("httpx.request")
|
||||
def test_test_webdav_failure_status(self, mock_request, int_client):
|
||||
"""WebDAV test fails if the server returns a 4xx or 5xx status code."""
|
||||
mock_response = unittest.mock.MagicMock()
|
||||
mock_response.status_code = 401
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
payload = {
|
||||
"integration_type": "WEBDAV",
|
||||
"config": {"url": "https://example.com/webdav"},
|
||||
"credentials": {"username": "user1", "password": "wrong"},
|
||||
}
|
||||
resp = int_client.post("/api/integrations/test", json=payload)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is False
|
||||
assert "401" in data["message"]
|
||||
|
||||
@unittest.mock.patch("httpx.request")
|
||||
def test_test_webdav_exception(self, mock_request, int_client):
|
||||
"""WebDAV test fails gracefully if an exception occurs during the request."""
|
||||
mock_request.side_effect = Exception("Connection error")
|
||||
|
||||
payload = {
|
||||
"integration_type": "WEBDAV",
|
||||
"config": {"url": "https://example.com/webdav"},
|
||||
"credentials": {},
|
||||
}
|
||||
resp = int_client.post("/api/integrations/test", json=payload)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is False
|
||||
assert "failed" in data["message"].lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Quota endpoint tests
|
||||
|
||||
Reference in New Issue
Block a user