From 5352f45498aa7688455425a1780b34360a005f8d Mon Sep 17 00:00:00 2001 From: Diego Ferrand Date: Fri, 10 Jul 2026 16:06:58 -0300 Subject: [PATCH 1/7] Upgrade ubuntu image to use latest lts (#34) --- .github/workflows/build.yaml | 8 ++++---- Dockerfile | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 1f329e8..128a9bc 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -36,19 +36,19 @@ jobs: steps: - uses: actions/checkout@v6 - - name: Build binary (Linux in Ubuntu 25.10 container) + - name: Build binary (Linux in Ubuntu 26.04 container) if: matrix.name == 'linux' shell: bash run: | - # Docker container with Ubuntu 25.10 and build the binary inside it + # Docker container with Ubuntu 26.04 and build the binary inside it rm -rf dist/* docker run --rm \ -v "$PWD:/workspace" \ -w /workspace \ - ubuntu:25.10 \ + ubuntu:26.04 \ bash -c " set -e - echo 'Setting up Ubuntu 25.10 build environment...' + echo 'Setting up Ubuntu 26.04 build environment...' apt-get update -y apt-get install -y --fix-missing curl python3 python3-pip python3-venv binutils diff --git a/Dockerfile b/Dockerfile index db66dd8..94313c0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -FROM ubuntu:25.10 +FROM ubuntu:26.04 WORKDIR /app From cc814cd1f2e2d7c33ca9f300fefb1b68ca1d07f3 Mon Sep 17 00:00:00 2001 From: Diego Ferrand Date: Fri, 17 Jul 2026 15:43:56 -0300 Subject: [PATCH 2/7] Add Perfecto MCP version info and GitHub update-check tools (#37) --- config/perfecto.py | 7 +- server.py | 2 + tests/test_tools_manager.py | 211 +++++++++++++++++++++ tools/tools_manager.py | 366 ++++++++++++++++++++++++++++++++++++ 4 files changed, 583 insertions(+), 3 deletions(-) create mode 100644 tests/test_tools_manager.py create mode 100644 tools/tools_manager.py diff --git a/config/perfecto.py b/config/perfecto.py index 1cbde76..1001cef 100644 --- a/config/perfecto.py +++ b/config/perfecto.py @@ -1,7 +1,8 @@ TOOLS_PREFIX: str = "perfecto" -WEBSITE: str = "https://github.com/PerfectoCore/perfecto-mcp/" -GITHUB: str = "https://github.com/PerfectoCore/perfecto-mcp" -SUPPORT_MESSAGE: str = "If you think this is a bug, please contact Perfecto support or report issue at https://github.com/PerfectoCore/perfecto-mcp/issues" +WEBSITE: str = "https://github.com/PerfectoCode/perfecto-mcp/" +GITHUB: str = "https://github.com/PerfectoCode/perfecto-mcp" +GITHUB_API_LATEST_RELEASE: str = "https://api.github.com/repos/PerfectoCode/perfecto-mcp/releases/latest" +SUPPORT_MESSAGE: str = "If you think this is a bug, please contact Perfecto support or report issue at https://github.com/PerfectoCode/perfecto-mcp/issues" SECURITY_TOKEN_FILE_ENV_NAME: str = "PERFECTO_SECURITY_TOKEN_FILE" SECURITY_TOKEN_ENV_NAME: str = "PERFECTO_SECURITY_TOKEN" diff --git a/server.py b/server.py index c85535a..1600788 100644 --- a/server.py +++ b/server.py @@ -5,6 +5,7 @@ from tools.device_manager import register as register_device_manager from tools.execution_manager import register as register_execution_manager from tools.help_manager import register as register_help_manager +from tools.tools_manager import register as register_tools_manager from tools.user_manager import register as register_user_manager @@ -21,3 +22,4 @@ def register_tools(mcp, token: Optional[PerfectoToken]): register_execution_manager(mcp, token) register_help_manager(mcp, token) register_ai_scriptless_manager(mcp, token) + register_tools_manager(mcp, token) diff --git a/tests/test_tools_manager.py b/tests/test_tools_manager.py new file mode 100644 index 0000000..99cf12a --- /dev/null +++ b/tests/test_tools_manager.py @@ -0,0 +1,211 @@ +""" +Copyright 2025 Perforce Software, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx + +from tools.tools_manager import ( + ToolsManager, + _match_recommended_asset, + _normalize_arch, + _normalize_system, +) + + +def _make_ctx(): + return MagicMock() + + +def test_version_returns_current_build_metadata(perfecto_token): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.version()) + + assert result.error is None + assert len(result.result) == 1 + payload = result.result[0] + assert payload["version"] + assert payload["user_agent"].startswith(f"perfecto-mcp/{payload['version']}") + assert "platform" in payload + assert "runtime" in payload + assert result.info + assert any("check_updates" in message for message in result.info) + + +def test_normalize_platform_helpers(): + assert _normalize_system("Darwin") == "macos" + assert _normalize_system("Windows") == "windows" + assert _normalize_system("Linux") == "linux" + assert _normalize_arch("x86_64") == "amd64" + assert _normalize_arch("amd64") == "amd64" + assert _normalize_arch("arm64") == "arm64" + assert _normalize_arch("aarch64") == "arm64" + + +def test_match_recommended_asset_prefers_zip(): + assets = [ + { + "name": "perfecto-mcp-macos-arm64", + "browser_download_url": "https://example.com/bin", + }, + { + "name": "perfecto-mcp-macos-arm64.zip", + "browser_download_url": "https://example.com/zip", + }, + ] + matched = _match_recommended_asset(assets, "macos", "arm64") + assert matched["name"] == "perfecto-mcp-macos-arm64.zip" + + +def test_check_updates_when_latest_is_newer(perfecto_token): + release = { + "tag_name": "v9.9.9", + "name": "v9.9.9", + "html_url": "https://github.com/PerfectoCode/perfecto-mcp/releases/tag/v9.9.9", + "published_at": "2026-07-14T00:00:00Z", + "body": "Release notes for 9.9.9", + "assets": [ + { + "name": "perfecto-mcp-macos-arm64.zip", + "browser_download_url": "https://example.com/perfecto-mcp-macos-arm64.zip", + "size": 123, + "content_type": "application/zip", + "updated_at": "2026-07-14T00:00:00Z", + } + ], + } + + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock() + mock_response.json = MagicMock(return_value=release) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \ + patch("tools.tools_manager.__version__", "1.0.0"), \ + patch("tools.tools_manager.platform.system", return_value="Darwin"), \ + patch("tools.tools_manager.platform.machine", return_value="arm64"): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.check_updates()) + + assert result.error is None + payload = result.result[0] + assert payload["update_available"] is True + assert payload["current_version"] == "1.0.0" + assert payload["latest_version"] == "9.9.9" + assert payload["release"]["body"] == "Release notes for 9.9.9" + assert payload["recommended_asset"]["name"] == "perfecto-mcp-macos-arm64.zip" + assert payload["update_guidance"]["status"] == "update_available" + + +def test_check_updates_when_up_to_date(perfecto_token): + release = { + "tag_name": "v1.1.1", + "name": "v1.1.1", + "html_url": "https://github.com/PerfectoCode/perfecto-mcp/releases/tag/v1.1.1", + "published_at": "2026-07-09T00:00:00Z", + "body": "Up to date notes", + "assets": [], + } + + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock() + mock_response.json = MagicMock(return_value=release) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \ + patch("tools.tools_manager.__version__", "1.1.1"): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.check_updates()) + + assert result.error is None + payload = result.result[0] + assert payload["update_available"] is False + assert payload["update_guidance"]["status"] == "up_to_date" + + +def test_check_updates_http_error(perfecto_token): + mock_response = MagicMock() + mock_response.status_code = 500 + request = httpx.Request("GET", "https://api.github.com") + mock_response.raise_for_status = MagicMock( + side_effect=httpx.HTTPStatusError("boom", request=request, response=mock_response) + ) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.check_updates()) + + assert result.error is None + payload = result.result[0] + assert payload["update_check_status"] == "unavailable" + assert payload["update_available"] is None + assert payload["latest_version"] is None + assert "releases" in payload["releases_url"] + assert result.warning + assert any("Could not reach GitHub" in message for message in result.warning) + assert result.info + assert any("running Perfecto MCP version" in message for message in result.info) + + +def test_check_updates_connect_error(perfecto_token): + request = httpx.Request("GET", "https://api.github.com") + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("dns failed", request=request)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \ + patch("tools.tools_manager.__version__", "1.1.1"): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.check_updates()) + + assert result.error is None + payload = result.result[0] + assert payload["update_check_status"] == "unavailable" + assert payload["current_version"] == "1.1.1" + assert "connect" in payload["reason"].lower() or "blocked" in payload["reason"].lower() + assert payload["update_guidance"]["status"] == "unavailable" + + +def test_check_updates_timeout(perfecto_token): + request = httpx.Request("GET", "https://api.github.com") + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=httpx.ReadTimeout("timed out", request=request)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client): + manager = ToolsManager(perfecto_token, _make_ctx()) + result = asyncio.run(manager.check_updates()) + + assert result.error is None + payload = result.result[0] + assert payload["update_check_status"] == "unavailable" + assert "Timed out" in payload["reason"] + assert any("corporate" in message for message in (result.warning or [])) diff --git a/tools/tools_manager.py b/tools/tools_manager.py new file mode 100644 index 0000000..6bb7ced --- /dev/null +++ b/tools/tools_manager.py @@ -0,0 +1,366 @@ +import os +import platform +import sys +import traceback +from typing import Any, Dict, List, Optional + +import httpx +from mcp.server.fastmcp import Context +from packaging.version import InvalidVersion, Version +from pydantic import Field + +from config.perfecto import ( + GITHUB, + GITHUB_API_LATEST_RELEASE, + SUPPORT_MESSAGE, + TOOLS_PREFIX, + WEBSITE, +) +from config.token import PerfectoToken +from config.version import __bundle__, __executable__, __uvx__, __version__ +from models.manager import Manager +from models.result import BaseResult +from telemetry import run_tool +from tools.utils import timeout, user_agent + + +def _normalize_system(system: str) -> str: + system = system.lower() + if system == "darwin": + return "macos" + if system == "windows": + return "windows" + return "linux" + + +def _normalize_arch(machine: str) -> str: + machine = machine.lower() + if machine in {"x86_64", "amd64"}: + return "amd64" + if machine in {"aarch64", "arm64"} or machine.startswith("arm"): + return "arm64" + return machine + + +def _parse_version(value: str) -> Optional[Version]: + try: + return Version(str(value).lstrip("vV")) + except InvalidVersion: + return None + + +def _detect_runtime() -> Dict[str, Any]: + return { + "frozen": bool(getattr(sys, "frozen", False)), + "uvx": bool(__uvx__), + "docker": os.getenv("MCP_DOCKER", "false").lower() == "true", + "executable": __executable__, + "bundle": __bundle__, + } + + +def _platform_info() -> Dict[str, str]: + return { + "system": platform.system(), + "release": platform.release(), + "machine": platform.machine(), + "normalized_system": _normalize_system(platform.system()), + "normalized_arch": _normalize_arch(platform.machine()), + } + + +def _match_recommended_asset(assets: List[Dict[str, Any]], system: str, arch: str) -> Optional[Dict[str, Any]]: + prefix = f"perfecto-mcp-{system}-{arch}" + exact = [ + asset for asset in assets + if str(asset.get("name", "")).startswith(prefix) + ] + if exact: + # Prefer zip packages when multiple assets share the same platform prefix. + zip_assets = [asset for asset in exact if str(asset.get("name", "")).endswith(".zip")] + return zip_assets[0] if zip_assets else exact[0] + return None + + +def _releases_url() -> str: + return f"{GITHUB}/releases" + + +def _update_guidance(runtime: Dict[str, Any], update_available: bool, recommended_asset: Optional[Dict[str, Any]]) -> Dict[str, str]: + releases_url = _releases_url() + manual = ( + f"Download the package for your platform from {releases_url}, " + "replace the current MCP executable with the new one, then restart the MCP client." + ) + if recommended_asset and recommended_asset.get("browser_download_url"): + manual = ( + f"Download [{recommended_asset['name']}]({recommended_asset['browser_download_url']}), " + "replace the current MCP executable with the new one, then restart the MCP client." + ) + + if runtime.get("docker"): + automatic = ( + "Pull the latest Docker image (`ghcr.io/perfectocode/perfecto-mcp:latest`) " + "and restart the MCP container/client." + ) + elif runtime.get("uvx"): + automatic = ( + f"Update the MCP client config git ref to the latest release tag " + f"(for example `git+{GITHUB}.git@v`) and restart the MCP client." + ) + else: + automatic = ( + "Automatic in-place update is not available yet. " + "Use the manual download path above, or ask the AI to guide a controlled manual replace." + ) + + if not update_available: + return { + "status": "up_to_date", + "manual": f"No update required. Releases are listed at {releases_url}.", + "automatic": "No update required.", + } + + return { + "status": "update_available", + "manual": manual, + "automatic": automatic, + } + + +def _github_unavailable_result(reason: str) -> BaseResult: + """Friendly response when GitHub cannot be reached to check for updates.""" + releases_url = _releases_url() + return BaseResult( + result=[{ + "current_version": __version__, + "latest_version": None, + "update_available": None, + "update_check_status": "unavailable", + "reason": reason, + "releases_url": releases_url, + "platform": _platform_info(), + "runtime": _detect_runtime(), + "update_guidance": { + "status": "unavailable", + "manual": ( + f"Update check could not reach GitHub. " + f"When you have access, open [{releases_url}]({releases_url}) " + f"and compare the latest release with your current version ({__version__})." + ), + "automatic": ( + "Automatic update check is unavailable while GitHub cannot be reached. " + "Retry later or check releases from a network that can access GitHub." + ), + }, + }], + warning=[ + "Could not reach GitHub to check for Perfecto MCP updates. " + "This is common on restricted/corporate networks or when github.com is blocked." + ], + info=[ + f"You are running Perfecto MCP version {__version__}.", + f"Check for newer releases manually at {releases_url} when GitHub is reachable.", + ], + ) + + +def _github_access_failure_result(exc: Exception) -> BaseResult: + if isinstance(exc, httpx.TimeoutException): + return _github_unavailable_result( + "Timed out while contacting GitHub. The network may be slow, filtered, or offline." + ) + if isinstance(exc, (httpx.ConnectError, httpx.NetworkError, httpx.ProxyError)): + return _github_unavailable_result( + "Could not connect to GitHub. The host may be offline, firewalled, or blocked from api.github.com." + ) + if isinstance(exc, httpx.HTTPStatusError): + status = exc.response.status_code + if status in {401, 403}: + return _github_unavailable_result( + f"GitHub refused the update check (HTTP {status}). " + "Access may be blocked, require authentication, or be rate-limited." + ) + if status == 404: + return _github_unavailable_result( + "GitHub release endpoint was not found (HTTP 404). " + "The repository or releases URL may be unavailable from this network." + ) + return _github_unavailable_result( + f"GitHub returned an unexpected status while checking for updates (HTTP {status})." + ) + if isinstance(exc, httpx.HTTPError): + return _github_unavailable_result( + "Could not complete the GitHub update check due to a network or HTTP error." + ) + return _github_unavailable_result( + "Could not complete the GitHub update check." + ) + + +class ToolsManager(Manager): + def __init__(self, token: Optional[PerfectoToken], ctx: Context): + super().__init__(token, ctx) + + async def version(self) -> BaseResult: + platform_data = _platform_info() + runtime = _detect_runtime() + return BaseResult( + result=[{ + "version": __version__, + "user_agent": user_agent, + "platform": platform_data, + "runtime": runtime, + "repository": GITHUB, + "website": WEBSITE, + }], + info=[ + f"Perfecto MCP version {__version__}.", + "Use action `check_updates` to compare against the latest GitHub release.", + ], + ) + + async def check_updates(self) -> BaseResult: + headers = { + "User-Agent": user_agent, + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + + async with httpx.AsyncClient(timeout=timeout) as client: + try: + resp = await client.get(GITHUB_API_LATEST_RELEASE, headers=headers) + resp.raise_for_status() + release = resp.json() + except httpx.HTTPError as exc: + return _github_access_failure_result(exc) + except Exception as exc: + return _github_access_failure_result(exc) + + tag_name = str(release.get("tag_name") or "") + latest_version = tag_name.lstrip("vV") or str(release.get("name") or "") + current = _parse_version(__version__) + latest = _parse_version(latest_version) + + if current is None or latest is None: + return BaseResult( + error=( + f"Unable to compare versions. " + f"current={__version__!r}, latest={latest_version!r}." + ) + ) + + update_available = latest > current + assets = [ + { + "name": asset.get("name"), + "browser_download_url": asset.get("browser_download_url"), + "size": asset.get("size"), + "content_type": asset.get("content_type"), + "updated_at": asset.get("updated_at"), + } + for asset in (release.get("assets") or []) + ] + + platform_data = _platform_info() + runtime = _detect_runtime() + recommended_asset = _match_recommended_asset( + assets, + platform_data["normalized_system"], + platform_data["normalized_arch"], + ) + guidance = _update_guidance(runtime, update_available, recommended_asset) + + info = [] + if update_available: + info.append( + f"Update available: current {__version__} -> latest {latest_version}." + ) + info.append( + "Share the release notes with the user and ask whether they want a manual " + "update now or guidance for an automatic update path." + ) + if recommended_asset: + info.append( + f"Recommended download for this host: {recommended_asset.get('name')}." + ) + else: + info.append( + "No exact platform asset match was found; show the full assets list " + "so the user can pick the correct package." + ) + else: + info.append(f"Perfecto MCP is up to date (version {__version__}).") + + return BaseResult( + result=[{ + "current_version": __version__, + "latest_version": latest_version, + "update_available": update_available, + "release": { + "tag_name": tag_name, + "name": release.get("name"), + "html_url": release.get("html_url"), + "published_at": release.get("published_at"), + "body": release.get("body") or "", + }, + "recommended_asset": recommended_asset, + "assets": assets, + "platform": platform_data, + "runtime": runtime, + "update_guidance": guidance, + }], + info=info, + ) + + +def register(mcp, token: Optional[PerfectoToken]): + @mcp.tool( + name=f"{TOOLS_PREFIX}_tools", + description=""" +Operations on Perfecto MCP tooling metadata (versioning and updates). +Actions: +- version: Return the current MCP version and runtime/platform information used by the + user-agent and `--version` / console display. +- check_updates: Query the GitHub repository for the latest release, compare it with the + current version, and return release notes plus download links when an update is available. +Hints: +- Prefer `version` first when the user asks what build they are running. +- Prefer `check_updates` when the user asks whether a newer MCP release exists. +- When an update is available, present release notes and ask before replacing the executable. +- If GitHub is unreachable (restricted network, firewall, timeout), explain that the update + check is unavailable, still share the current version, and point to the releases page for a + manual check — do not treat it as a hard failure. +- Render release and download URLs as markdown links. +""" + ) + async def tools( + action: str = Field(description="The action id to execute"), + args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None), + ctx: Context = Field(description="Context object providing access to MCP capabilities") + ) -> BaseResult: + if args is None: + args = {} + tools_manager = ToolsManager(token, ctx) + + async def _dispatch(): + match action: + case "version": + return await tools_manager.version() + case "check_updates": + return await tools_manager.check_updates() + case _: + return BaseResult( + error=f"Action {action} not found in tools manager tool" + ) + + try: + return await run_tool(f"{TOOLS_PREFIX}_tools", action, ctx, _dispatch) + except httpx.HTTPStatusError: + return BaseResult( + error=f"Error: {traceback.format_exc()}" + ) + except Exception: + return BaseResult( + error=f"Error: {traceback.format_exc()}\n{SUPPORT_MESSAGE}" + ) From b1ee6c41852495dfba4c0944c92513b818ba190c Mon Sep 17 00:00:00 2001 From: Diego Ferrand Date: Fri, 17 Jul 2026 15:44:22 -0300 Subject: [PATCH 3/7] Add HTTP allowlist, traceback sanitization, and help HTML hardening with regression tests (#38) --- config/security.py | 140 +++++++++++++ config/token.py | 21 +- tests/test_datetime_utils.py | 14 ++ tests/test_help_utils_href_interpolation.py | 104 ++++++++++ tests/test_http_request_security.py | 99 +++++++++ tests/test_token_security.py | 33 +++ tests/test_traceback_sanitization.py | 213 ++++++++++++++++++++ tools/ai_scriptless_manager.py | 7 +- tools/device_manager.py | 7 +- tools/execution_manager.py | 7 +- tools/help_manager.py | 45 ++++- tools/help_utils.py | 11 +- tools/user_manager.py | 7 +- tools/utils.py | 82 +++++++- 14 files changed, 747 insertions(+), 43 deletions(-) create mode 100644 config/security.py create mode 100644 tests/test_datetime_utils.py create mode 100644 tests/test_help_utils_href_interpolation.py create mode 100644 tests/test_http_request_security.py create mode 100644 tests/test_token_security.py create mode 100644 tests/test_traceback_sanitization.py diff --git a/config/security.py b/config/security.py new file mode 100644 index 0000000..cd6734b --- /dev/null +++ b/config/security.py @@ -0,0 +1,140 @@ +"""Security helpers for HTTP endpoint allowlisting and sensitive path detection.""" + +import re +from pathlib import Path +from typing import Optional +from urllib.parse import urlparse + + +SENSITIVE_PATH_PREFIXES = ( + "/etc/", + "/proc/", + "/sys/", + "/dev/", + "/boot/", + "/root/", + "/run/secrets/", + "/var/run/", + "/var/db/", + "/var/root/", + "/var/log/", + "/var/spool/", + "/private/etc/", + # macOS: /var symlinks to /private/var. Block sensitive subdirs; allow /private/var/folders/ + "/private/var/run/", + "/private/var/db/", + "/private/var/root/", + "/private/var/log/", + "/private/var/spool/", + "/system/", + "/library/keychains/", + "/windows/", + "/program files/", + "/program files (x86)/", + "/programdata/", +) + +SENSITIVE_PATH_CONTAINS = ( + "/.ssh/", + "/.aws/", + "/.azure/", + "/.gnupg/", + "/.kube/", + "/.docker/", + "/.terraform/", + "/.pulumi/", + "/.config/gcloud/", + "/appdata/roaming/microsoft/credentials/", + "/appdata/roaming/gnupg/", + "/appdata/roaming/aws/", +) + +SENSITIVE_FILE_NAMES = { + ".env", + ".netrc", + ".git-credentials", + "kubeconfig", + ".npmrc", + ".pypirc", + "id_rsa", + "id_dsa", + "id_ecdsa", + "id_ed25519", + "known_hosts", + "authorized_keys", + "credentials", + "credentials.db", + "terraform.tfstate", + "terraform.tfstate.backup", +} + +SENSITIVE_FILE_EXTENSIONS = { + ".pem", + ".key", + ".p12", + ".pfx", + ".kdbx", + ".tfstate", + ".tfvars", + ".ovpn", +} + +# Domains allowed for unauthenticated http_request (help documentation fetches). +ALLOWED_HTTP_REQUEST_DOMAINS = ( + "help.perfecto.io", +) + + +def normalize_path_for_security(file_path: str) -> str: + return file_path.replace("\\", "/").strip().lower() + + +def normalize_windows_drive_prefix(file_path: str) -> str: + if re.match(r"^[a-z]:/", file_path): + return file_path[2:] + return file_path + + +def detect_sensitive_upload_path_reason(file_path: str) -> Optional[str]: + # Denylist of sensitive local paths/files. Available for future upload flows; + # http_request allowlisting is the active control in this project today. + normalized_path = normalize_path_for_security(file_path) + normalized_without_drive = normalize_windows_drive_prefix(normalized_path) + base_name = Path(normalized_path).name + + for prefix in SENSITIVE_PATH_PREFIXES: + if normalized_path.startswith(prefix) or normalized_without_drive.startswith(prefix): + return f"sensitive system path prefix '{prefix}'" + + for sensitive_fragment in SENSITIVE_PATH_CONTAINS: + if sensitive_fragment in normalized_path: + return f"sensitive path segment '{sensitive_fragment}'" + + if base_name in SENSITIVE_FILE_NAMES: + return f"sensitive file name '{base_name}'" + + for sensitive_extension in SENSITIVE_FILE_EXTENSIONS: + if base_name.endswith(sensitive_extension): + return f"sensitive file extension '{sensitive_extension}'" + + if base_name.startswith(".env."): + return "sensitive environment file pattern '.env.*'" + + return None + + +def validate_http_request_endpoint(endpoint: str) -> Optional[str]: + parsed_url = urlparse(endpoint) + + if parsed_url.scheme != "https": + return "Invalid endpoint scheme. Only https URLs are allowed." + + if not parsed_url.hostname: + return "Invalid endpoint URL. Absolute URL with hostname is required." + + host = parsed_url.hostname.lower() + if host not in ALLOWED_HTTP_REQUEST_DOMAINS: + allowed = ", ".join(ALLOWED_HTTP_REQUEST_DOMAINS) + return f"Endpoint host '{host}' is not allowed. Allowed hosts: {allowed}" + + return None diff --git a/config/token.py b/config/token.py index dc55521..624bc10 100644 --- a/config/token.py +++ b/config/token.py @@ -26,6 +26,11 @@ class PerfectoToken: __slots__ = ("token", "cloud_name") def __init__(self, token: str, cloud_name: str): + if not token or not isinstance(token, str): + raise PerfectoTokenError("Invalid security token format: expected non-empty string") + if cloud_name is not None and (not isinstance(cloud_name, str) or not cloud_name): + raise PerfectoTokenError("Invalid cloud name format: expected non-empty string") + self.token = token self.cloud_name = cloud_name @@ -34,20 +39,18 @@ def __init__(self, token: str, cloud_name: str): def from_file(cls, path: Union[str, Path], cloud_name: str) -> "PerfectoToken": p = Path(path) if not p.exists() or not p.is_file(): - raise PerfectoTokenError(f"directory or file does not exist: {p!r}") + raise PerfectoTokenError("Token file does not exist or is not a file") try: raw = p.read_text(encoding="utf-8") except Exception as e: - raise PerfectoTokenError(f"Error reading/parsing file at {p!r}: {e}") from e + raise PerfectoTokenError(f"Error reading token file: {type(e).__name__}") from e - try: - token_val = raw - cloud_name_val = cloud_name - except KeyError as e: - raise PerfectoTokenError(f"missing field {e.args[0]!r} at {p!r}") from e + token_val = raw.strip() + if not token_val: + raise PerfectoTokenError("Token file is empty") - return cls(token=token_val, cloud_name=cloud_name_val) + return cls(token=token_val, cloud_name=cloud_name) def __repr__(self): - return f"" + return "" diff --git a/tests/test_datetime_utils.py b/tests/test_datetime_utils.py new file mode 100644 index 0000000..600756b --- /dev/null +++ b/tests/test_datetime_utils.py @@ -0,0 +1,14 @@ +"""Datetime utility timezone tests.""" + +from tools.utils import get_date_time_iso + + +class TestDateTimeIsoTimezone: + def test_returns_none_for_none_timestamp(self): + assert get_date_time_iso(None) is None + + def test_returns_utc_timezone_for_unix_epoch(self): + assert get_date_time_iso(0) == "1970-01-01T00:00:00+00:00" + + def test_returns_utc_timezone_for_known_timestamp(self): + assert get_date_time_iso(1710000000) == "2024-03-09T16:00:00+00:00" diff --git a/tests/test_help_utils_href_interpolation.py b/tests/test_help_utils_href_interpolation.py new file mode 100644 index 0000000..994afee --- /dev/null +++ b/tests/test_help_utils_href_interpolation.py @@ -0,0 +1,104 @@ +"""Help utils href interpolation and HTML sanitization tests.""" + +import lxml.html + +from tools.help_utils import process_inline_elements, table_to_markdown, html_to_markdown + + +class TestHelpUtilsHrefInterpolation: + def test_process_inline_elements_interpolates_href_in_html_mode(self): + element = lxml.html.fromstring("

Go here

") + + rendered = process_inline_elements( + element, + base_url="https://help.perfecto.io", + as_html=True, + ) + + assert "here" in rendered + assert "{href}" not in rendered + + def test_table_to_markdown_interpolates_href_inside_html_table_cells(self): + table = lxml.html.fromstring( + "" + "" + "" + "
Doc
Guide
" + ) + + rendered = table_to_markdown( + table, + base_url="https://help.perfecto.io", + as_html=True, + ) + + assert "Guide" in rendered + assert "{href}" not in rendered + + def test_html_to_markdown_outputs_markdown_links_without_literal_template(self): + html = ( + "
" + "

Read Start

" + "
" + ) + rendered = html_to_markdown(html, base_url="https://help.perfecto.io") + + assert "[Start](https://help.perfecto.io/docs/start.html)" in rendered + assert "{href}" not in rendered + + +class TestHelpUtilsHtmlSanitization: + def test_javascript_href_case_insensitive_blocked(self): + element = lxml.html.fromstring("

click

") + + rendered = process_inline_elements(element, as_html=True) + + assert "JAVASCRIPT:" not in rendered + assert "alert" not in rendered + + def test_javascript_href_mixed_case_blocked(self): + element = lxml.html.fromstring("

click

") + + rendered = process_inline_elements(element, as_html=True) + + assert "JavaScript:" not in rendered + + def test_html_special_chars_escaped_in_link_text(self): + element = lxml.html.fromstring("

A & B

") + + rendered = process_inline_elements(element, as_html=True) + + assert "A & B" in rendered + assert "