diff --git a/.jules/sentinel.md b/.jules/sentinel.md index fa35a57f..4ad0579f 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -1,8 +1,4 @@ -## 2024-05-24 - SSRF in WebDAV connection test -**Vulnerability:** The `_test_webdav_connection` function had a custom SSRF check that failed to resolve DNS names, allowing attackers to bypass the check by providing a domain that resolves to an internal IP (e.g., `127.0.0.1`). -**Learning:** DNS resolution is required for robust SSRF protection when validating URLs provided by users. -**Prevention:** Use a centralized `is_private_ip` function (now in `app/utils/network.py`) that resolves the hostname to its IPs and checks if any are private. -## 2026-03-22 - B310: urllib.request.urlopen replaced with httpx -**Vulnerability:** The `_test_webdav_connection` function used `urllib.request.urlopen`, which natively supports dangerous schemes like `file://` or `ftp://` and follows redirects by default, potentially allowing SSRF bypasses or Local File Inclusion. -**Learning:** `urllib.request` should be avoided for user-supplied URLs. Even when URL schemes are manually validated, `urllib`'s default redirect following behavior can bypass SSRF protections (e.g. redirecting to `127.0.0.1`). -**Prevention:** Use a modern, safer HTTP client like `httpx` with `follow_redirects=False` when testing user-provided URLs. +## 2025-05-18 - [SSRF Bypass via DNS Resolution Failure] +**Vulnerability:** The `is_private_ip` function in `app/utils/network.py` failed open (returned `False`) when a hostname could not be resolved (`socket.gaierror`). +**Learning:** This fail-open pattern was originally added to allow external domains in tests, but in production, it created a severe SSRF risk. An attacker could bypass SSRF protections by providing a URL that fails to resolve during the security check but resolves later (DNS rebinding), or by exploiting internal routing behaviors via unresolvable addresses. +**Prevention:** Always fail securely in network authorization functions. If a domain cannot be resolved to verify its safety, the request must be blocked (`return True` / default-deny). Tests should mock DNS resolution correctly instead of compromising production security logic. \ No newline at end of file diff --git a/BUILD_DATE b/BUILD_DATE index 9adfb980..7c7cd695 100644 --- a/BUILD_DATE +++ b/BUILD_DATE @@ -1 +1 @@ -2026-03-22T18:47:07Z +2026-03-23T14:11:22Z diff --git a/CHANGELOG.md b/CHANGELOG.md index becc0061..c441ae6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,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/Dockerfile b/Dockerfile index 3fd59488..82cefb60 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,7 +34,7 @@ WORKDIR /frontend # Install dependencies first (layer-cached unless package.json/lockfile changes) COPY frontend/package.json frontend/package-lock.json ./ -RUN npm ci --omit=dev +RUN npm ci # Copy source files and compile Tailwind CSS COPY frontend/ ./ diff --git a/GIT_SHA b/GIT_SHA index 49106340..a66d3a39 100644 --- a/GIT_SHA +++ b/GIT_SHA @@ -1 +1 @@ -76c0e91 +34457f9 diff --git a/RUNTIME_INFO b/RUNTIME_INFO index b562196b..3b67ecaa 100644 --- a/RUNTIME_INFO +++ b/RUNTIME_INFO @@ -1,10 +1,10 @@ DocuElevate Build Information ============================== -Version: 0.172.1 -Build Date: 2026-03-22T18:47:07Z -Git Commit: 76c0e91500963fac4e8d4a43123340a7cc64731f -Git Short SHA: 76c0e91 +Version: 0.172.2 +Build Date: 2026-03-23T14:11:22Z +Git Commit: 34457f977509ce145b7411e83982a96b0fd0e33e +Git Short SHA: 34457f9 Git Branch: main -Commit Date: 2026-03-22T19:46:48+01:00 -Build Timestamp: 2026-03-22T18:47:07Z +Commit Date: 2026-03-23T15:10:59+01:00 +Build Timestamp: 2026-03-23T14:11:22Z ============================== diff --git a/VERSION b/VERSION index f6bbd8d3..3c99c40a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.172.1 +0.172.2 diff --git a/app/api/billing.py b/app/api/billing.py index 9c5582b0..85528608 100644 --- a/app/api/billing.py +++ b/app/api/billing.py @@ -260,7 +260,7 @@ async def stripe_webhook(request: Request, db: Session = Depends(get_db)) -> dic @require_login async def billing_success(request: Request) -> Any: """Show a success page after a completed Stripe Checkout.""" - return _templates.TemplateResponse("billing_success.html", {"request": request}) + return _templates.TemplateResponse(request, "billing_success.html") # --------------------------------------------------------------------------- diff --git a/app/api/local_auth.py b/app/api/local_auth.py index 2b68003f..e946f484 100644 --- a/app/api/local_auth.py +++ b/app/api/local_auth.py @@ -101,9 +101,9 @@ async def signup_page(request: Request) -> Any: if not settings.allow_local_signup: return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302) return templates.TemplateResponse( + request, "signup.html", - { - "request": request, + context={ "csrf_token": getattr(request.state, "csrf_token", ""), "app_version": settings.version, }, @@ -113,16 +113,16 @@ async def signup_page(request: Request) -> Any: @router.get("/verify-email-sent", include_in_schema=False) async def verify_email_sent_page(request: Request) -> Any: """Render the verify-email-sent confirmation page.""" - return templates.TemplateResponse("verify_email_sent.html", {"request": request}) + return templates.TemplateResponse(request, "verify_email_sent.html") @router.get("/forgot-username", include_in_schema=False) async def forgot_username_page(request: Request) -> Any: """Render the forgot-username page where users can request a username reminder email.""" return templates.TemplateResponse( + request, "forgot_username.html", - { - "request": request, + context={ "csrf_token": getattr(request.state, "csrf_token", ""), "app_version": settings.version, }, @@ -133,9 +133,9 @@ async def forgot_username_page(request: Request) -> Any: async def forgot_password_page(request: Request) -> Any: """Render the forgot-password page where users can request a reset email.""" return templates.TemplateResponse( + request, "forgot_password.html", - { - "request": request, + context={ "csrf_token": getattr(request.state, "csrf_token", ""), "app_version": settings.version, }, @@ -147,9 +147,9 @@ async def reset_password_page(request: Request) -> Any: """Render the password reset form page.""" token = request.query_params.get("token", "") return templates.TemplateResponse( + request, "password_reset_form.html", - { - "request": request, + context={ "token": token, "csrf_token": getattr(request.state, "csrf_token", ""), "app_version": settings.version, diff --git a/app/auth.py b/app/auth.py index 613e6948..14756c47 100644 --- a/app/auth.py +++ b/app/auth.py @@ -536,9 +536,9 @@ async def login(request: Request): return RedirectResponse(url="/oauth-login", status_code=status.HTTP_302_FOUND) return templates.TemplateResponse( + request, "login.html", - { - "request": request, + context={ "error": error, "message": message, "show_oauth": show_oauth, diff --git a/app/main.py b/app/main.py index 89477ceb..b11b3148 100644 --- a/app/main.py +++ b/app/main.py @@ -424,15 +424,13 @@ async def http_exception_handler(request: Request, exc: HTTPException): # For frontend routes, return appropriate HTML templates # Handle 404 errors with a custom template if exc.status_code == 404: - return _error_templates.TemplateResponse( - "404.html", {"request": request}, status_code=status.HTTP_404_NOT_FOUND - ) + return _error_templates.TemplateResponse(request, "404.html", status_code=status.HTTP_404_NOT_FOUND) # For other HTTP errors, we could create specific templates or use a generic one # For now, return a simple error page return _error_templates.TemplateResponse( + request, "404.html", # Reuse 404 template for other errors, or create a generic error template - {"request": request}, status_code=exc.status_code, ) @@ -452,8 +450,9 @@ async def custom_500_handler(request: Request, exc: Exception): # Serve the 500 template for non-API routes return _error_templates.TemplateResponse( + request, "500.html", - {"request": request, "exc": exc}, + context={"exc": exc}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) 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 7ea6d88b..2bc87c9b 100644 --- a/app/utils/network.py +++ b/app/utils/network.py @@ -27,8 +27,21 @@ def is_private_ip(hostname: str) -> bool: return True return False except (socket.gaierror, socket.error): - # Cannot resolve - allow for testing/development - # In production, DNS should work properly - # Log this for debugging - logger.warning(f"Could not resolve hostname: {hostname}") - return False # Changed from True to False to allow external domains in tests + # Cannot resolve. + # Fail securely: block unresolved domains to prevent DNS rebinding + # 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 and multiple path parts. + Handles double slashes while preserving the protocol '://'. + """ + url = "/".join([base, *parts]) + url = url.replace("://", "$PLACEHOLDER$") + while "//" in url: + url = url.replace("//", "/") + url = url.replace("$PLACEHOLDER$", "://") + return url diff --git a/app/views/base.py b/app/views/base.py index 6a116733..f3ffa157 100644 --- a/app/views/base.py +++ b/app/views/base.py @@ -162,12 +162,36 @@ def _inject_global_context(ctx: dict) -> None: def template_response_with_version(*args, **kwargs): - """Wrapper for TemplateResponse to include version and CSRF token in all templates""" - # If context dict is provided, add version to it - if len(args) >= 2 and isinstance(args[1], dict): - _inject_global_context(args[1]) - elif "context" in kwargs and isinstance(kwargs["context"], dict): + """Wrapper for TemplateResponse to include version and CSRF token in all templates. + + Handles both old-style and new-style Starlette TemplateResponse calls: + - Old-style (Starlette <1.0): TemplateResponse(name, {"request": req, ...}, ...) + - New-style (Starlette 1.0+): TemplateResponse(request, name, context={...}, ...) + """ + if len(args) >= 1 and isinstance(args[0], str): + # Old-style call: first positional arg is the template name (string). + # Convert to new-style: (request, name, context=..., ...) + name = args[0] + if len(args) >= 2 and isinstance(args[1], dict): + context = args[1] + # Old-style may have status_code as 3rd positional arg + if len(args) >= 3 and "status_code" not in kwargs: + kwargs["status_code"] = args[2] + else: + context = kwargs.pop("context", {}) + request_obj = context.pop("request", None) + if request_obj is not None: + context["request"] = request_obj + _inject_global_context(context) + if request_obj is not None: + return original_template_response(request_obj, name, context=context, **kwargs) + return original_template_response(name, context=context, **kwargs) + + # New-style call: (request, name, context=..., ...) + if "context" in kwargs and isinstance(kwargs["context"], dict): _inject_global_context(kwargs["context"]) + elif len(args) >= 3 and isinstance(args[2], dict): + _inject_global_context(args[2]) return original_template_response(*args, **kwargs) diff --git a/app/views/share.py b/app/views/share.py index 118e0ee9..925342f6 100644 --- a/app/views/share.py +++ b/app/views/share.py @@ -23,6 +23,7 @@ templates = Jinja2Templates(directory=str(_templates_dir)) async def shared_link_view(request: Request, token: str): """Render the public share landing page for a given token.""" return templates.TemplateResponse( + request, "shared_link_view.html", - {"request": request, "token": token}, + context={"token": token}, ) diff --git a/tests/test_auth.py b/tests/test_auth.py index 85e047e8..d3af75ca 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -430,8 +430,8 @@ class TestLoginFunction: # Verify TemplateResponse was called with correct context mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args - assert call_args[0][0] == "login.html" - context = call_args[0][1] + assert call_args[0][1] == "login.html" + context = call_args.kwargs["context"] assert context["error"] == "Test error" assert context["message"] == "Test message" @@ -450,7 +450,7 @@ class TestLoginFunction: mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args - context = call_args[0][1] + context = call_args.kwargs["context"] assert context["error"] is None assert context["message"] is None diff --git a/tests/test_auth_module.py b/tests/test_auth_module.py index 233dfd5e..ba41b756 100644 --- a/tests/test_auth_module.py +++ b/tests/test_auth_module.py @@ -281,7 +281,7 @@ class TestLoginEndpoint: # Verify template was rendered with OAuth enabled mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args - context = call_args[0][1] + context = call_args.kwargs["context"] assert context["show_oauth"] is True assert context["oauth_provider_name"] == "Test SSO" diff --git a/tests/test_coverage_polish.py b/tests/test_coverage_polish.py index c5b7c4de..1f196ba3 100644 --- a/tests/test_coverage_polish.py +++ b/tests/test_coverage_polish.py @@ -520,14 +520,14 @@ class TestURLUploadAdditionalCoverage: assert exc_info.value.status_code == 400 def test_is_private_ip_unresolvable_hostname(self): - """Cover DNS resolution failure branch (lines 67-72).""" + """Cover DNS resolution failure branch blocking unresolvable domains.""" import socket as _socket from app.utils.network import is_private_ip with patch("socket.getaddrinfo", side_effect=_socket.gaierror("nope")): result = is_private_ip("nonexistent.invalid.hostname.test") - assert result is False + assert result is True # Fail securely by returning True def test_is_private_ip_hostname_resolves_to_private(self): """Cover branch where hostname resolves to a private IP (line 64-65).""" diff --git a/tests/test_coverage_remaining_gaps.py b/tests/test_coverage_remaining_gaps.py index b4bf96ec..e68fdb93 100644 --- a/tests/test_coverage_remaining_gaps.py +++ b/tests/test_coverage_remaining_gaps.py @@ -69,8 +69,9 @@ class TestViewsBase: context = {"request": req} template_response_with_version("template.html", context) - args, _ = mock_orig.call_args - assert args[1].get("csrf_token") == "my-csrf" + args, kwargs = mock_orig.call_args + context = kwargs.get("context", {}) + assert context.get("csrf_token") == "my-csrf" def test_kwargs_context_no_request(self): """Test kwargs context path when request is not in context.""" diff --git a/tests/test_dark_mode.py b/tests/test_dark_mode.py index 82514a20..a39752c7 100644 --- a/tests/test_dark_mode.py +++ b/tests/test_dark_mode.py @@ -55,8 +55,8 @@ class TestDarkModeTemplateInjection: captured = {} - def fake_original(name, ctx, **kw): - captured.update(ctx) + def fake_original(request_obj, name, context=None, **kw): + captured.update(context or {}) with patch("app.views.base.original_template_response", side_effect=fake_original): mock_request = MagicMock() @@ -73,8 +73,8 @@ class TestDarkModeTemplateInjection: captured = {} - def fake_original(name, ctx, **kw): - captured.update(ctx) + def fake_original(request_obj, name, context=None, **kw): + captured.update(context or {}) with patch("app.views.base.original_template_response", side_effect=fake_original): mock_request = MagicMock() diff --git a/tests/test_frontend_build.py b/tests/test_frontend_build.py new file mode 100644 index 00000000..528bbfff --- /dev/null +++ b/tests/test_frontend_build.py @@ -0,0 +1,145 @@ +"""Tests for frontend build configuration and Docker build consistency. + +Validates that the frontend build toolchain (Tailwind CSS) is correctly +configured in package.json and that the Dockerfile installs all required +dependencies for the build step. +""" + +import json +import re +from pathlib import Path + +import pytest + +# Resolve the project root from the test file location +PROJECT_ROOT = Path(__file__).resolve().parent.parent +FRONTEND_DIR = PROJECT_ROOT / "frontend" +DOCKERFILE_PATH = PROJECT_ROOT / "Dockerfile" + + +@pytest.mark.unit +class TestFrontendPackageJson: + """Validate frontend/package.json structure and scripts.""" + + def test_package_json_exists(self) -> None: + """package.json must exist in the frontend directory.""" + pkg_path = FRONTEND_DIR / "package.json" + assert pkg_path.exists(), "frontend/package.json not found" + + def test_package_json_is_valid_json(self) -> None: + """package.json must be parseable JSON.""" + pkg_path = FRONTEND_DIR / "package.json" + data = json.loads(pkg_path.read_text(encoding="utf-8")) + assert isinstance(data, dict), "package.json must be a JSON object" + + def test_build_script_defined(self) -> None: + """A 'build' script must be defined in package.json.""" + pkg_path = FRONTEND_DIR / "package.json" + data = json.loads(pkg_path.read_text(encoding="utf-8")) + scripts = data.get("scripts", {}) + assert "build" in scripts, "Missing 'build' script in package.json" + + def test_build_script_uses_tailwindcss(self) -> None: + """The build script must invoke the tailwindcss CLI.""" + pkg_path = FRONTEND_DIR / "package.json" + data = json.loads(pkg_path.read_text(encoding="utf-8")) + build_cmd = data["scripts"]["build"] + assert "tailwindcss" in build_cmd, f"Build script does not reference tailwindcss: {build_cmd}" + + def test_tailwindcss_listed_as_dependency(self) -> None: + """tailwindcss must be listed in dependencies or devDependencies.""" + pkg_path = FRONTEND_DIR / "package.json" + data = json.loads(pkg_path.read_text(encoding="utf-8")) + deps = data.get("dependencies", {}) + dev_deps = data.get("devDependencies", {}) + all_deps = {**deps, **dev_deps} + assert "tailwindcss" in all_deps, "tailwindcss is not listed in dependencies or devDependencies" + + +@pytest.mark.unit +class TestFrontendBuildAssets: + """Validate that required frontend build source files exist.""" + + def test_input_css_exists(self) -> None: + """The Tailwind CSS input file must exist.""" + input_css = FRONTEND_DIR / "input.css" + assert input_css.exists(), "frontend/input.css not found" + + def test_input_css_has_tailwind_directives(self) -> None: + """input.css must include Tailwind CSS directives.""" + input_css = FRONTEND_DIR / "input.css" + content = input_css.read_text(encoding="utf-8") + assert "@tailwind base" in content, "Missing @tailwind base directive" + assert "@tailwind components" in content, "Missing @tailwind components directive" + assert "@tailwind utilities" in content, "Missing @tailwind utilities directive" + + def test_tailwind_config_exists(self) -> None: + """tailwind.config.js must exist in the frontend directory.""" + config_path = FRONTEND_DIR / "tailwind.config.js" + assert config_path.exists(), "frontend/tailwind.config.js not found" + + def test_package_lock_exists(self) -> None: + """package-lock.json must exist for reproducible installs.""" + lock_path = FRONTEND_DIR / "package-lock.json" + assert lock_path.exists(), "frontend/package-lock.json not found" + + +@pytest.mark.unit +class TestDockerfileFrontendBuilder: + """Validate the Dockerfile frontend-builder stage installs build dependencies.""" + + def test_dockerfile_exists(self) -> None: + """Production Dockerfile must exist at the project root.""" + assert DOCKERFILE_PATH.exists(), "Dockerfile not found at project root" + + def test_dockerfile_has_frontend_builder_stage(self) -> None: + """Dockerfile must define a frontend-builder stage.""" + content = DOCKERFILE_PATH.read_text(encoding="utf-8") + assert "AS frontend-builder" in content, "Dockerfile does not define a frontend-builder stage" + + def test_dockerfile_npm_ci_does_not_omit_dev(self) -> None: + """npm ci must NOT use --omit=dev in the frontend-builder stage. + + The tailwindcss CLI is a devDependency required at build time. + Using --omit=dev would skip installing it, causing the build to + fail with 'tailwindcss: not found'. + """ + content = DOCKERFILE_PATH.read_text(encoding="utf-8") + + # Extract the frontend-builder stage content + # Look for the stage start and the next stage (or end of file) + stage_pattern = re.compile( + r"FROM\s+\S+\s+AS\s+frontend-builder\b(.*?)(?=FROM\s|\Z)", + re.DOTALL, + ) + match = stage_pattern.search(content) + assert match is not None, "Could not find frontend-builder stage in Dockerfile" + + stage_content = match.group(1) + assert "--omit=dev" not in stage_content, ( + "Dockerfile frontend-builder stage uses 'npm ci --omit=dev' which " + "excludes tailwindcss (a devDependency) needed for the build step. " + "Use 'npm ci' instead to install all dependencies." + ) + + def test_dockerfile_runs_npm_build(self) -> None: + """Dockerfile frontend-builder stage must run npm run build.""" + content = DOCKERFILE_PATH.read_text(encoding="utf-8") + + stage_pattern = re.compile( + r"FROM\s+\S+\s+AS\s+frontend-builder\b(.*?)(?=FROM\s|\Z)", + re.DOTALL, + ) + match = stage_pattern.search(content) + assert match is not None, "Could not find frontend-builder stage in Dockerfile" + + stage_content = match.group(1) + assert "npm run build" in stage_content, "Dockerfile frontend-builder stage does not run 'npm run build'" + + def test_dockerfile_copies_compiled_css(self) -> None: + """Dockerfile must copy the compiled styles.css from the frontend-builder stage.""" + content = DOCKERFILE_PATH.read_text(encoding="utf-8") + assert "COPY --from=frontend-builder" in content, ( + "Dockerfile does not copy assets from the frontend-builder stage" + ) + assert "styles.css" in content, "Dockerfile does not reference the compiled styles.css" diff --git a/tests/test_social_login.py b/tests/test_social_login.py index 54c81931..b4b330bc 100644 --- a/tests/test_social_login.py +++ b/tests/test_social_login.py @@ -345,7 +345,7 @@ class TestLoginPageSocialProviders: mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args - context = call_args[0][1] + context = call_args.kwargs.get("context", {}) assert context["social_providers"] == mock_providers @pytest.mark.asyncio @@ -371,7 +371,7 @@ class TestLoginPageSocialProviders: mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args - context = call_args[0][1] + context = call_args.kwargs.get("context", {}) assert context["social_providers"] == {} 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..de14ed60 --- /dev/null +++ b/tests/test_upload_to_nextcloud_join_url.py @@ -0,0 +1,54 @@ +import os +from unittest.mock import MagicMock, patch + +import pytest + +from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + +@pytest.fixture +def mock_settings(): + 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 = "/tmp/workdir" + 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(mock_settings, mock_requests): + file_path = "/tmp/workdir/test_file.txt" + + # Create dummy file + os.makedirs("/tmp/workdir", exist_ok=True) + 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" diff --git a/tests/test_url_upload.py b/tests/test_url_upload.py index 7fead0ae..5dff00ac 100644 --- a/tests/test_url_upload.py +++ b/tests/test_url_upload.py @@ -698,6 +698,18 @@ class TestURLUploadCoverageGaps: assert result is False mock_getaddrinfo.assert_called_once() + @patch("app.utils.network.socket.getaddrinfo") + def test_is_private_ip_unresolvable_hostname_fails_securely(self, mock_getaddrinfo): + """Test that unresolvable hostnames fail securely by blocking access.""" + import socket + + from app.utils.network import is_private_ip + + mock_getaddrinfo.side_effect = socket.gaierror("Name or service not known") + + result = is_private_ip("unresolvable.example.internal") + assert result is True # Fails securely + @patch("socket.getaddrinfo") def test_is_private_ip_hostname_resolves_multiple_ips_all_public(self, mock_getaddrinfo): """Test hostname with multiple public IPs returns False (covers 65->61 loop branch)"""