diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml new file mode 100644 index 000000000..a78656519 --- /dev/null +++ b/.github/codeql/codeql-config.yml @@ -0,0 +1,15 @@ +name: TimeTracker CodeQL config + +paths-ignore: + - tests/** + - "**/node_modules/**" + - "**/dist/**" + - "**/build/**" + - "**/coverage/**" + - desktop/**/node_modules/** + - mobile/**/node_modules/** + - .github/workflows-archive/** + +# Security findings only (not maintainability/quality noise from security-and-quality) +queries: + - uses: security-extended diff --git a/.github/codeql/timetracker-python-models/models/sanitizers.yml b/.github/codeql/timetracker-python-models/models/sanitizers.yml new file mode 100644 index 000000000..0b37599a9 --- /dev/null +++ b/.github/codeql/timetracker-python-models/models/sanitizers.yml @@ -0,0 +1,36 @@ +extensions: + - addsTo: + pack: codeql/python-all + extensible: barrierModel + data: + - ["app.utils.safe_redirect", "Member[safe_redirect_target].ReturnValue", "url-redirection"] + - ["app.utils.safe_redirect", "Member[_rebuild_relative].ReturnValue", "url-redirection"] + - ["app.utils.safe_redirect", "Member[_rebuild_same_host].ReturnValue", "url-redirection"] + - ["app.utils.error_handling", "Member[sanitize_log_value].ReturnValue", "log-injection"] + - ["app.utils.paths", "Member[safe_path_under].ReturnValue", "path-injection"] + - ["app.utils.paths", "Member[safe_path_under_or_none].ReturnValue", "path-injection"] + - ["app.utils.outbound_url", "Member[validate_outbound_url].ReturnValue", "request-forgery"] + - ["app.utils.outbound_url", "Member[validate_oidc_issuer_url].ReturnValue", "request-forgery"] + + - addsTo: + pack: codeql/python-all + extensible: barrierGuardModel + data: + - [ + "app.utils.safe_redirect", + "Member[is_safe_next_url].Argument[0,next_url:]", + "true", + "url-redirection", + ] + - [ + "app.utils.safe_redirect", + "Member[is_safe_referrer].Argument[0,referrer:]", + "true", + "url-redirection", + ] + - [ + "app.utils.safe_redirect", + "Member[is_safe_relative_url].Argument[0,next_url:]", + "true", + "url-redirection", + ] diff --git a/.github/codeql/timetracker-python-models/qlpack.yml b/.github/codeql/timetracker-python-models/qlpack.yml new file mode 100644 index 000000000..ce2ac7c61 --- /dev/null +++ b/.github/codeql/timetracker-python-models/qlpack.yml @@ -0,0 +1,7 @@ +name: drytrix/timetracker-python-models +version: 0.0.1 +library: true +extensionTargets: + codeql/python-all: "*" +dataExtensions: + - models/**/*.yml diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 000000000..7c2039f75 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,48 @@ +name: CodeQL + +on: + push: + branches: [develop, main] + pull_request: + branches: [develop, main] + schedule: + # Weekly re-scan (Mondays 06:00 UTC) + - cron: "0 6 * * 1" + workflow_dispatch: + +jobs: + analyze: + name: Analyze (${{ matrix.language }}) + runs-on: ubuntu-latest + permissions: + actions: read + contents: read + security-events: write + + strategy: + fail-fast: false + matrix: + include: + - language: python + build-mode: none + - language: javascript-typescript + build-mode: none + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Initialize CodeQL + uses: github/codeql-action/init@v4 + with: + languages: ${{ matrix.language }} + build-mode: ${{ matrix.build-mode }} + config-file: ./.github/codeql/codeql-config.yml + + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v4 + with: + category: "/language:${{ matrix.language }}" + env: + # Load local models-as-data pack for Python only (safe_redirect, safe_path_under, …) + CODEQL_ACTION_EXTRA_OPTIONS: ${{ matrix.language == 'python' && format('{{"database":{{"run-queries":["--additional-packs={0}/.github/codeql","--extension-packs=drytrix/timetracker-python-models"]}}}}', github.workspace) || '' }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ce18e4d9..d7921b8df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [5.18.2] - 2026-10-13 + +### Security + +- **CodeQL / XSS / redirects** — Clear CodeQL findings; harden DOM sinks and PDF templates; block open redirects via referrer/`next`; centralize `safe_redirect`; keep 5xx API errors generic; rebuild OIDC discovery URLs from validated issuer components. +- **Desktop (Dependabot #222)** — Override transitive `sprintf-js` with a vendored build that caps numeric precision (GHSA-hp3w-g68c-fv3c / CVE-2026-97058) until upstream publishes a fixed release. +- **CSP** — Add nonces to remaining inline scripts. + +### Fixed + +- **Timer review / idle response** — Import `db` in timer review and idle-response handlers. +- **Docker** — Refresh base packages and upgrade pip/setuptools. +- **Models** — Strip UTF-8 BOMs from timesheet/time-off model modules. + +### Changed + +- **CI** — Add CodeQL workflow to resume code scanning. + +### Documentation + +- **Version** — Bumped `setup.py` to **5.18.2**; `VERSION` and desktop/browser-extension/mobile client versions aligned. + ## [5.18.1] - 2026-10-11 ### Fixed diff --git a/Dockerfile b/Dockerfile index 879e87793..651bb18ee 100644 --- a/Dockerfile +++ b/Dockerfile @@ -40,10 +40,13 @@ ENV TZ=Europe/Rome ENV DONATE_HIDE_PUBLIC_KEY_FILE=/app/donate_hide_public.pem LABEL org.opencontainers.image.description="Self-hosted time tracking web application for projects, clients, and reports." -# Install all system dependencies in a single layer +# Install all system dependencies in a single layer. +# apt-get upgrade picks up security fixes for the slim-bookworm base image. RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ --mount=type=cache,target=/var/lib/apt,sharing=locked \ - apt-get update && apt-get install -y --no-install-recommends \ + apt-get update \ + && apt-get upgrade -y --no-install-recommends \ + && apt-get install -y --no-install-recommends \ # Core utilities curl \ tzdata \ @@ -73,10 +76,12 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ # Set work directory WORKDIR /app -# Install Python dependencies with cache mount for pip +# Upgrade pip/setuptools first (base image pins are often stale / CVE-flagged), +# then install app requirements (includes Authlib pin from requirements.txt). COPY requirements.txt . RUN --mount=type=cache,target=/root/.cache/pip \ - pip install --no-cache-dir -r requirements.txt + pip install --no-cache-dir --upgrade "pip>=25.0" "setuptools>=78.1.1" wheel \ + && pip install --no-cache-dir -r requirements.txt # Create non-root user early (before copying files) RUN useradd -m -u 1000 timetracker diff --git a/README.md b/README.md index d05eb0c1a..0ec0f4485 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ docker compose -f docker-compose.nas.yml up -d # open http://localhost:8080 — first login creates the admin ``` -**Latest release: v5.18.1** — see [CHANGELOG.md](CHANGELOG.md). +**Latest release: v5.18.2** — see [CHANGELOG.md](CHANGELOG.md). More install paths (HTTPS production, NAS UI paste, cloud): [Install options](#install-options). diff --git a/VERSION b/VERSION index 125e07c97..2ad48e580 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -5.18.1 +5.18.2 diff --git a/app/__init__.py b/app/__init__.py index 7ea85fd18..aaa784689 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -96,7 +96,11 @@ def _rate_limit_key(): def log_event(name: str, **kwargs): """Log an event with structured JSON format including request context""" try: - extra = {"request_id": getattr(g, "request_id", None), "event": name, **kwargs} + from app.utils.error_handling import sanitize_log_value + + safe_name = sanitize_log_value(name, max_len=200) + safe_kwargs = {k: sanitize_log_value(v) for k, v in kwargs.items()} + extra = {"request_id": getattr(g, "request_id", None), "event": safe_name, **safe_kwargs} try: from app.telemetry.otel_setup import get_trace_context_for_logs, is_otel_tracing_active @@ -104,9 +108,9 @@ def log_event(name: str, **kwargs): extra.update(get_trace_context_for_logs()) except Exception: logging.getLogger(__name__).debug("Could not attach trace context to log_event", exc_info=True) - json_logger.info(name, extra=extra) + json_logger.info(safe_name, extra=extra) except Exception as e: - logging.getLogger(__name__).debug("Structured log_event failed: %s", e) + logging.getLogger(__name__).debug("Structured log_event failed: %s", type(e).__name__) def identify_user(user_id, properties=None): @@ -600,13 +604,23 @@ def _normalize_locale(locale_code): def _handle_unauthorized(): """Send portal sessions to the client portal login instead of the main app login.""" from app.utils.client_portal_session import get_active_native_client_portal_id + from app.utils.safe_redirect import is_safe_relative_url + + # Prefer a same-origin relative path for next= (never pass absolute request.url) + next_target = request.path or "/" + if request.query_string: + try: + next_target = f"{request.path}?{request.query_string.decode('utf-8', 'ignore')}" + except Exception: + next_target = request.path or "/" + next_kw = {"next": next_target} if is_safe_relative_url(next_target) else {} if get_active_native_client_portal_id() or request.path.startswith("/client-portal"): flash(_("Please log in to access the client portal."), "error") - return redirect(url_for("client_portal.login", next=request.url)) + return redirect(url_for("client_portal.login", **next_kw)) flash(login_manager.login_message, login_manager.login_message_category) - return redirect(url_for(login_manager.login_view, next=request.url)) + return redirect(url_for(login_manager.login_view, **next_kw)) # Internationalization selector handled via babel.init_app(locale_selector=...) @@ -855,12 +869,14 @@ def prom_start_timer(): def log_login_requests(): try: if request.path == "/login": + from app.utils.error_handling import sanitize_log_value + app.logger.info( "%s %s from %s UA=%s", - request.method, - request.path, - request.headers.get("X-Forwarded-For") or request.remote_addr, - request.headers.get("User-Agent"), + sanitize_log_value(request.method), + sanitize_log_value(request.path), + sanitize_log_value(request.headers.get("X-Forwarded-For") or request.remote_addr), + sanitize_log_value(request.headers.get("User-Agent")), ) except Exception: app.logger.debug("log_login_requests failed", exc_info=True) @@ -889,12 +905,14 @@ def record_metrics_and_log(response): try: # Log write operations if request.method in ("POST", "PUT", "PATCH", "DELETE"): + from app.utils.error_handling import sanitize_log_value + app.logger.info( "%s %s -> %s from %s", - request.method, - request.path, + sanitize_log_value(request.method), + sanitize_log_value(request.path), response.status_code, - request.headers.get("X-Forwarded-For") or request.remote_addr, + sanitize_log_value(request.headers.get("X-Forwarded-For") or request.remote_addr), ) except Exception: app.logger.debug("Write-operation request logging failed", exc_info=True) @@ -1144,15 +1162,17 @@ def handle_csrf_error(e): user_id = getattr(_cu, "id", None) if getattr(_cu, "is_authenticated", False) else None except Exception: user_id = None + from app.utils.error_handling import sanitize_log_value + app.logger.warning( "CSRF failure: path=%s method=%s form=%s json=%s ref=%s user=%s reason=%s", - request.path, - request.method, + sanitize_log_value(request.path), + sanitize_log_value(request.method), bool(request.form), request.is_json, - request.referrer, + sanitize_log_value(request.referrer), user_id, - getattr(e, "description", ""), + sanitize_log_value(getattr(e, "description", "")), ) except Exception: app.logger.debug("CSRF failure diagnostic logging failed", exc_info=True) @@ -1172,29 +1192,23 @@ def handle_csrf_error(e): except Exception: is_authed = False + from app.utils.safe_redirect import is_safe_next_url, safe_redirect_target + if (not is_authed) or request.endpoint == "auth.login" or request.path.rstrip("/").endswith("/login"): next_url = request.args.get("next") or request.form.get("next") - if next_url: + if next_url and is_safe_next_url(next_url): dest = url_for("auth.login", next=next_url) else: dest = url_for("auth.login") else: - dest = url_for("main.dashboard") - try: - # Prefer returning to the same path that failed when no referrer is present - if request.path: - dest = request.path - except Exception: - pass - try: - ref = request.referrer - if ref: - ref_host = urlparse(ref).netloc - cur_host = urlparse(request.host_url).netloc - if ref_host and ref_host == cur_host: - dest = ref - except Exception: - app.logger.debug("CSRF redirect referrer parse failed", exc_info=True) + dest = safe_redirect_target( + request.referrer, + allow_absolute_same_host=True, + fallback=request.path if (request.path or "").startswith("/") else None, + fallback_endpoint="main.dashboard", + ) + # dest is url_for(...) or safe_redirect_target(...) + # codeql[py/url-redirection] return redirect(dest) # JSON/XHR fall-through @@ -1224,21 +1238,23 @@ def handle_csrf_error(e): is_authed = False if (not is_authed) or request.endpoint == "auth.login" or request.path.rstrip("/").endswith("/login"): + from app.utils.safe_redirect import is_safe_next_url + next_url = request.args.get("next") or request.form.get("next") - if next_url: + if next_url and is_safe_next_url(next_url): return redirect(url_for("auth.login", next=next_url)) return redirect(url_for("auth.login")) - dest = request.path or url_for("main.dashboard") - try: - ref = request.referrer - if ref: - ref_host = urlparse(ref).netloc - cur_host = urlparse(request.host_url).netloc - if ref_host and ref_host == cur_host: - dest = ref - except Exception: - app.logger.debug("CSRF redirect referrer parse failed", exc_info=True) + from app.utils.safe_redirect import safe_redirect_target + + dest = safe_redirect_target( + request.referrer, + allow_absolute_same_host=True, + fallback=request.path if (request.path or "").startswith("/") else None, + fallback_endpoint="main.dashboard", + ) + # dest produced by safe_redirect_target + # codeql[py/url-redirection] return redirect(dest) # Expose csrf_token() in Jinja templates even without FlaskForm diff --git a/app/integrations/github_connector.py b/app/integrations/github_connector.py index cabab0726..e2da457ac 100644 --- a/app/integrations/github_connector.py +++ b/app/integrations/github_connector.py @@ -174,8 +174,8 @@ def test_connection(self) -> Dict[str, Any]: timeout=HTTP_TIMEOUT, ) except requests.RequestException as exc: - logger.warning("GitHub test_connection request failed: %s", exc) - return {"success": False, "message": f"Connection error: {exc}"} + logger.warning("GitHub test_connection request failed: %s", type(exc).__name__) + return {"success": False, "message": "Connection error"} if resp.status_code == 200: return {"success": True, "message": f"Connected as {resp.json().get('login', 'Unknown')}"} @@ -207,8 +207,8 @@ def sync(self) -> Dict[str, Any]: timeout=HTTP_TIMEOUT, ) except requests.RequestException as exc: - logger.warning("GitHub sync request failed for %s/%s: %s", owner, repo, exc) - return {"ok": False, "error": f"GitHub request failed: {exc}"} + logger.warning("GitHub sync request failed for %s/%s: %s", owner, repo, type(exc).__name__) + return {"ok": False, "error": "GitHub request failed"} if resp.status_code != 200: return {"ok": False, "error": f"GitHub returned HTTP {resp.status_code}"} @@ -247,8 +247,8 @@ def sync(self) -> Dict[str, Any]: db.session.add(task) synced += 1 except Exception as exc: # pragma: no cover - defensive - logger.warning("Failed to create task for issue #%s: %s", number, exc) - errors.append(f"#{number}: {exc}") + logger.warning("Failed to create task for issue #%s: %s", number, type(exc).__name__) + errors.append(f"#{number}: failed to create task") if synced: if not safe_commit("github_connector_sync", {"integration_id": self.integration.id}): diff --git a/app/integrations/google_calendar_connector.py b/app/integrations/google_calendar_connector.py index e9dcedc08..6184ce4ae 100644 --- a/app/integrations/google_calendar_connector.py +++ b/app/integrations/google_calendar_connector.py @@ -168,16 +168,16 @@ def test_connection(self) -> Dict[str, Any]: return {"success": False, "message": "Integration not configured"} try: token = self._refresh_token_if_needed() - except Exception as exc: - return {"success": False, "message": str(exc)} + except Exception: + return {"success": False, "message": "Token refresh failed"} try: resp = requests.get( f"{GOOGLE_CALENDAR_API}/users/me/calendarList", headers={"Authorization": f"Bearer {token}"}, timeout=HTTP_TIMEOUT, ) - except requests.RequestException as exc: - return {"success": False, "message": f"Connection error: {exc}"} + except requests.RequestException: + return {"success": False, "message": "Connection error"} if resp.status_code != 200: return {"success": False, "message": f"Google returned HTTP {resp.status_code}"} return {"success": True, "message": f"Connected ({len(resp.json().get('items') or [])} calendars)"} @@ -363,8 +363,8 @@ def sync(self) -> Dict[str, Any]: result = {"ok": True, "imported": 0, "skipped": 0, "exported": 0, "errors": []} try: token = self._refresh_token_if_needed() - except Exception as exc: - return {"ok": False, "error": str(exc)} + except Exception: + return {"ok": False, "error": "Token refresh failed"} try: if direction in ("import", "both"): @@ -377,9 +377,9 @@ def sync(self) -> Dict[str, Any]: result["exported"] += exp.get("exported", 0) result["errors"].extend(exp.get("errors", [])) except Exception as exc: - logger.warning("Google sync failed: %s", exc) + logger.warning("Google sync failed: %s", type(exc).__name__) result["ok"] = False - result["error"] = str(exc) + result["error"] = "Sync failed" try: cfg["last_sync_at"] = datetime.now(timezone.utc).isoformat() @@ -425,7 +425,7 @@ def _sync_import(self, token: str, cfg: Dict[str, Any]) -> Dict[str, Any]: timeout=HTTP_TIMEOUT, ) except requests.RequestException as exc: - return {"imported": 0, "skipped": 0, "errors": [f"Calendar fetch failed: {exc}"]} + return {"imported": 0, "skipped": 0, "errors": ["Calendar fetch failed"]} if resp.status_code != 200: return {"imported": 0, "skipped": 0, "errors": [f"Google returned HTTP {resp.status_code}"]} @@ -489,7 +489,7 @@ def _sync_import(self, token: str, cfg: Dict[str, Any]) -> Dict[str, Any]: db.session.add(entry) imported += 1 except Exception as exc: - errors.append(f"{event_id}: {exc}") + errors.append(f"{event_id}: import failed") if imported: if not safe_commit("google_connector_import", {"integration_id": self.integration.id}): @@ -553,7 +553,7 @@ def _sync_export(self, token: str, cfg: Dict[str, Any]) -> Dict[str, Any]: timeout=HTTP_TIMEOUT, ) except requests.RequestException as exc: - errors.append(f"entry {entry.id}: {exc}") + errors.append(f"entry {entry.id}: export failed") continue if resp.status_code in (200, 201): exported += 1 diff --git a/app/models/api_token.py b/app/models/api_token.py index dbd517a49..faed4505c 100644 --- a/app/models/api_token.py +++ b/app/models/api_token.py @@ -52,12 +52,47 @@ def generate_token(): random_part = secrets.token_urlsafe(32)[:32] return f"tt_{random_part}" + @staticmethod + def _token_pepper() -> bytes: + from flask import current_app, has_app_context + + pepper = "" + if has_app_context(): + pepper = ( + current_app.config.get("API_TOKEN_PEPPER") + or current_app.config.get("SECRET_KEY") + or "" + ) + if isinstance(pepper, bytes): + return pepper + return str(pepper or "timetracker-api-token").encode("utf-8") + @staticmethod def hash_token(token): - """Hash a token for storage""" + """ + Hash a high-entropy API token for indexed lookup storage. + + Uses HMAC-SHA256 with the app secret (or API_TOKEN_PEPPER) as key. + This is a lookup hash for secrets.token_urlsafe tokens, not a password KDF. + """ import hashlib + import hmac - return hashlib.sha256(token.encode()).hexdigest() + # codeql[py/weak-sensitive-data-hashing]: API token lookup HMAC, not password hashing + return hmac.new(ApiToken._token_pepper(), token.encode("utf-8"), hashlib.sha256).hexdigest() + + @staticmethod + def legacy_hash_token(token): + """Pre-HMAC SHA256 lookup hash (for verifying existing stored tokens).""" + import hashlib + + # codeql[py/weak-sensitive-data-hashing]: legacy API token lookup hash, not password hashing + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + @classmethod + def lookup_hashes(cls, token): + """Return candidate hashes for DB lookup (HMAC first, then legacy).""" + return [cls.hash_token(token), cls.legacy_hash_token(token)] @classmethod def create_token(cls, user_id, name, description="", scopes="", expires_days=None, client_id=None): @@ -96,8 +131,13 @@ def create_token(cls, user_id, name, description="", scopes="", expires_days=Non return api_token, plain_token def verify_token(self, plain_token): - """Verify if the provided token matches this record""" - return self.token_hash == self.hash_token(plain_token) + """Verify if the provided token matches this record (HMAC or legacy SHA256).""" + import hmac as hmac_mod + + for candidate in self.lookup_hashes(plain_token): + if hmac_mod.compare_digest(self.token_hash, candidate): + return True + return False def is_valid(self): """Check if token is valid (active and not expired)""" diff --git a/app/models/calendar_event.py b/app/models/calendar_event.py index a67b42298..a9a1519f1 100644 --- a/app/models/calendar_event.py +++ b/app/models/calendar_event.py @@ -149,8 +149,15 @@ def get_events_in_range(user_id, start_date, end_date, include_tasks=False, incl print(f" include_time_entries={include_time_entries} (type: {type(include_time_entries)})") print(f"{'*'*80}\n") + from app.utils.error_handling import sanitize_log_value + logger.info( - f"get_events_in_range called: user_id={user_id}, start={start_date}, end={end_date}, include_tasks={include_tasks}, include_time_entries={include_time_entries}" + "get_events_in_range called: user_id=%s, start=%s, end=%s, include_tasks=%s, include_time_entries=%s", + sanitize_log_value(user_id), + sanitize_log_value(start_date), + sanitize_log_value(end_date), + sanitize_log_value(include_tasks), + sanitize_log_value(include_time_entries), ) result = {"events": [], "tasks": [], "time_entries": []} @@ -173,7 +180,7 @@ def get_events_in_range(user_id, start_date, end_date, include_tasks=False, incl # Optionally include tasks with due dates if include_tasks: print(f"MODEL - Querying tasks for user {user_id}") - logger.info(f"Querying tasks for user {user_id}") + logger.info("Querying tasks for user %s", sanitize_log_value(user_id)) tasks = Task.query.filter( Task.assigned_to == user_id, Task.due_date.isnot(None), @@ -205,7 +212,7 @@ def get_events_in_range(user_id, start_date, end_date, include_tasks=False, incl # Optionally include time entries if include_time_entries: print(f"MODEL - Querying time entries for user {user_id}") - logger.info(f"Querying time entries for user {user_id}") + logger.info("Querying time entries for user %s", sanitize_log_value(user_id)) time_entries = ( TimeEntry.query.filter( TimeEntry.user_id == user_id, diff --git a/app/models/time_off.py b/app/models/time_off.py index fc0228efb..613e1eafc 100644 --- a/app/models/time_off.py +++ b/app/models/time_off.py @@ -1,4 +1,4 @@ -import enum +import enum from datetime import datetime from sqlalchemy import Enum as SQLEnum diff --git a/app/models/timesheet_period.py b/app/models/timesheet_period.py index a45a44b07..8ba804f63 100644 --- a/app/models/timesheet_period.py +++ b/app/models/timesheet_period.py @@ -1,4 +1,4 @@ -import enum +import enum from datetime import date, datetime from sqlalchemy import Enum as SQLEnum diff --git a/app/models/timesheet_policy.py b/app/models/timesheet_policy.py index ddf826799..9462dbeb1 100644 --- a/app/models/timesheet_policy.py +++ b/app/models/timesheet_policy.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import datetime from app import db diff --git a/app/routes/admin.py b/app/routes/admin.py index f62e0616a..e39c12577 100644 --- a/app/routes/admin.py +++ b/app/routes/admin.py @@ -6,6 +6,7 @@ import requests from flask import ( Blueprint, + Response, current_app, flash, jsonify, @@ -40,7 +41,7 @@ ) from app.utils.auth_method import auth_includes_ldap, auth_includes_oidc, normalize_auth_method from app.utils.db import safe_commit -from app.utils.error_handling import safe_log +from app.utils.error_handling import client_error_response, client_error_payload, DEFAULT_CLIENT_ERROR, safe_log from app.utils.installation import get_installation_config from app.utils.invoice_numbering import sanitize_invoice_pattern, sanitize_invoice_prefix, validate_invoice_pattern from app.utils.permissions import admin_or_permission_required @@ -54,6 +55,163 @@ _VALID_ROUNDING_METHODS = {"nearest", "up", "down", "boundary"} _VALID_ROUNDING_MINIMUMS = {0, 5, 10, 15, 30, 60} +_PDF_PREVIEW_ALLOWED_TAGS = frozenset( + { + "html", + "head", + "body", + "meta", + "title", + "style", + "div", + "span", + "p", + "br", + "hr", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "table", + "thead", + "tbody", + "tfoot", + "tr", + "th", + "td", + "ul", + "ol", + "li", + "strong", + "em", + "b", + "i", + "u", + "s", + "small", + "sup", + "sub", + "img", + "a", + "section", + "header", + "footer", + "main", + "article", + "pre", + "code", + "blockquote", + } +) +_PDF_PREVIEW_ALLOWED_ATTRS = { + "*": ["class", "id", "style"], + "img": ["src", "alt", "width", "height", "style", "class", "id"], + "a": ["href", "title", "style", "class", "id"], + "td": ["colspan", "rowspan", "style", "class", "id"], + "th": ["colspan", "rowspan", "style", "class", "id"], + "meta": ["charset", "name", "content"], + "style": ["type", "media"], +} + + +def _pdf_preview_html_response(page_html: str) -> Response: + """Return PDF preview HTML with bleach sanitization and a restrictive CSP.""" + import bleach + from bleach.css_sanitizer import CSSSanitizer + + # Broad but non-executable CSS allowlist for invoice/quote layout preview. + css_sanitizer = CSSSanitizer( + allowed_css_properties=[ + "color", + "background", + "background-color", + "background-image", + "background-size", + "background-position", + "background-repeat", + "border", + "border-top", + "border-right", + "border-bottom", + "border-left", + "border-color", + "border-style", + "border-width", + "border-radius", + "border-collapse", + "border-spacing", + "box-sizing", + "box-shadow", + "display", + "position", + "top", + "right", + "bottom", + "left", + "float", + "clear", + "width", + "height", + "max-width", + "max-height", + "min-width", + "min-height", + "margin", + "margin-top", + "margin-right", + "margin-bottom", + "margin-left", + "padding", + "padding-top", + "padding-right", + "padding-bottom", + "padding-left", + "font", + "font-family", + "font-size", + "font-weight", + "font-style", + "line-height", + "letter-spacing", + "text-align", + "text-decoration", + "text-transform", + "vertical-align", + "white-space", + "overflow", + "opacity", + "z-index", + "table-layout", + "flex", + "flex-direction", + "flex-wrap", + "justify-content", + "align-items", + "align-self", + "gap", + "grid-template-columns", + "grid-template-rows", + "transform", + "page-break-after", + "page-break-before", + "page-break-inside", + "size", + ] + ) + cleaned = bleach.clean( + page_html, + tags=_PDF_PREVIEW_ALLOWED_TAGS, + attributes=_PDF_PREVIEW_ALLOWED_ATTRS, + css_sanitizer=css_sanitizer, + strip=True, + ) + resp = Response(cleaned, mimetype="text/html") + resp.headers["Content-Security-Policy"] = "script-src 'none'; object-src 'none'; base-uri 'none'" + resp.headers["X-Content-Type-Options"] = "nosniff" + return resp + def _admin_settings_template_kwargs( settings_obj, timezones, kiosk_settings, peppol_env_enabled, ai_config, system_instance_id @@ -285,16 +443,55 @@ def get_font_style(font_name): return "normal" # Helper function to convert color (supports hex, rgb, named colors) + _NAMED_COLORS = frozenset( + {"black", "white", "red", "blue", "green", "yellow", "cyan", "magenta", "gray", "grey", "orange", "purple"} + ) + import re as _re_color + + _HEX_COLOR_RE = _re_color.compile(r"^#(?:[0-9a-fA-F]{3}|[0-9a-fA-F]{6}|[0-9a-fA-F]{8})$") + def format_color(color): + """Return an allowlisted CSS color; never echo arbitrary user strings into markup.""" if not color: return "#000000" - # If already hex or named color, return as-is - if color.startswith("#") or color in ["black", "white", "red", "blue", "green", "yellow", "cyan", "magenta"]: - return color - # If RGB tuple/list, convert to hex if isinstance(color, (list, tuple)) and len(color) >= 3: - return f"#{int(color[0]):02x}{int(color[1]):02x}{int(color[2]):02x}" - return str(color) + try: + r, g, b = (max(0, min(255, int(color[i]))) for i in range(3)) + return f"#{r:02x}{g:02x}{b:02x}" + except (TypeError, ValueError): + return "#000000" + if not isinstance(color, str): + return "#000000" + candidate = color.strip() + if _HEX_COLOR_RE.match(candidate): + return candidate.lower() + if candidate.lower() in _NAMED_COLORS: + return candidate.lower() + return "#000000" + + def format_align(align): + return align if align in ("left", "right", "center", "justify") else "left" + + def safe_img_src(source: str) -> str: + """Allow only data: image URLs or jailed template-image paths.""" + if not source or not isinstance(source, str): + return "" + source = source.strip() + if source.startswith("data:image/"): + # Reject SVG data URLs (scriptable); allow raster only + header = source.split(",", 1)[0].lower() + if "svg" in header: + return "" + return source + if source.startswith("/uploads/template_images/"): + from werkzeug.utils import secure_filename as _secure_filename + + raw_name = source.split("/uploads/template_images/", 1)[-1] + filename = _secure_filename(raw_name.replace("\\", "/").split("/")[-1]) + if not filename: + return "" + return f"/uploads/template_images/{filename}" + return "" # Helper function to render text with Jinja2-like template variables def render_text_template(text, data_obj, settings_obj=None): @@ -332,7 +529,8 @@ def replace_var(match): except Exception: return match.group(0) # Return original on error - return re.sub(r"\{\{\s*([^}]+)\s*\}\}", replace_var, text) + # Avoid polynomial ReDoS from \s* adjacent to [^}]+; whitespace is stripped in replace_var. + return re.sub(r"\{\{([^{}]+)\}\}", replace_var, text) # Build HTML from elements html_parts = ['
'] if invoice else ['
'] @@ -370,17 +568,20 @@ def replace_var(match): width_px_elem = int(width * 96 / 72) font_name = style.get("font", "Helvetica") - font_size = style.get("size", 10) + try: + font_size = float(style.get("size", 10)) + except (TypeError, ValueError): + font_size = 10.0 color = format_color(style.get("color", "#000000")) - align = style.get("align", "left") - valign = style.get("valign", "top") + align = format_align(style.get("align", "left")) + valign = style.get("valign", "top") if style.get("valign") in ("top", "middle", "bottom") else "top" # Build complete style string style_parts = [style_str_base] style_parts.append(f"width: {width_px_elem}px") if height: style_parts.append(f"height: {int(height * 96 / 72)}px") - style_parts.append(f"font-family: {get_font_family(font_name)}") + style_parts.append(f"font-family: {html_escape.escape(get_font_family(font_name), quote=True)}") style_parts.append(f"font-size: {font_size}pt") style_parts.append(f"font-weight: {get_font_weight(font_name)}") style_parts.append(f"font-style: {get_font_style(font_name)}") @@ -394,10 +595,8 @@ def replace_var(match): data_obj = invoice if invoice else quote rendered_text = render_text_template(text, data_obj, settings) if (data_obj or settings) else text - # Escape HTML but preserve any remaining template syntax + # Escape HTML fully — never re-inject {{ }} (avoids SSTI if HTML is later templated) text_escaped = html_escape.escape(rendered_text) - # Restore template syntax if any remains (shouldn't after rendering, but just in case) - text_escaped = text_escaped.replace("<{{", "{{").replace("}}>", "}}") html_parts.append(f'
{text_escaped}
') @@ -407,50 +606,39 @@ def replace_var(match): width_px_elem = int(width * 96 / 72) height_px_elem = int(height * 96 / 72) source = element.get("source", "") - is_decorative = element.get("decorative", False) - - # Handle base64 data URLs or file paths - img_src = "" - if source.startswith("data:"): - img_src = source - elif source.startswith("/uploads/template_images/"): - # Template image - convert to base64 for PDF generation + + img_src = safe_img_src(source) + if img_src.startswith("/uploads/template_images/"): try: from app.utils.template_filters import get_image_base64 + from werkzeug.utils import secure_filename as _secure_filename - # Extract filename from URL - filename = source.split("/uploads/template_images/")[-1] - # Build file path relative to app root (as get_image_base64 expects) + filename = _secure_filename(img_src.rsplit("/", 1)[-1]) relative_path = f"app/static/uploads/template_images/{filename}" - # Convert to base64 - img_src = get_image_base64(relative_path) or source # Fallback to URL if conversion fails + img_src = get_image_base64(relative_path) or img_src except Exception as e: - # If conversion fails, use the URL directly - current_app.logger.warning(f"Failed to convert template image to base64: {e}") - img_src = source - elif source.startswith("/") or source.startswith("http"): - img_src = source - else: - # Assume it's a relative path or placeholder - if source: - img_src = source - else: - # Placeholder for decorative images without source - img_src = "data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' width='100' height='100'%3E%3Crect fill='%23ddd' width='100' height='100'/%3E%3Ctext x='50%25' y='50%25' text-anchor='middle' dy='.3em' fill='%23999'%3EImage%3C/text%3E%3C/svg%3E" + safe_log(current_app.logger, "warning", "Failed to convert template image to base64: %s", e) style_parts = [style_str_base] style_parts.append(f"width: {width_px_elem}px") style_parts.append(f"height: {height_px_elem}px") style_str = "; ".join(style_parts) + ";" - if img_src and not img_src.startswith("data:image/svg+xml"): + if img_src and img_src.startswith("data:image/") and "svg" not in img_src[:64].lower(): + html_parts.append( + f'Decorative image' + ) + elif img_src.startswith("/uploads/template_images/"): html_parts.append( - f'Decorative image' + f'Decorative image' ) else: - # Show placeholder for decorative images without source html_parts.append( - f'
Decorative Image
' + f'
' + f"Decorative Image
" ) elif elem_type == "rectangle": @@ -536,10 +724,16 @@ def replace_var(match): # Build header row for col in columns: header = col.get("header", "") - align = col.get("align", "left") + align = format_align(col.get("align", "left")) col_width = col.get("width", None) - width_attr = f' width="{int(col_width * 96 / 72)}px"' if col_width else "" - table_html += f'{html_escape.escape(header)}' + try: + width_attr = f' width="{int(float(col_width) * 96 / 72)}"' if col_width is not None else "" + except (TypeError, ValueError): + width_attr = "" + table_html += ( + f'{html_escape.escape(str(header))}' + ) table_html += "" # Resolve table data from element's data source (e.g. invoice.all_line_items or invoice.items) @@ -588,29 +782,33 @@ def replace_var(match): table_html += "" for col in columns: field = col.get("field", "") - align = col.get("align", "left") - # Try to get value from item + align = format_align(col.get("align", "left")) + # Try to get value from item (field must be a simple identifier) value = "" - try: - if hasattr(item, field): - value = str(getattr(item, field, "")) - elif isinstance(item, dict): - value = str(item.get(field, "")) - else: + if isinstance(field, str) and field.isidentifier(): + try: + if hasattr(item, field): + value = str(getattr(item, field, "")) + elif isinstance(item, dict): + value = str(item.get(field, "")) + except Exception as e: + safe_log( + current_app.logger, "debug", "Template value for field %s failed: %s", field, e + ) value = "" - except Exception as e: - safe_log(current_app.logger, "debug", "Template value for field %s failed: %s", field, e) - value = "" value_escaped = html_escape.escape(str(value)) - table_html += f'{value_escaped}' + table_html += ( + f'{value_escaped}' + ) table_html += "" else: # No data available, show template placeholders table_html += "" for col in columns: - field = col.get("field", "") - align = col.get("align", "left") + field = col.get("field", "") if isinstance(col.get("field", ""), str) else "" + align = format_align(col.get("align", "left")) placeholder = f"{{{{ {field} }}}}" table_html += ( f'{html_escape.escape(placeholder)}' @@ -1852,7 +2050,15 @@ def admin_ldap_test(): """Test LDAP connectivity (service bind + user subtree count). Returns JSON only.""" from app.services.ldap_service import LDAPService - return jsonify(LDAPService.test_connection()) + result = LDAPService.test_connection() + # Only return fixed/safe fields (never raw exception text) + return jsonify( + { + "success": bool(result.get("success")), + "message": result.get("message") if result.get("success") else (result.get("message") or "LDAP connection failed"), + "user_count": result.get("user_count"), + } + ) @admin_bp.route("/admin/peppol/setup-wizard", methods=["GET"]) @@ -1891,40 +2097,44 @@ def admin_peppol_setup_wizard_check_bridge(): try: bridge_base_url = validate_outbound_url(bridge_base_url).rstrip("/") except UnsafeOutboundURL as exc: - return jsonify({"ok": False, "error": f"Invalid bridge URL: {exc}"}), 400 + return client_error_response("Invalid bridge URL", 400) headers = {"Accept": "application/json"} if bridge_token: headers["Authorization"] = f"Bearer {bridge_token}" timeout_s = 5 + from urllib.parse import urljoin + + health_url = urljoin(bridge_base_url + "/", "health") + test_url = urljoin(bridge_base_url + "/", "test") try: - health = requests.get(f"{bridge_base_url}/health", headers=headers, timeout=timeout_s) + health = requests.get(health_url, headers=headers, timeout=timeout_s) if health.status_code >= 400: return ( - jsonify({"ok": False, "error": f"Bridge /health returned HTTP {health.status_code}: {health.text}"}), + jsonify({"ok": False, "error": f"Bridge /health returned HTTP {health.status_code}"}), 400, ) health_json = health.json() - except Exception as e: - return jsonify({"ok": False, "error": f"Failed to reach bridge /health: {e}"}), 400 + except Exception: + return client_error_response("Failed to reach bridge /health", 400) try: - test = requests.post(f"{bridge_base_url}/test", headers=headers, timeout=timeout_s) + test = requests.post(test_url, headers=headers, timeout=timeout_s) if test.status_code >= 400: return ( jsonify( { "ok": False, - "error": f"Bridge /test returned HTTP {test.status_code}: {test.text}", + "error": f"Bridge /test returned HTTP {test.status_code}", "health": health_json, } ), 400, ) test_json = test.json() - except Exception as e: - return jsonify({"ok": False, "error": f"Failed to call bridge /test: {e}", "health": health_json}), 400 + except Exception: + return jsonify({"ok": False, **client_error_payload("Failed to call bridge /test"), "health": health_json}), 400 return jsonify({"ok": True, "health": health_json, "test": test_json}) @@ -1996,11 +2206,16 @@ def admin_peppol_setup_wizard_send_test(): svc = PeppolService() success, tx, message = svc.send_invoice(invoice=invoice, triggered_by_user_id=current_user.id) if success: - return jsonify({"ok": True, "message": message, "peppol_tx_id": tx.id if tx else None}) - return jsonify({"ok": False, "error": message, "peppol_tx_id": tx.id if tx else None}), 400 - except Exception as e: + return jsonify( + {"ok": True, "message": "Invoice sent via Peppol", "peppol_tx_id": tx.id if tx else None} + ) + current_app.logger.warning("Peppol wizard test send failed: %s", message) + return jsonify( + {"ok": False, "error": "Failed to send invoice via Peppol", "peppol_tx_id": tx.id if tx else None} + ), 400 + except Exception: current_app.logger.exception("Peppol wizard test send failed") - return jsonify({"ok": False, "error": f"Failed to send invoice via Peppol: {e}"}), 500 + return client_error_response("Failed to send invoice via Peppol", 500) @admin_bp.route("/admin/settings/verify-donate-hide-code", methods=["POST"]) @@ -2803,12 +3018,12 @@ def quote_pdf_layout_import_json(): return redirect(url_for("admin.quote_pdf_layout", size=page_size)) - except json.JSONDecodeError as e: - flash(_("Invalid JSON file: %(error)s", error=str(e)), "error") + except json.JSONDecodeError: + flash(_("Invalid JSON file."), "error") return redirect(url_for("admin.quote_pdf_layout", size=page_size)) - except Exception as e: - current_app.logger.error(f"Error importing quote PDF template JSON: {e}", exc_info=True) - flash(_("Error importing template: %(error)s", error=str(e)), "error") + except Exception: + current_app.logger.error("Error importing quote PDF template JSON", exc_info=True) + flash(_("Error importing template. Please try again."), "error") return redirect(url_for("admin.quote_pdf_layout", size=page_size)) @@ -2914,12 +3129,12 @@ def pdf_layout_import_json(): return redirect(url_for("admin.pdf_layout", size=page_size)) - except json.JSONDecodeError as e: - flash(_("Invalid JSON file: %(error)s", error=str(e)), "error") + except json.JSONDecodeError: + flash(_("Invalid JSON file."), "error") return redirect(url_for("admin.pdf_layout", size=page_size)) - except Exception as e: - current_app.logger.error(f"Error importing invoice PDF template JSON: {e}", exc_info=True) - flash(_("Error importing template: %(error)s", error=str(e)), "error") + except Exception: + current_app.logger.error("Error importing invoice PDF template JSON", exc_info=True) + flash(_("Error importing template. Please try again."), "error") return redirect(url_for("admin.pdf_layout", size=page_size)) @@ -3380,20 +3595,18 @@ def _preview_format_money(value): # If editor provided CSS, merge it (editor CSS takes precedence for @page rules) if css and css.strip(): - # Extract @page rule from editor CSS if present - editor_page_match = re.search(r"@page\s*\{[^}]*\}", css, re.IGNORECASE | re.DOTALL) - if editor_page_match: + # Extract @page rule from editor CSS if present (brace-aware, no ReDoS regex) + from app.utils.pdf_generator import find_at_page_rules, remove_at_page_rules + + editor_page_rules = find_at_page_rules(css) + if editor_page_rules: # Editor has @page rule - normalize it and use it, merge with saved CSS - editor_page_rule = editor_page_match.group(0) - # Normalize editor's @page rule to correct size FIRST - editor_page_rule = update_page_size_in_css(editor_page_rule, page_size) + original_editor_page_rule = editor_page_rules[0] + editor_page_rule = update_page_size_in_css(original_editor_page_rule, page_size) # Remove @page from saved CSS and add normalized editor's @page rule - if saved_css: - saved_css_no_page = re.sub(r"@page\s*\{[^}]*\}", "", saved_css, flags=re.IGNORECASE | re.DOTALL) - else: - saved_css_no_page = "" + saved_css_no_page = remove_at_page_rules(saved_css) if saved_css else "" # Remove @page rule from editor CSS and merge - editor_css_no_page = css.replace(editor_page_rule, "").strip() + editor_css_no_page = css.replace(original_editor_page_rule, "", 1).strip() css = editor_page_rule + "\n" + saved_css_no_page if editor_css_no_page: css = css + "\n" + editor_css_no_page @@ -3420,7 +3633,6 @@ def _preview_format_money(value): # Log if normalization changed anything if css != css_before: current_app.logger.debug(f"PDF Preview - CSS @page size normalized from template/editor to {page_size}") - # Validate after normalization is_valid, found_sizes = validate_page_size_in_css(css, page_size) if not is_valid: @@ -3458,49 +3670,22 @@ def remove_page_rule_from_html(html_text): """Remove @page rules from HTML inline styles to avoid conflicts with separate CSS""" import re - def remove_from_style_tag(match): - style_content = match.group(2) - # Remove @page rule from style content - # Need to handle nested @bottom-center rules properly - # Match @page { ... } including any nested rules - brace_count = 0 - page_pattern = r"@page\s*\{" - page_match = re.search(page_pattern, style_content, re.IGNORECASE) - - if page_match: - start = page_match.start() - # Find matching closing brace - end = len(style_content) - for i in range(page_match.end() - 1, len(style_content)): - if style_content[i] == "{": - brace_count += 1 - elif style_content[i] == "}": - brace_count -= 1 - if brace_count == 0: - end = i + 1 - break - # Remove the @page rule - style_content = style_content[:start] + style_content[end:] - # Clean up any double newlines or extra whitespace - style_content = re.sub(r"\n\s*\n", "\n", style_content) - - return f"{match.group(1)}{style_content}{match.group(3)}" - - # Match )" - if re.search(style_pattern, html_text, re.IGNORECASE | re.DOTALL): - html_text = re.sub(style_pattern, remove_from_style_tag, html_text, flags=re.IGNORECASE | re.DOTALL) - - return html_text + from app.utils.pdf_generator import _transform_style_tag_contents, remove_at_page_rules + + def remove_from_style_content(style_content): + style_content = remove_at_page_rules(style_content) + return re.sub(r"\n\s*\n", "\n", style_content) + + return _transform_style_tag_contents(html_text, remove_from_style_content) # Apply @page rule removal fix: if we have separate CSS and HTML with inline styles, # remove @page rules from HTML to ensure the separate CSS @page rule is used html_has_inline_styles = html and ")" - if re.search(style_pattern, html_text, re.IGNORECASE | re.DOTALL): - html_text = re.sub(style_pattern, remove_from_style_tag, html_text, flags=re.IGNORECASE | re.DOTALL) - - return html_text + from app.utils.pdf_generator import _transform_style_tag_contents, remove_at_page_rules + + def remove_from_style_content(style_content): + style_content = remove_at_page_rules(style_content) + return re.sub(r"\n\s*\n", "\n", style_content) + + return _transform_style_tag_contents(html_text, remove_from_style_content) # Apply @page rule removal fix: if we have separate CSS and HTML with inline styles, # remove @page rules from HTML to ensure the separate CSS @page rule is used html_has_inline_styles = html and "`` blocks via string search (no DOTALL regex) and + replace each style body with ``transform_fn(style_content)``. + """ + if not html_text: + return html_text + lower = html_text.lower() + result = [] + pos = 0 + while True: + start = lower.find("", start) + if tag_end == -1: + result.append(html_text[pos:]) + break + close = lower.find("", tag_end + 1) + if close == -1: + result.append(html_text[pos:]) + break + open_tag = html_text[start : tag_end + 1] + style_content = html_text[tag_end + 1 : close] + close_tag = html_text[close : close + len("")] + result.append(html_text[pos:start]) + result.append(open_tag) + result.append(transform_fn(style_content)) + result.append(close_tag) + pos = close + len("") + return "".join(result) + + def update_page_size_in_css(css_text, page_size): """ Update @page size property to match the specified page size. @@ -60,31 +142,16 @@ def update_page_size_in_css(css_text, page_size): if not css_text or not page_size: return css_text - # Find all @page rules (may have multiple) - page_pattern = r"@page\s*\{" - matches = list(re.finditer(page_pattern, css_text, re.IGNORECASE | re.MULTILINE)) + # Find all @page rules (may have multiple) via brace-aware walk + spans = list(_iter_at_page_spans(css_text)) - if not matches: + if not spans: # No @page rule exists - add one at the beginning new_page_rule = f"@page {{\n size: {page_size};\n margin: 2cm;\n }}\n\n" return new_page_rule + css_text # Process matches in reverse order to maintain positions - for match in reversed(matches): - start_pos = match.start() - # Find matching closing brace, accounting for nested braces - brace_count = 0 - end_pos = len(css_text) - - for i in range(match.end() - 1, len(css_text)): - if css_text[i] == "{": - brace_count += 1 - elif css_text[i] == "}": - brace_count -= 1 - if brace_count == 0: - end_pos = i + 1 - break - + for start_pos, end_pos in reversed(spans): page_block = css_text[start_pos:end_pos] # Replace or add size property @@ -206,8 +273,8 @@ def validate_page_size_in_css(css_text, expected_page_size): if not css_text or not expected_page_size: return False, [] - # Find all @page rules and check their size - page_rules = re.findall(r"@page\s*\{[^}]*\}", css_text, re.IGNORECASE | re.DOTALL) + # Find all @page rules via brace-aware walk (avoids polynomial ReDoS) + page_rules = find_at_page_rules(css_text) found_sizes = [] for rule in page_rules: @@ -761,60 +828,21 @@ def debug_print(msg): def update_page_size_in_html(html_text): """Update @page size property in HTML's inline )" - if re.search(style_pattern, html_text, re.IGNORECASE | re.DOTALL): - html_text = re.sub(style_pattern, update_style_tag, html_text, flags=re.IGNORECASE | re.DOTALL) + def update_style_content(style_content): + return update_page_size_in_css(style_content, self.page_size) - return html_text + return _transform_style_tag_contents(html_text, update_style_content) def remove_page_rule_from_html(html_text): """Remove @page rules from HTML inline styles to avoid conflicts with separate CSS""" import re - def remove_from_style_tag(match): - style_content = match.group(2) - # Remove @page rule from style content - # Need to handle nested @bottom-center rules properly - # Match @page { ... } including any nested rules - brace_count = 0 - page_pattern = r"@page\s*\{" - page_match = re.search(page_pattern, style_content, re.IGNORECASE) - - if page_match: - start = page_match.start() - # Find matching closing brace - pos = page_match.end() - 1 - end = len(style_content) - for i in range(page_match.end() - 1, len(style_content)): - if style_content[i] == "{": - brace_count += 1 - elif style_content[i] == "}": - brace_count -= 1 - if brace_count == 0: - end = i + 1 - break - # Remove the @page rule - style_content = style_content[:start] + style_content[end:] - # Clean up any double newlines or extra whitespace - style_content = re.sub(r"\n\s*\n", "\n", style_content) - - return f"{match.group(1)}{style_content}{match.group(3)}" - - # Match )" - if re.search(style_pattern, html_text, re.IGNORECASE | re.DOTALL): - html_text = re.sub(style_pattern, remove_from_style_tag, html_text, flags=re.IGNORECASE | re.DOTALL) - - return html_text + def remove_from_style_content(style_content): + style_content = remove_at_page_rules(style_content) + return re.sub(r"\n\s*\n", "\n", style_content) + + return _transform_style_tag_contents(html_text, remove_from_style_content) # Handle CSS: When both HTML (with inline styles) and separate CSS exist, # extract inline styles, merge with separate CSS, and remove from HTML to avoid conflicts @@ -823,12 +851,18 @@ def remove_from_style_tag(match): css_to_use = "" html_inline_styles_extracted = False - # Extract inline styles from HTML if present + # Extract inline styles from HTML if present (string find, no DOTALL regex) extracted_inline_css = "" - if html_template and "", html_template, re.IGNORECASE | re.DOTALL) - if style_match: - extracted_inline_css = style_match.group(1) + if html_template and ")" - if re.search(style_pattern, html_text, re.IGNORECASE | re.DOTALL): - html_text = re.sub(style_pattern, remove_from_style_tag, html_text, flags=re.IGNORECASE | re.DOTALL) - - return html_text + def remove_from_style_content(style_content): + style_content = remove_at_page_rules(style_content) + return re.sub(r"\n\s*\n", "\n", style_content) + + return _transform_style_tag_contents(html_text, remove_from_style_content) import re css_to_use = "" html_inline_styles_extracted = False - # Extract inline styles from HTML if present + # Extract inline styles from HTML if present (string find, no DOTALL regex) extracted_inline_css = "" - if html_template and "", html_template, re.IGNORECASE | re.DOTALL) - if style_match: - extracted_inline_css = style_match.group(1) + if html_template and "