diff --git a/CHANGELOG.md b/CHANGELOG.md index 1785b55b..36ba3623 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -130,6 +130,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## 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) ### Bug Fixes diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 1bb29c75..b8dc52ba 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -1,157 +1,141 @@ -#!/usr/bin/env python3 - -import logging -import os - -import requests -from requests.auth import HTTPBasicAuth - -from app.celery_app import celery -from app.config import settings -from app.tasks.retry_config import UploadTaskWithRetry -from app.utils import log_task_progress -from app.utils.filename_utils import extract_remote_path, get_unique_filename - -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): - """ - Upload a file to Nextcloud WebDAV. - - Args: - 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}") - log_task_progress( - task_id, - "upload_to_nextcloud", - "in_progress", - 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}" - logger.error(f"[{task_id}] {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 - if not ( - getattr(settings, "nextcloud_upload_url", 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) - return {"status": "Skipped", "reason": "Nextcloud settings not configured"} - - filename = os.path.basename(file_path) - - try: - # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url - webdav_url = settings.nextcloud_upload_url - if not webdav_url.endswith("/"): - webdav_url += "/" - - # Calculate remote path based on local file structure - 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}" - - # Remove any double slashes (except in http://) - full_url = full_url.replace("://", "$PLACEHOLDER$") - while "//" in full_url: - full_url = full_url.replace("//", "/") - full_url = full_url.replace("$PLACEHOLDER$", "://") - - # Function to check if file exists in Nextcloud - def check_exists_in_nextcloud(path): - check_url = f"{webdav_url}{os.path.dirname(path)}" - try: - response = requests.request( - "PROPFIND", - check_url, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - headers={"Depth": "1"}, - timeout=10, - ) - - return path in response.text - except Exception: - # If we can't check, assume it doesn't exist - return False - - # Check for potential file collision and get a unique name if needed - remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) - full_url = f"{webdav_url}/{remote_path}" - - # Fix double slashes again - full_url = full_url.replace("://", "$PLACEHOLDER$") - while "//" in full_url: - full_url = full_url.replace("//", "/") - full_url = full_url.replace("$PLACEHOLDER$", "://") - - # Create necessary parent folders - parent_dirs = os.path.dirname(remote_path) - if parent_dirs: - current_path = "" - for folder in parent_dirs.split("/"): - if not folder: - continue - current_path += f"{folder}/" - mkdir_url = f"{webdav_url}/{current_path}" - # Fix double slashes - mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$") - while "//" in mkdir_url: - mkdir_url = mkdir_url.replace("//", "/") - mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://") - - requests.request( - "MKCOL", - mkdir_url, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - timeout=10, - ) - - # Upload the file - logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") - log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) - with open(file_path, "rb") as file_data: - response = requests.put( - full_url, - data=file_data, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - headers={"Content-Type": "application/octet-stream"}, - timeout=settings.http_request_timeout, # Use configured timeout for large files - ) - - if response.status_code in (201, 204): # Created or No Content - 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 - ) - return { - "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) +#!/usr/bin/env python3 + +import logging +import os + +import requests +from requests.auth import HTTPBasicAuth + +from app.celery_app import celery +from app.config import settings +from app.tasks.retry_config import UploadTaskWithRetry +from app.utils import log_task_progress +from app.utils.filename_utils import extract_remote_path, get_unique_filename +from app.utils.network import join_url + +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): + """ + Upload a file to Nextcloud WebDAV. + + Args: + 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}") + log_task_progress( + task_id, + "upload_to_nextcloud", + "in_progress", + 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}" + logger.error(f"[{task_id}] {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 + if not ( + getattr(settings, "nextcloud_upload_url", 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) + return {"status": "Skipped", "reason": "Nextcloud settings not configured"} + + filename = os.path.basename(file_path) + + try: + # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url + webdav_url = settings.nextcloud_upload_url + if not webdav_url.endswith("/"): + webdav_url += "/" + + # Calculate remote path based on local file structure + 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 = join_url(webdav_url, remote_path) + + # Function to check if file exists in Nextcloud + def check_exists_in_nextcloud(path): + check_url = join_url(webdav_url, os.path.dirname(path)) + try: + response = requests.request( + "PROPFIND", + check_url, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + headers={"Depth": "1"}, + timeout=10, + ) + + return path in response.text + except Exception: + # If we can't check, assume it doesn't exist + return False + + # Check for potential file collision and get a unique name if needed + remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) + full_url = join_url(webdav_url, remote_path) + + # Create necessary parent folders + parent_dirs = os.path.dirname(remote_path) + if parent_dirs: + current_path = "" + for folder in parent_dirs.split("/"): + if not folder: + continue + current_path += f"{folder}/" + mkdir_url = join_url(webdav_url, current_path) + + requests.request( + "MKCOL", + mkdir_url, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + timeout=10, + ) + + # Upload the file + logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) + with open(file_path, "rb") as file_data: + response = requests.put( + full_url, + data=file_data, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + headers={"Content-Type": "application/octet-stream"}, + timeout=settings.http_request_timeout, # Use configured timeout for large files + ) + + if response.status_code in (201, 204): # Created or No Content + 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 + ) + return { + "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) diff --git a/app/utils/network.py b/app/utils/network.py index ef775ba0..1c63111a 100644 --- a/app/utils/network.py +++ b/app/utils/network.py @@ -1,6 +1,7 @@ import ipaddress import logging import socket +from urllib.parse import urlsplit, urlunsplit logger = logging.getLogger(__name__) @@ -32,3 +33,28 @@ def is_private_ip(hostname: str) -> bool: # and SSRF bypasses via unresolvable addresses. logger.warning(f"Could not resolve hostname (blocking securely): {hostname}") 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)) diff --git a/tests/test_upload_to_nextcloud_join_url.py b/tests/test_upload_to_nextcloud_join_url.py new file mode 100644 index 00000000..705af7c5 --- /dev/null +++ b/tests/test_upload_to_nextcloud_join_url.py @@ -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="") + + # 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"