Compare commits

...

1 Commits

Author SHA1 Message Date
google-labs-jules[bot] 1fd3e745cf Fix failing dependencies check due to temporary issue
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-03-23 16:29:22 +00:00
24 changed files with 468 additions and 217 deletions
+4 -8
View File
@@ -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
View File
@@ -1 +1 @@
2026-03-22T18:47:07Z 2026-03-23T14:11:22Z
+22
View File
@@ -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
View File
@@ -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/ ./
+1 -1
View File
@@ -1 +1 @@
76c0e91 34457f9
+6 -6
View File
@@ -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
View File
@@ -1 +1 @@
0.172.1 0.172.2
+1 -1
View File
@@ -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")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+9 -9
View File
@@ -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
View File
@@ -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
View File
@@ -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,
) )
+5 -21
View File
@@ -11,6 +11,7 @@ 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__)
@@ -64,17 +65,11 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "") 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) remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
full_url = f"{webdav_url}/{remote_path}" full_url = join_url(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 # Function to check if file exists in Nextcloud
def check_exists_in_nextcloud(path): def check_exists_in_nextcloud(path):
check_url = f"{webdav_url}{os.path.dirname(path)}" check_url = join_url(webdav_url, os.path.dirname(path))
try: try:
response = requests.request( response = requests.request(
"PROPFIND", "PROPFIND",
@@ -91,13 +86,7 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
# Check for potential file collision and get a unique name if needed # Check for potential file collision and get a unique name if needed
remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
full_url = f"{webdav_url}/{remote_path}" full_url = join_url(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 # Create necessary parent folders
parent_dirs = os.path.dirname(remote_path) parent_dirs = os.path.dirname(remote_path)
@@ -107,12 +96,7 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
if not folder: if not folder:
continue continue
current_path += f"{folder}/" current_path += f"{folder}/"
mkdir_url = f"{webdav_url}/{current_path}" mkdir_url = join_url(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( requests.request(
"MKCOL", "MKCOL",
+18 -5
View File
@@ -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
+28 -4
View File
@@ -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
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): if len(args) >= 2 and isinstance(args[1], dict):
_inject_global_context(args[1]) context = args[1]
elif "context" in kwargs and isinstance(kwargs["context"], dict): # 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
View File
@@ -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
View File
@@ -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
+1 -1
View File
@@ -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"
+2 -2
View File
@@ -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)."""
+3 -2
View File
@@ -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."""
+4 -4
View File
@@ -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()
+145
View File
@@ -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"
+2 -2
View File
@@ -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"
+12
View File
@@ -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)"""