gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #808 from christianlouis/fix/double-slashes-join-url-12822045781097996485

Browse Source 
Fix double slashes again
This commit is contained in:
Christian Krakau-Louis
2026-03-23 17:27:08 +01:00
committed by GitHub
parent a052b2fbe0 a384b222f1
commit 2a5296d7e7
4 changed files with 241 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+22
View File
@@ -130,6 +130,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## Unreleased ## Unreleased
## v0.172.2 (2026-03-23)
### Bug Fixes
- Adapt TemplateResponse calls to Starlette 1.0 new-style API
([`c4e10be`](https://github.com/christianlouis/DocuElevate/commit/c4e10bee5e096e71a5bc4fac4928f69e5c04f2fb))
- Update test assertions and lint fixes for Starlette 1.0 TemplateResponse API
([`93629ff`](https://github.com/christianlouis/DocuElevate/commit/93629ff44083d43f79fdd49431457023e53d13e4))
- **build**: Remove --omit=dev from npm ci in Dockerfile frontend-builder stage
([`b4e0067`](https://github.com/christianlouis/DocuElevate/commit/b4e0067a27e2fb161349bd38c6d3b3f3bcb86972))
### Documentation
- **changelog**: Update changelog [skip ci]
([`0841713`](https://github.com/christianlouis/DocuElevate/commit/084171395d1076c716aa500a516118db49468ff5))
## Unreleased
## v0.172.1 (2026-03-22) ## v0.172.1 (2026-03-22)
### Bug Fixes ### Bug Fixes
app/tasks/upload_to_nextcloud.py
+141 -157
View File
@@ -1,157 +1,141 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import logging import logging
import os import os
import requests import requests
from requests.auth import HTTPBasicAuth from requests.auth import HTTPBasicAuth
from app.celery_app import celery from app.celery_app import celery
from app.config import settings from app.config import settings
from app.tasks.retry_config import UploadTaskWithRetry from app.tasks.retry_config import UploadTaskWithRetry
from app.utils import log_task_progress from app.utils import log_task_progress
from app.utils.filename_utils import extract_remote_path, get_unique_filename from app.utils.filename_utils import extract_remote_path, get_unique_filename
from app.utils.network import join_url
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
@celery.task(base=UploadTaskWithRetry, bind=True)
def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None): @celery.task(base=UploadTaskWithRetry, bind=True)
""" def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
Upload a file to Nextcloud WebDAV. """
Upload a file to Nextcloud WebDAV.
Args:
file_path: Path to the file to upload Args:
file_id: Optional file ID to associate with logs file_path: Path to the file to upload
""" file_id: Optional file ID to associate with logs
task_id = self.request.id """
logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}") task_id = self.request.id
log_task_progress( logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
task_id, log_task_progress(
"upload_to_nextcloud", task_id,
"in_progress", "upload_to_nextcloud",
f"Uploading to Nextcloud: {os.path.basename(file_path)}", "in_progress",
file_id=file_id, f"Uploading to Nextcloud: {os.path.basename(file_path)}",
) file_id=file_id,
)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}" if not os.path.exists(file_path):
logger.error(f"[{task_id}] {error_msg}") error_msg = f"File not found: {file_path}"
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) logger.error(f"[{task_id}] {error_msg}")
raise FileNotFoundError(error_msg) log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
# This is what's shown in your env view # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
if not ( # This is what's shown in your env view
getattr(settings, "nextcloud_upload_url", None) if not (
and getattr(settings, "nextcloud_username", None) getattr(settings, "nextcloud_upload_url", None)
and getattr(settings, "nextcloud_password", None) and getattr(settings, "nextcloud_username", None)
): and getattr(settings, "nextcloud_password", None)
logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration") ):
log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id) logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
return {"status": "Skipped", "reason": "Nextcloud settings not configured"} log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
filename = os.path.basename(file_path)
filename = os.path.basename(file_path)
try:
# Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url try:
webdav_url = settings.nextcloud_upload_url # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
if not webdav_url.endswith("/"): webdav_url = settings.nextcloud_upload_url
webdav_url += "/" if not webdav_url.endswith("/"):
webdav_url += "/"
# Calculate remote path based on local file structure
remote_base = ( # Calculate remote path based on local file structure
folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "") remote_base = (
) folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
remote_path = extract_remote_path(file_path, settings.workdir, remote_base) )
full_url = f"{webdav_url}/{remote_path}" remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
full_url = join_url(webdav_url, remote_path)
# Remove any double slashes (except in http://)
full_url = full_url.replace("://", "$PLACEHOLDER$") # Function to check if file exists in Nextcloud
while "//" in full_url: def check_exists_in_nextcloud(path):
full_url = full_url.replace("//", "/") check_url = join_url(webdav_url, os.path.dirname(path))
full_url = full_url.replace("$PLACEHOLDER$", "://") try:
response = requests.request(
# Function to check if file exists in Nextcloud "PROPFIND",
def check_exists_in_nextcloud(path): check_url,
check_url = f"{webdav_url}{os.path.dirname(path)}" auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
try: headers={"Depth": "1"},
response = requests.request( timeout=10,
"PROPFIND", )
check_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), return path in response.text
headers={"Depth": "1"}, except Exception:
timeout=10, # If we can't check, assume it doesn't exist
) return False
return path in response.text # Check for potential file collision and get a unique name if needed
except Exception: remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
# If we can't check, assume it doesn't exist full_url = join_url(webdav_url, remote_path)
return False
# Create necessary parent folders
# Check for potential file collision and get a unique name if needed parent_dirs = os.path.dirname(remote_path)
remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) if parent_dirs:
full_url = f"{webdav_url}/{remote_path}" current_path = ""
for folder in parent_dirs.split("/"):
# Fix double slashes again if not folder:
full_url = full_url.replace("://", "$PLACEHOLDER$") continue
while "//" in full_url: current_path += f"{folder}/"
full_url = full_url.replace("//", "/") mkdir_url = join_url(webdav_url, current_path)
full_url = full_url.replace("$PLACEHOLDER$", "://")
requests.request(
# Create necessary parent folders "MKCOL",
parent_dirs = os.path.dirname(remote_path) mkdir_url,
if parent_dirs: auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
current_path = "" timeout=10,
for folder in parent_dirs.split("/"): )
if not folder:
continue # Upload the file
current_path += f"{folder}/" logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
mkdir_url = f"{webdav_url}/{current_path}" log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
# Fix double slashes with open(file_path, "rb") as file_data:
mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$") response = requests.put(
while "//" in mkdir_url: full_url,
mkdir_url = mkdir_url.replace("//", "/") data=file_data,
mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://") auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
headers={"Content-Type": "application/octet-stream"},
requests.request( timeout=settings.http_request_timeout, # Use configured timeout for large files
"MKCOL", )
mkdir_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), if response.status_code in (201, 204): # Created or No Content
timeout=10, logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
) log_task_progress(
task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
# Upload the file )
logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") return {
log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) "status": "Completed",
with open(file_path, "rb") as file_data: "file_path": file_path,
response = requests.put( "nextcloud_path": remote_path,
full_url, "response_code": response.status_code,
data=file_data, }
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), else:
headers={"Content-Type": "application/octet-stream"}, error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
timeout=settings.http_request_timeout, # Use configured timeout for large files logger.error(f"[{task_id}] {error_msg}")
) log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
if response.status_code in (201, 204): # Created or No Content
logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}") except Exception as e:
log_task_progress( error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id logger.error(f"[{task_id}] {error_msg}")
) log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
return { raise Exception(error_msg)
"status": "Completed",
"file_path": file_path,
"nextcloud_path": remote_path,
"response_code": response.status_code,
}
else:
error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except Exception as e:
error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
app/utils/network.py
+26
View File
@@ -1,6 +1,7 @@
import ipaddress import ipaddress
import logging import logging
import socket import socket
from urllib.parse import urlsplit, urlunsplit
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -32,3 +33,28 @@ def is_private_ip(hostname: str) -> bool:
# and SSRF bypasses via unresolvable addresses. # and SSRF bypasses via unresolvable addresses.
logger.warning(f"Could not resolve hostname (blocking securely): {hostname}") logger.warning(f"Could not resolve hostname (blocking securely): {hostname}")
return True return True
def join_url(base: str, *parts: str) -> str:
"""
Safely join a base URL with one or more path parts.
Uses urllib.parse to correctly handle scheme/netloc/query/fragment so that
only the path component is modified. Leading and trailing slashes are
stripped from each part before joining, preventing double-slash sequences
at segment boundaries without touching the scheme separator or query string.
Examples:
join_url("https://example.com/dav/", "/remote/", "file.pdf")
-> "https://example.com/dav/remote/file.pdf"
"""
parsed = urlsplit(base)
# Strip each part once and filter out empty segments; use walrus operator
# to avoid calling strip twice per iteration.
stripped_parts = [s for p in parts if (s := p.strip("/"))]
base_path = parsed.path.rstrip("/")
new_path = base_path + "/" + "/".join(stripped_parts) if stripped_parts else base_path
# Ensure path is non-empty so the reconstructed URL is valid.
if not new_path:
new_path = "/"
return urlunsplit((parsed.scheme, parsed.netloc, new_path, parsed.query, parsed.fragment))
tests/test_upload_to_nextcloud_join_url.py
+52
View File
@@ -0,0 +1,52 @@
from unittest.mock import MagicMock, patch
import pytest
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
@pytest.fixture
def mock_settings(tmp_path):
with patch("app.tasks.upload_to_nextcloud.settings") as mock:
mock.nextcloud_upload_url = "http://nextcloud.local/"
mock.nextcloud_username = "testuser"
mock.nextcloud_password = "testpassword"
mock.nextcloud_folder = "uploads"
mock.workdir = str(tmp_path)
mock.http_request_timeout = 30
yield mock
@pytest.fixture
def mock_requests():
with patch("app.tasks.upload_to_nextcloud.requests") as mock:
# Mock PROPFIND to always return false (file doesn't exist)
mock.request.return_value = MagicMock(text="<response></response>")
# Mock PUT to return success
put_response = MagicMock()
put_response.status_code = 201
mock.put.return_value = put_response
yield mock
def test_upload_to_nextcloud_url_construction(tmp_path, mock_settings, mock_requests):
file_path = str(tmp_path / "test_file.txt")
# Create dummy file
with open(file_path, "w") as f:
f.write("test content")
# Call the task directly
with patch("celery.app.task.Task.request", new_callable=MagicMock) as mock_req:
mock_req.id = "test-task-123"
result = upload_to_nextcloud(file_path)
assert result["status"] == "Completed"
assert result["nextcloud_path"] == "uploads/test_file.txt"
# Verify requests.put was called with the correct URL
mock_requests.put.assert_called_once()
args, kwargs = mock_requests.put.call_args
url = args[0]
assert url == "http://nextcloud.local/uploads/test_file.txt"