Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1fd3e745cf |
+4
-8
@@ -1,8 +1,4 @@
|
|||||||
## 2024-05-24 - SSRF in WebDAV connection test
|
## 2025-05-18 - [SSRF Bypass via DNS Resolution Failure]
|
||||||
**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`).
|
**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:** DNS resolution is required for robust SSRF protection when validating URLs provided by users.
|
**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:** 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.
|
**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.
|
||||||
## 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.
|
|
||||||
+1
-1
@@ -1 +1 @@
|
|||||||
2026-03-22T18:47:07Z
|
2026-03-23T14:11:22Z
|
||||||
|
|||||||
@@ -13,6 +13,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
|
||||||
|
|||||||
+1
-1
@@ -34,7 +34,7 @@ WORKDIR /frontend
|
|||||||
|
|
||||||
# Install dependencies first (layer-cached unless package.json/lockfile changes)
|
# Install dependencies first (layer-cached unless package.json/lockfile changes)
|
||||||
COPY frontend/package.json frontend/package-lock.json ./
|
COPY frontend/package.json frontend/package-lock.json ./
|
||||||
RUN npm ci --omit=dev
|
RUN npm ci
|
||||||
|
|
||||||
# Copy source files and compile Tailwind CSS
|
# Copy source files and compile Tailwind CSS
|
||||||
COPY frontend/ ./
|
COPY frontend/ ./
|
||||||
|
|||||||
+6
-6
@@ -1,10 +1,10 @@
|
|||||||
DocuElevate Build Information
|
DocuElevate Build Information
|
||||||
==============================
|
==============================
|
||||||
Version: 0.172.1
|
Version: 0.172.2
|
||||||
Build Date: 2026-03-22T18:47:07Z
|
Build Date: 2026-03-23T14:11:22Z
|
||||||
Git Commit: 76c0e91500963fac4e8d4a43123340a7cc64731f
|
Git Commit: 34457f977509ce145b7411e83982a96b0fd0e33e
|
||||||
Git Short SHA: 76c0e91
|
Git Short SHA: 34457f9
|
||||||
Git Branch: main
|
Git Branch: main
|
||||||
Commit Date: 2026-03-22T19:46:48+01:00
|
Commit Date: 2026-03-23T15:10:59+01:00
|
||||||
Build Timestamp: 2026-03-22T18:47:07Z
|
Build Timestamp: 2026-03-23T14:11:22Z
|
||||||
==============================
|
==============================
|
||||||
|
|||||||
+1
-1
@@ -260,7 +260,7 @@ async def stripe_webhook(request: Request, db: Session = Depends(get_db)) -> dic
|
|||||||
@require_login
|
@require_login
|
||||||
async def billing_success(request: Request) -> Any:
|
async def billing_success(request: Request) -> Any:
|
||||||
"""Show a success page after a completed Stripe Checkout."""
|
"""Show a success page after a completed Stripe Checkout."""
|
||||||
return _templates.TemplateResponse("billing_success.html", {"request": request})
|
return _templates.TemplateResponse(request, "billing_success.html")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -101,9 +101,9 @@ async def signup_page(request: Request) -> Any:
|
|||||||
if not settings.allow_local_signup:
|
if not settings.allow_local_signup:
|
||||||
return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302)
|
return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"signup.html",
|
"signup.html",
|
||||||
{
|
context={
|
||||||
"request": request,
|
|
||||||
"csrf_token": getattr(request.state, "csrf_token", ""),
|
"csrf_token": getattr(request.state, "csrf_token", ""),
|
||||||
"app_version": settings.version,
|
"app_version": settings.version,
|
||||||
},
|
},
|
||||||
@@ -113,16 +113,16 @@ async def signup_page(request: Request) -> Any:
|
|||||||
@router.get("/verify-email-sent", include_in_schema=False)
|
@router.get("/verify-email-sent", include_in_schema=False)
|
||||||
async def verify_email_sent_page(request: Request) -> Any:
|
async def verify_email_sent_page(request: Request) -> Any:
|
||||||
"""Render the verify-email-sent confirmation page."""
|
"""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)
|
@router.get("/forgot-username", include_in_schema=False)
|
||||||
async def forgot_username_page(request: Request) -> Any:
|
async def forgot_username_page(request: Request) -> Any:
|
||||||
"""Render the forgot-username page where users can request a username reminder email."""
|
"""Render the forgot-username page where users can request a username reminder email."""
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"forgot_username.html",
|
"forgot_username.html",
|
||||||
{
|
context={
|
||||||
"request": request,
|
|
||||||
"csrf_token": getattr(request.state, "csrf_token", ""),
|
"csrf_token": getattr(request.state, "csrf_token", ""),
|
||||||
"app_version": settings.version,
|
"app_version": settings.version,
|
||||||
},
|
},
|
||||||
@@ -133,9 +133,9 @@ async def forgot_username_page(request: Request) -> Any:
|
|||||||
async def forgot_password_page(request: Request) -> Any:
|
async def forgot_password_page(request: Request) -> Any:
|
||||||
"""Render the forgot-password page where users can request a reset email."""
|
"""Render the forgot-password page where users can request a reset email."""
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"forgot_password.html",
|
"forgot_password.html",
|
||||||
{
|
context={
|
||||||
"request": request,
|
|
||||||
"csrf_token": getattr(request.state, "csrf_token", ""),
|
"csrf_token": getattr(request.state, "csrf_token", ""),
|
||||||
"app_version": settings.version,
|
"app_version": settings.version,
|
||||||
},
|
},
|
||||||
@@ -147,9 +147,9 @@ async def reset_password_page(request: Request) -> Any:
|
|||||||
"""Render the password reset form page."""
|
"""Render the password reset form page."""
|
||||||
token = request.query_params.get("token", "")
|
token = request.query_params.get("token", "")
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"password_reset_form.html",
|
"password_reset_form.html",
|
||||||
{
|
context={
|
||||||
"request": request,
|
|
||||||
"token": token,
|
"token": token,
|
||||||
"csrf_token": getattr(request.state, "csrf_token", ""),
|
"csrf_token": getattr(request.state, "csrf_token", ""),
|
||||||
"app_version": settings.version,
|
"app_version": settings.version,
|
||||||
|
|||||||
+2
-2
@@ -536,9 +536,9 @@ async def login(request: Request):
|
|||||||
return RedirectResponse(url="/oauth-login", status_code=status.HTTP_302_FOUND)
|
return RedirectResponse(url="/oauth-login", status_code=status.HTTP_302_FOUND)
|
||||||
|
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"login.html",
|
"login.html",
|
||||||
{
|
context={
|
||||||
"request": request,
|
|
||||||
"error": error,
|
"error": error,
|
||||||
"message": message,
|
"message": message,
|
||||||
"show_oauth": show_oauth,
|
"show_oauth": show_oauth,
|
||||||
|
|||||||
+4
-5
@@ -424,15 +424,13 @@ async def http_exception_handler(request: Request, exc: HTTPException):
|
|||||||
# For frontend routes, return appropriate HTML templates
|
# For frontend routes, return appropriate HTML templates
|
||||||
# Handle 404 errors with a custom template
|
# Handle 404 errors with a custom template
|
||||||
if exc.status_code == 404:
|
if exc.status_code == 404:
|
||||||
return _error_templates.TemplateResponse(
|
return _error_templates.TemplateResponse(request, "404.html", status_code=status.HTTP_404_NOT_FOUND)
|
||||||
"404.html", {"request": request}, status_code=status.HTTP_404_NOT_FOUND
|
|
||||||
)
|
|
||||||
|
|
||||||
# For other HTTP errors, we could create specific templates or use a generic one
|
# For other HTTP errors, we could create specific templates or use a generic one
|
||||||
# For now, return a simple error page
|
# For now, return a simple error page
|
||||||
return _error_templates.TemplateResponse(
|
return _error_templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"404.html", # Reuse 404 template for other errors, or create a generic error template
|
"404.html", # Reuse 404 template for other errors, or create a generic error template
|
||||||
{"request": request},
|
|
||||||
status_code=exc.status_code,
|
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
|
# Serve the 500 template for non-API routes
|
||||||
return _error_templates.TemplateResponse(
|
return _error_templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"500.html",
|
"500.html",
|
||||||
{"request": request, "exc": exc},
|
context={"exc": exc},
|
||||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
+141
-157
@@ -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)
|
|
||||||
|
|||||||
+18
-5
@@ -27,8 +27,21 @@ def is_private_ip(hostname: str) -> bool:
|
|||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
except (socket.gaierror, socket.error):
|
except (socket.gaierror, socket.error):
|
||||||
# Cannot resolve - allow for testing/development
|
# Cannot resolve.
|
||||||
# In production, DNS should work properly
|
# Fail securely: block unresolved domains to prevent DNS rebinding
|
||||||
# Log this for debugging
|
# and SSRF bypasses via unresolvable addresses.
|
||||||
logger.warning(f"Could not resolve hostname: {hostname}")
|
logger.warning(f"Could not resolve hostname (blocking securely): {hostname}")
|
||||||
return False # Changed from True to False to allow external domains in tests
|
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
|
||||||
|
|||||||
+29
-5
@@ -162,12 +162,36 @@ def _inject_global_context(ctx: dict) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def template_response_with_version(*args, **kwargs):
|
def template_response_with_version(*args, **kwargs):
|
||||||
"""Wrapper for TemplateResponse to include version and CSRF token in all templates"""
|
"""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):
|
Handles both old-style and new-style Starlette TemplateResponse calls:
|
||||||
_inject_global_context(args[1])
|
- Old-style (Starlette <1.0): TemplateResponse(name, {"request": req, ...}, ...)
|
||||||
elif "context" in kwargs and isinstance(kwargs["context"], dict):
|
- 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"])
|
_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)
|
return original_template_response(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+2
-1
@@ -23,6 +23,7 @@ templates = Jinja2Templates(directory=str(_templates_dir))
|
|||||||
async def shared_link_view(request: Request, token: str):
|
async def shared_link_view(request: Request, token: str):
|
||||||
"""Render the public share landing page for a given token."""
|
"""Render the public share landing page for a given token."""
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
"shared_link_view.html",
|
"shared_link_view.html",
|
||||||
{"request": request, "token": token},
|
context={"token": token},
|
||||||
)
|
)
|
||||||
|
|||||||
+3
-3
@@ -430,8 +430,8 @@ class TestLoginFunction:
|
|||||||
# Verify TemplateResponse was called with correct context
|
# Verify TemplateResponse was called with correct context
|
||||||
mock_templates.TemplateResponse.assert_called_once()
|
mock_templates.TemplateResponse.assert_called_once()
|
||||||
call_args = mock_templates.TemplateResponse.call_args
|
call_args = mock_templates.TemplateResponse.call_args
|
||||||
assert call_args[0][0] == "login.html"
|
assert call_args[0][1] == "login.html"
|
||||||
context = call_args[0][1]
|
context = call_args.kwargs["context"]
|
||||||
assert context["error"] == "Test error"
|
assert context["error"] == "Test error"
|
||||||
assert context["message"] == "Test message"
|
assert context["message"] == "Test message"
|
||||||
|
|
||||||
@@ -450,7 +450,7 @@ class TestLoginFunction:
|
|||||||
|
|
||||||
mock_templates.TemplateResponse.assert_called_once()
|
mock_templates.TemplateResponse.assert_called_once()
|
||||||
call_args = mock_templates.TemplateResponse.call_args
|
call_args = mock_templates.TemplateResponse.call_args
|
||||||
context = call_args[0][1]
|
context = call_args.kwargs["context"]
|
||||||
assert context["error"] is None
|
assert context["error"] is None
|
||||||
assert context["message"] is None
|
assert context["message"] is None
|
||||||
|
|
||||||
|
|||||||
@@ -281,7 +281,7 @@ class TestLoginEndpoint:
|
|||||||
# Verify template was rendered with OAuth enabled
|
# Verify template was rendered with OAuth enabled
|
||||||
mock_templates.TemplateResponse.assert_called_once()
|
mock_templates.TemplateResponse.assert_called_once()
|
||||||
call_args = mock_templates.TemplateResponse.call_args
|
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["show_oauth"] is True
|
||||||
assert context["oauth_provider_name"] == "Test SSO"
|
assert context["oauth_provider_name"] == "Test SSO"
|
||||||
|
|
||||||
|
|||||||
@@ -520,14 +520,14 @@ class TestURLUploadAdditionalCoverage:
|
|||||||
assert exc_info.value.status_code == 400
|
assert exc_info.value.status_code == 400
|
||||||
|
|
||||||
def test_is_private_ip_unresolvable_hostname(self):
|
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
|
import socket as _socket
|
||||||
|
|
||||||
from app.utils.network import is_private_ip
|
from app.utils.network import is_private_ip
|
||||||
|
|
||||||
with patch("socket.getaddrinfo", side_effect=_socket.gaierror("nope")):
|
with patch("socket.getaddrinfo", side_effect=_socket.gaierror("nope")):
|
||||||
result = is_private_ip("nonexistent.invalid.hostname.test")
|
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):
|
def test_is_private_ip_hostname_resolves_to_private(self):
|
||||||
"""Cover branch where hostname resolves to a private IP (line 64-65)."""
|
"""Cover branch where hostname resolves to a private IP (line 64-65)."""
|
||||||
|
|||||||
@@ -69,8 +69,9 @@ class TestViewsBase:
|
|||||||
context = {"request": req}
|
context = {"request": req}
|
||||||
template_response_with_version("template.html", context)
|
template_response_with_version("template.html", context)
|
||||||
|
|
||||||
args, _ = mock_orig.call_args
|
args, kwargs = mock_orig.call_args
|
||||||
assert args[1].get("csrf_token") == "my-csrf"
|
context = kwargs.get("context", {})
|
||||||
|
assert context.get("csrf_token") == "my-csrf"
|
||||||
|
|
||||||
def test_kwargs_context_no_request(self):
|
def test_kwargs_context_no_request(self):
|
||||||
"""Test kwargs context path when request is not in context."""
|
"""Test kwargs context path when request is not in context."""
|
||||||
|
|||||||
@@ -55,8 +55,8 @@ class TestDarkModeTemplateInjection:
|
|||||||
|
|
||||||
captured = {}
|
captured = {}
|
||||||
|
|
||||||
def fake_original(name, ctx, **kw):
|
def fake_original(request_obj, name, context=None, **kw):
|
||||||
captured.update(ctx)
|
captured.update(context or {})
|
||||||
|
|
||||||
with patch("app.views.base.original_template_response", side_effect=fake_original):
|
with patch("app.views.base.original_template_response", side_effect=fake_original):
|
||||||
mock_request = MagicMock()
|
mock_request = MagicMock()
|
||||||
@@ -73,8 +73,8 @@ class TestDarkModeTemplateInjection:
|
|||||||
|
|
||||||
captured = {}
|
captured = {}
|
||||||
|
|
||||||
def fake_original(name, ctx, **kw):
|
def fake_original(request_obj, name, context=None, **kw):
|
||||||
captured.update(ctx)
|
captured.update(context or {})
|
||||||
|
|
||||||
with patch("app.views.base.original_template_response", side_effect=fake_original):
|
with patch("app.views.base.original_template_response", side_effect=fake_original):
|
||||||
mock_request = MagicMock()
|
mock_request = MagicMock()
|
||||||
|
|||||||
@@ -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"
|
||||||
@@ -345,7 +345,7 @@ class TestLoginPageSocialProviders:
|
|||||||
|
|
||||||
mock_templates.TemplateResponse.assert_called_once()
|
mock_templates.TemplateResponse.assert_called_once()
|
||||||
call_args = mock_templates.TemplateResponse.call_args
|
call_args = mock_templates.TemplateResponse.call_args
|
||||||
context = call_args[0][1]
|
context = call_args.kwargs.get("context", {})
|
||||||
assert context["social_providers"] == mock_providers
|
assert context["social_providers"] == mock_providers
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@@ -371,7 +371,7 @@ class TestLoginPageSocialProviders:
|
|||||||
|
|
||||||
mock_templates.TemplateResponse.assert_called_once()
|
mock_templates.TemplateResponse.assert_called_once()
|
||||||
call_args = mock_templates.TemplateResponse.call_args
|
call_args = mock_templates.TemplateResponse.call_args
|
||||||
context = call_args[0][1]
|
context = call_args.kwargs.get("context", {})
|
||||||
assert context["social_providers"] == {}
|
assert context["social_providers"] == {}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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="<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(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"
|
||||||
@@ -698,6 +698,18 @@ class TestURLUploadCoverageGaps:
|
|||||||
assert result is False
|
assert result is False
|
||||||
mock_getaddrinfo.assert_called_once()
|
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")
|
@patch("socket.getaddrinfo")
|
||||||
def test_is_private_ip_hostname_resolves_multiple_ips_all_public(self, mock_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)"""
|
"""Test hostname with multiple public IPs returns False (covers 65->61 loop branch)"""
|
||||||
|
|||||||
Reference in New Issue
Block a user