Initial public upload from Projekte/Connector (part 16)

bf06a0d verified about 1 month ago

24.7 kB

import base64
import io
import json
import logging
import os
import zipfile
import ssl
import asyncio
import functools

from pathlib import Path
from typing import Annotated, Any, List, Optional

from pydantic import BeforeValidator
from defusedxml import ElementTree as ET

from googleapiclient.errors import HttpError
from .api_enablement import get_api_enablement_message
from auth.google_auth import GoogleAuthenticationError
from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider

logger = logging.getLogger(__name__)


class TransientNetworkError(Exception):
    """Custom exception for transient network errors after retries."""

    pass


class UserInputError(Exception):
    """Raised for user-facing input/validation errors that shouldn't be retried."""

    pass


def _coerce_json_str_to_type(v: Any, expected_type: type) -> Any:
    """Coerce a JSON-encoded string to a specific container type."""
    if not isinstance(v, str):
        return v

    try:
        parsed = json.loads(v)
    except (json.JSONDecodeError, TypeError):
        return v

    return parsed if isinstance(parsed, expected_type) else v


def _coerce_json_str_to_list(v: Any) -> Any:
    """Coerce a JSON-encoded string to a list.
    Some MCP clients (e.g. Cowork) serialise array parameters as JSON strings
    rather than native arrays.  This ``BeforeValidator`` transparently converts
    ``'["a","b"]'`` → ``["a", "b"]`` so Pydantic validation succeeds.
    """
    return _coerce_json_str_to_type(v, list)


StringList = Annotated[List[str], BeforeValidator(_coerce_json_str_to_list)]
"""``List[str]`` that also accepts a JSON-encoded string of an array.
Use in tool signatures instead of ``List[str]`` to work around MCP clients
that send ``'["value"]'`` instead of ``["value"]``.
"""


DictList = Annotated[List[dict[str, Any]], BeforeValidator(_coerce_json_str_to_list)]
"""``List[dict]`` that also accepts a JSON-encoded string of an array.
Use in tool signatures instead of ``List[dict]`` to work around MCP clients
that send ``'[{"key":"val"}]'`` instead of ``[{"key":"val"}]``.
"""


def _coerce_json_str_to_dict(v: Any) -> Any:
    """Coerce a JSON-encoded string to a dict.
    Some MCP clients serialise dict parameters as JSON strings rather than
    native objects.  This ``BeforeValidator`` transparently converts
    ``'{"key":"val"}'`` -> ``{"key": "val"}`` so Pydantic validation succeeds.
    """
    return _coerce_json_str_to_type(v, dict)


JsonDict = Annotated[dict[str, Any], BeforeValidator(_coerce_json_str_to_dict)]
"""``dict`` that also accepts a JSON-encoded string of an object.
Use in tool signatures instead of ``Dict[str, Any]`` to work around MCP clients
that send ``'{"key":"val"}'`` instead of ``{"key": "val"}``.
"""


# Directories from which local file reads are allowed.
# By default, only the managed attachment storage directory is trusted.
# Override via ALLOWED_FILE_DIRS env var (os.pathsep-separated paths).
_ALLOWED_FILE_DIRS_ENV = "ALLOWED_FILE_DIRS"


def _get_allowed_file_dirs() -> list[Path]:
    """Return the list of directories from which local file access is permitted."""
    from core.attachment_storage import STORAGE_DIR

    allowed_dirs: list[Path] = [STORAGE_DIR]
    env_val = os.environ.get(_ALLOWED_FILE_DIRS_ENV)
    if env_val:
        allowed_dirs.extend(
            Path(p_stripped).expanduser().resolve()
            for p in env_val.split(os.pathsep)
            if (p_stripped := p.strip())
        )

    unique_dirs: list[Path] = []
    seen: set[Path] = set()
    for path in allowed_dirs:
        if path in seen:
            continue
        seen.add(path)
        unique_dirs.append(path)
    return unique_dirs


def validate_file_path(file_path: str) -> Path:
    """
    Validate that a file path is safe to read from the server filesystem.
    Resolves the path canonically (following symlinks), then verifies it falls
    within one of the allowed base directories. Rejects paths to sensitive
    system locations regardless of allowlist.
    Args:
        file_path: The raw file path string to validate.
    Returns:
        Path: The resolved, validated Path object.
    Raises:
        ValueError: If the path is outside allowed directories or targets
                    a sensitive location.
    """
    resolved = Path(file_path).resolve()

    if not resolved.exists():
        raise FileNotFoundError(f"Path does not exist: {resolved}")

    # Block sensitive file patterns regardless of allowlist
    resolved_str = str(resolved)
    file_name = resolved.name.lower()

    path_parts = [part.lower() for part in resolved.parts]

    # Block .env files and variants (.env, .env.local, .env.production, etc.)
    if any(part == ".env" or part.startswith(".env.") for part in path_parts):
        raise ValueError(
            f"Access to '{resolved_str}' is not allowed: "
            ".env files may contain secrets and cannot be read, uploaded, or attached."
        )

    # Block well-known sensitive system paths (including macOS /private variants)
    sensitive_prefixes = (
        "/proc",
        "/sys",
        "/dev",
        "/etc/shadow",
        "/etc/passwd",
        "/private/etc/shadow",
        "/private/etc/passwd",
    )
    for prefix in sensitive_prefixes:
        if resolved_str == prefix or resolved_str.startswith(prefix + "/"):
            raise ValueError(
                f"Access to '{resolved_str}' is not allowed: "
                "path is in a restricted system location."
            )

    # Block sensitive directories that commonly contain credentials/keys.
    if ".ssh" in path_parts or ".aws" in path_parts:
        raise ValueError(
            f"Access to '{resolved_str}' is not allowed: "
            "path is in a directory that commonly contains secrets or credentials."
        )

    home = Path.home()
    sensitive_home_dirs = (
        ".kube",
        ".gnupg",
        ".config/gcloud",
    )
    for sensitive_dir in sensitive_home_dirs:
        blocked = home / sensitive_dir
        if resolved == blocked or str(resolved).startswith(str(blocked) + "/"):
            raise ValueError(
                f"Access to '{resolved_str}' is not allowed: "
                "path is in a directory that commonly contains secrets or credentials."
            )

    # Block other credential/secret file patterns
    sensitive_names = {
        ".credentials",
        ".credentials.json",
        "credentials.json",
        "client_secret.json",
        "client_secrets.json",
        "service_account.json",
        "service-account.json",
        ".npmrc",
        ".pypirc",
        ".netrc",
        ".git-credentials",
        ".docker/config.json",
    }
    if file_name in sensitive_names:
        raise ValueError(
            f"Access to '{resolved_str}' is not allowed: "
            "this file commonly contains secrets or credentials."
        )

    allowed_dirs = _get_allowed_file_dirs()
    if not allowed_dirs:
        raise ValueError(
            "No allowed file directories configured. "
            "Set the ALLOWED_FILE_DIRS environment variable or configure "
            "WORKSPACE_ATTACHMENT_DIR."
        )

    for allowed in allowed_dirs:
        try:
            resolved.relative_to(allowed)
            return resolved
        except ValueError:
            continue

    raise ValueError(
        f"Access to '{resolved_str}' is not allowed: "
        f"path is outside permitted directories ({', '.join(str(d) for d in allowed_dirs)}). "
        "Set ALLOWED_FILE_DIRS to adjust."
    )


def check_credentials_directory_permissions(credentials_dir: str = None) -> None:
    """
    Check if the service has appropriate permissions to create and write to the .credentials directory.
    Args:
        credentials_dir: Path to the credentials directory (default: uses get_default_credentials_dir())
    Raises:
        PermissionError: If the service lacks necessary permissions
        OSError: If there are other file system issues
    """
    if credentials_dir is None:
        from auth.google_auth import get_default_credentials_dir

        credentials_dir = get_default_credentials_dir()

    try:
        # Check if directory exists
        if os.path.exists(credentials_dir):
            # Directory exists, check if we can write to it
            test_file = os.path.join(credentials_dir, ".permission_test")
            try:
                with open(test_file, "w") as f:
                    f.write("test")
                os.remove(test_file)
                logger.info(
                    f"Credentials directory permissions check passed: {os.path.abspath(credentials_dir)}"
                )
            except (PermissionError, OSError) as e:
                raise PermissionError(
                    f"Cannot write to existing credentials directory '{os.path.abspath(credentials_dir)}': {e}"
                )
        else:
            # Directory doesn't exist, try to create it and its parent directories
            try:
                os.makedirs(credentials_dir, exist_ok=True)
                # Test writing to the new directory
                test_file = os.path.join(credentials_dir, ".permission_test")
                with open(test_file, "w") as f:
                    f.write("test")
                os.remove(test_file)
                logger.info(
                    f"Created credentials directory with proper permissions: {os.path.abspath(credentials_dir)}"
                )
            except (PermissionError, OSError) as e:
                # Clean up if we created the directory but can't write to it
                try:
                    if os.path.exists(credentials_dir):
                        os.rmdir(credentials_dir)
                except (PermissionError, OSError):
                    pass
                raise PermissionError(
                    f"Cannot create or write to credentials directory '{os.path.abspath(credentials_dir)}': {e}"
                )

    except PermissionError:
        raise
    except Exception as e:
        raise OSError(
            f"Unexpected error checking credentials directory permissions: {e}"
        )


def extract_office_xml_text(file_bytes: bytes, mime_type: str) -> Optional[str]:
    """
    Very light-weight XML scraper for Word, Excel, PowerPoint files.
    Returns plain-text if something readable is found, else None.
    Uses zipfile + defusedxml.ElementTree.
    """
    shared_strings: List[str] = []
    ns_excel_main = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"

    try:
        with zipfile.ZipFile(io.BytesIO(file_bytes)) as zf:
            targets: List[str] = []
            # Map MIME → iterable of XML files to inspect
            if (
                mime_type
                == "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
            ):
                targets = ["word/document.xml"]
            elif (
                mime_type
                == "application/vnd.openxmlformats-officedocument.presentationml.presentation"
            ):
                targets = [n for n in zf.namelist() if n.startswith("ppt/slides/slide")]
            elif (
                mime_type
                == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
            ):
                targets = [
                    n
                    for n in zf.namelist()
                    if n.startswith("xl/worksheets/sheet") and "drawing" not in n
                ]
                # Attempt to parse sharedStrings.xml for Excel files
                try:
                    shared_strings_xml = zf.read("xl/sharedStrings.xml")
                    shared_strings_root = ET.fromstring(shared_strings_xml)
                    for si_element in shared_strings_root.findall(
                        f"{{{ns_excel_main}}}si"
                    ):
                        text_parts = []
                        # Find all <t> elements, simple or within <r> runs, and concatenate their text
                        for t_element in si_element.findall(f".//{{{ns_excel_main}}}t"):
                            if t_element.text:
                                text_parts.append(t_element.text)
                        shared_strings.append("".join(text_parts))
                except KeyError:
                    logger.info(
                        "No sharedStrings.xml found in Excel file (this is optional)."
                    )
                except ET.ParseError as e:
                    logger.error(f"Error parsing sharedStrings.xml: {e}")
                except (
                    Exception
                ) as e:  # Catch any other unexpected error during sharedStrings parsing
                    logger.error(
                        f"Unexpected error processing sharedStrings.xml: {e}",
                        exc_info=True,
                    )
            else:
                return None

            pieces: List[str] = []
            for member in targets:
                try:
                    xml_content = zf.read(member)
                    xml_root = ET.fromstring(xml_content)
                    member_texts: List[str] = []

                    if (
                        mime_type
                        == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
                    ):
                        for cell_element in xml_root.findall(
                            f".//{{{ns_excel_main}}}c"
                        ):  # Find all <c> elements
                            value_element = cell_element.find(
                                f"{{{ns_excel_main}}}v"
                            )  # Find <v> under <c>

                            # Skip if cell has no value element or value element has no text
                            if value_element is None or value_element.text is None:
                                continue

                            cell_type = cell_element.get("t")
                            if cell_type == "s":  # Shared string
                                try:
                                    ss_idx = int(value_element.text)
                                    if 0 <= ss_idx < len(shared_strings):
                                        member_texts.append(shared_strings[ss_idx])
                                    else:
                                        logger.warning(
                                            f"Invalid shared string index {ss_idx} in {member}. Max index: {len(shared_strings) - 1}"
                                        )
                                except ValueError:
                                    logger.warning(
                                        f"Non-integer shared string index: '{value_element.text}' in {member}."
                                    )
                            else:  # Direct value (number, boolean, inline string if not 's')
                                member_texts.append(value_element.text)
                    else:  # Word or PowerPoint
                        for elem in xml_root.iter():
                            # For Word: <w:t> where w is "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
                            # For PowerPoint: <a:t> where a is "http://schemas.openxmlformats.org/drawingml/2006/main"
                            if (
                                elem.tag.endswith("}t") and elem.text
                            ):  # Check for any namespaced tag ending with 't'
                                cleaned_text = elem.text.strip()
                                if (
                                    cleaned_text
                                ):  # Add only if there's non-whitespace text
                                    member_texts.append(cleaned_text)

                    if member_texts:
                        pieces.append(
                            " ".join(member_texts)
                        )  # Join texts from one member with spaces

                except ET.ParseError as e:
                    logger.warning(
                        f"Could not parse XML in member '{member}' for {mime_type} file: {e}"
                    )
                except Exception as e:
                    logger.error(
                        f"Error processing member '{member}' for {mime_type}: {e}",
                        exc_info=True,
                    )
                    # continue processing other members

            if not pieces:  # If no text was extracted at all
                return None

            # Join content from different members (sheets/slides) with double newlines for separation
            text = "\n\n".join(pieces).strip()
            return text or None  # Ensure None is returned if text is empty after strip

    except zipfile.BadZipFile:
        logger.warning(f"File is not a valid ZIP archive (mime_type: {mime_type}).")
        return None
    except (
        ET.ParseError
    ) as e:  # Catch parsing errors at the top level if zipfile itself is XML-like
        logger.error(f"XML parsing error at a high level for {mime_type}: {e}")
        return None
    except Exception as e:
        logger.error(
            f"Failed to extract office XML text for {mime_type}: {e}", exc_info=True
        )
        return None


IMAGE_MIME_TYPES = {
    "image/png",
    "image/jpeg",
    "image/gif",
    "image/webp",
    "image/bmp",
    "image/tiff",
    "image/svg+xml",
}


def extract_pdf_text(file_bytes: bytes) -> Optional[str]:
    """
    Extract text from a PDF using pypdf.
    Returns plain text with pages separated by double newlines, or None on failure.
    """
    try:
        from pypdf import PdfReader

        reader = PdfReader(io.BytesIO(file_bytes))
        pages = []
        for page in reader.pages:
            text = page.extract_text()
            if text:
                pages.append(text)
        if not pages:
            return None
        return "\n\n".join(pages).strip() or None
    except Exception as e:
        logger.warning(f"Failed to extract PDF text: {e}")
        return None


def encode_image_content(file_bytes: bytes, mime_type: str) -> str:
    """
    Base64-encode image bytes with a mime type metadata prefix.
    Args:
        file_bytes: The image file content as bytes.
        mime_type: The MIME type of the image (must start with "image/").
    Returns:
        str: Base64-encoded image with mime type prefix.
    Raises:
        ValueError: If mime_type is not an image MIME type.
    """
    if not mime_type.startswith("image/"):
        raise ValueError(
            f"Expected image/* MIME type, got '{mime_type}'. "
            "Only image content can be base64-encoded for multimodal clients."
        )
    encoded = base64.b64encode(file_bytes).decode("ascii")
    return f"[base64_image:{mime_type}]{encoded}"


def handle_http_errors(
    tool_name: str, is_read_only: bool = False, service_type: Optional[str] = None
):
    """
    A decorator to handle Google API HttpErrors and transient SSL errors in a standardized way.
    It wraps a tool function, catches HttpError, logs a detailed error message,
    and raises a generic Exception with a user-friendly message.
    If is_read_only is True, it will also catch ssl.SSLError and retry with
    exponential backoff. After exhausting retries, it raises a TransientNetworkError.
    Args:
        tool_name (str): The name of the tool being decorated (e.g., 'list_calendars').
        is_read_only (bool): If True, the operation is considered safe to retry on
                             transient network errors. Defaults to False.
        service_type (str): Optional. The Google service type (e.g., 'calendar', 'gmail').
    """

    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            max_retries = 3
            base_delay = 1

            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except ssl.SSLError as e:
                    if is_read_only and attempt < max_retries - 1:
                        delay = base_delay * (2**attempt)
                        logger.warning(
                            f"SSL error in {tool_name} on attempt {attempt + 1}: {e}. Retrying in {delay} seconds..."
                        )
                        await asyncio.sleep(delay)
                    else:
                        logger.error(
                            f"SSL error in {tool_name} on final attempt: {e}. Raising exception."
                        )
                        raise TransientNetworkError(
                            f"A transient SSL error occurred in '{tool_name}' after {max_retries} attempts. "
                            "This is likely a temporary network or certificate issue. Please try again shortly."
                        ) from e
                except UserInputError as e:
                    message = f"Input error in {tool_name}: {e}"
                    logger.warning(message)
                    raise e
                except HttpError as error:
                    user_google_email = kwargs.get("user_google_email", "N/A")
                    error_details = str(error)

                    # Check if this is an API not enabled error
                    if (
                        error.resp.status == 403
                        and "accessNotConfigured" in error_details
                    ):
                        enablement_msg = get_api_enablement_message(
                            error_details, service_type
                        )

                        if enablement_msg:
                            message = (
                                f"API error in {tool_name}: {enablement_msg}\n\n"
                                f"User: {user_google_email}"
                            )
                        else:
                            message = (
                                f"API error in {tool_name}: {error}. "
                                f"The required API is not enabled for your project. "
                                f"Please check the Google Cloud Console to enable it."
                            )
                    elif error.resp.status in [401, 403]:
                        # Authentication/authorization errors
                        if is_oauth21_enabled():
                            if is_external_oauth21_provider():
                                auth_hint = (
                                    "LLM: Ask the user to provide a valid OAuth 2.1 "
                                    "bearer token in the Authorization header and retry."
                                )
                            else:
                                auth_hint = (
                                    "LLM: Ask the user to authenticate via their MCP "
                                    "client's OAuth 2.1 flow and retry."
                                )
                        else:
                            auth_hint = (
                                "LLM: Try 'start_google_auth' with the user's email "
                                "and the appropriate service_name."
                            )
                        message = (
                            f"API error in {tool_name}: {error}. "
                            f"You might need to re-authenticate for user '{user_google_email}'. "
                            f"{auth_hint}"
                        )
                    else:
                        # Other HTTP errors (400 Bad Request, etc.) - don't suggest re-auth
                        message = f"API error in {tool_name}: {error}"

                    logger.error(f"API error in {tool_name}: {error}", exc_info=True)
                    raise Exception(message) from error
                except TransientNetworkError:
                    # Re-raise without wrapping to preserve the specific error type
                    raise
                except GoogleAuthenticationError:
                    # Re-raise authentication errors without wrapping
                    raise
                except Exception as e:
                    message = f"An unexpected error occurred in {tool_name}: {e}"
                    logger.exception(message)
                    raise Exception(message) from e

        # Propagate _required_google_scopes if present (for tool filtering)
        if hasattr(func, "_required_google_scopes"):
            wrapper._required_google_scopes = func._required_google_scopes

        return wrapper

    return decorator

	import base64
	import io
	import json
	import logging
	import os
	import zipfile
	import ssl
	import asyncio
	import functools

	from pathlib import Path
	from typing import Annotated, Any, List, Optional

	from pydantic import BeforeValidator
	from defusedxml import ElementTree as ET

	from googleapiclient.errors import HttpError
	from .api_enablement import get_api_enablement_message
	from auth.google_auth import GoogleAuthenticationError
	from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider

	logger = logging.getLogger(__name__)


	class TransientNetworkError(Exception):
	"""Custom exception for transient network errors after retries."""

	pass


	class UserInputError(Exception):
	"""Raised for user-facing input/validation errors that shouldn't be retried."""

	pass


	def _coerce_json_str_to_type(v: Any, expected_type: type) -> Any:
	"""Coerce a JSON-encoded string to a specific container type."""
	if not isinstance(v, str):
	return v

	try:
	parsed = json.loads(v)
	except (json.JSONDecodeError, TypeError):
	return v

	return parsed if isinstance(parsed, expected_type) else v


	def _coerce_json_str_to_list(v: Any) -> Any:
	"""Coerce a JSON-encoded string to a list.

	Some MCP clients (e.g. Cowork) serialise array parameters as JSON strings
	rather than native arrays. This ``BeforeValidator`` transparently converts
	``'["a","b"]'`` → ``["a", "b"]`` so Pydantic validation succeeds.
	"""
	return _coerce_json_str_to_type(v, list)


	StringList = Annotated[List[str], BeforeValidator(_coerce_json_str_to_list)]
	"""``List[str]`` that also accepts a JSON-encoded string of an array.

	Use in tool signatures instead of ``List[str]`` to work around MCP clients
	that send ``'["value"]'`` instead of ``["value"]``.
	"""


	DictList = Annotated[List[dict[str, Any]], BeforeValidator(_coerce_json_str_to_list)]
	"""``List[dict]`` that also accepts a JSON-encoded string of an array.

	Use in tool signatures instead of ``List[dict]`` to work around MCP clients
	that send ``'[{"key":"val"}]'`` instead of ``[{"key":"val"}]``.
	"""


	def _coerce_json_str_to_dict(v: Any) -> Any:
	"""Coerce a JSON-encoded string to a dict.

	Some MCP clients serialise dict parameters as JSON strings rather than
	native objects. This ``BeforeValidator`` transparently converts
	``'{"key":"val"}'`` -> ``{"key": "val"}`` so Pydantic validation succeeds.
	"""
	return _coerce_json_str_to_type(v, dict)


	JsonDict = Annotated[dict[str, Any], BeforeValidator(_coerce_json_str_to_dict)]
	"""``dict`` that also accepts a JSON-encoded string of an object.

	Use in tool signatures instead of ``Dict[str, Any]`` to work around MCP clients
	that send ``'{"key":"val"}'`` instead of ``{"key": "val"}``.
	"""


	# Directories from which local file reads are allowed.
	# By default, only the managed attachment storage directory is trusted.
	# Override via ALLOWED_FILE_DIRS env var (os.pathsep-separated paths).
	_ALLOWED_FILE_DIRS_ENV = "ALLOWED_FILE_DIRS"


	def _get_allowed_file_dirs() -> list[Path]:
	"""Return the list of directories from which local file access is permitted."""
	from core.attachment_storage import STORAGE_DIR

	allowed_dirs: list[Path] = [STORAGE_DIR]
	env_val = os.environ.get(_ALLOWED_FILE_DIRS_ENV)
	if env_val:
	allowed_dirs.extend(
	Path(p_stripped).expanduser().resolve()
	for p in env_val.split(os.pathsep)
	if (p_stripped := p.strip())
	)

	unique_dirs: list[Path] = []
	seen: set[Path] = set()
	for path in allowed_dirs:
	if path in seen:
	continue
	seen.add(path)
	unique_dirs.append(path)
	return unique_dirs


	def validate_file_path(file_path: str) -> Path:
	"""
	Validate that a file path is safe to read from the server filesystem.

	Resolves the path canonically (following symlinks), then verifies it falls
	within one of the allowed base directories. Rejects paths to sensitive
	system locations regardless of allowlist.

	Args:
	file_path: The raw file path string to validate.

	Returns:
	Path: The resolved, validated Path object.

	Raises:
	ValueError: If the path is outside allowed directories or targets
	a sensitive location.
	"""
	resolved = Path(file_path).resolve()

	if not resolved.exists():
	raise FileNotFoundError(f"Path does not exist: {resolved}")

	# Block sensitive file patterns regardless of allowlist
	resolved_str = str(resolved)
	file_name = resolved.name.lower()

	path_parts = [part.lower() for part in resolved.parts]

	# Block .env files and variants (.env, .env.local, .env.production, etc.)
	if any(part == ".env" or part.startswith(".env.") for part in path_parts):
	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	".env files may contain secrets and cannot be read, uploaded, or attached."
	)

	# Block well-known sensitive system paths (including macOS /private variants)
	sensitive_prefixes = (
	"/proc",
	"/sys",
	"/dev",
	"/etc/shadow",
	"/etc/passwd",
	"/private/etc/shadow",
	"/private/etc/passwd",
	)
	for prefix in sensitive_prefixes:
	if resolved_str == prefix or resolved_str.startswith(prefix + "/"):
	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	"path is in a restricted system location."
	)

	# Block sensitive directories that commonly contain credentials/keys.
	if ".ssh" in path_parts or ".aws" in path_parts:
	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	"path is in a directory that commonly contains secrets or credentials."
	)

	home = Path.home()
	sensitive_home_dirs = (
	".kube",
	".gnupg",
	".config/gcloud",
	)
	for sensitive_dir in sensitive_home_dirs:
	blocked = home / sensitive_dir
	if resolved == blocked or str(resolved).startswith(str(blocked) + "/"):
	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	"path is in a directory that commonly contains secrets or credentials."
	)

	# Block other credential/secret file patterns
	sensitive_names = {
	".credentials",
	".credentials.json",
	"credentials.json",
	"client_secret.json",
	"client_secrets.json",
	"service_account.json",
	"service-account.json",
	".npmrc",
	".pypirc",
	".netrc",
	".git-credentials",
	".docker/config.json",
	}
	if file_name in sensitive_names:
	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	"this file commonly contains secrets or credentials."
	)

	allowed_dirs = _get_allowed_file_dirs()
	if not allowed_dirs:
	raise ValueError(
	"No allowed file directories configured. "
	"Set the ALLOWED_FILE_DIRS environment variable or configure "
	"WORKSPACE_ATTACHMENT_DIR."
	)

	for allowed in allowed_dirs:
	try:
	resolved.relative_to(allowed)
	return resolved
	except ValueError:
	continue

	raise ValueError(
	f"Access to '{resolved_str}' is not allowed: "
	f"path is outside permitted directories ({', '.join(str(d) for d in allowed_dirs)}). "
	"Set ALLOWED_FILE_DIRS to adjust."
	)


	def check_credentials_directory_permissions(credentials_dir: str = None) -> None:
	"""
	Check if the service has appropriate permissions to create and write to the .credentials directory.

	Args:
	credentials_dir: Path to the credentials directory (default: uses get_default_credentials_dir())

	Raises:
	PermissionError: If the service lacks necessary permissions
	OSError: If there are other file system issues
	"""
	if credentials_dir is None:
	from auth.google_auth import get_default_credentials_dir

	credentials_dir = get_default_credentials_dir()

	try:
	# Check if directory exists
	if os.path.exists(credentials_dir):
	# Directory exists, check if we can write to it
	test_file = os.path.join(credentials_dir, ".permission_test")
	try:
	with open(test_file, "w") as f:
	f.write("test")
	os.remove(test_file)
	logger.info(
	f"Credentials directory permissions check passed: {os.path.abspath(credentials_dir)}"
	)
	except (PermissionError, OSError) as e:
	raise PermissionError(
	f"Cannot write to existing credentials directory '{os.path.abspath(credentials_dir)}': {e}"
	)
	else:
	# Directory doesn't exist, try to create it and its parent directories
	try:
	os.makedirs(credentials_dir, exist_ok=True)
	# Test writing to the new directory
	test_file = os.path.join(credentials_dir, ".permission_test")
	with open(test_file, "w") as f:
	f.write("test")
	os.remove(test_file)
	logger.info(
	f"Created credentials directory with proper permissions: {os.path.abspath(credentials_dir)}"
	)
	except (PermissionError, OSError) as e:
	# Clean up if we created the directory but can't write to it
	try:
	if os.path.exists(credentials_dir):
	os.rmdir(credentials_dir)
	except (PermissionError, OSError):
	pass
	raise PermissionError(
	f"Cannot create or write to credentials directory '{os.path.abspath(credentials_dir)}': {e}"
	)

	except PermissionError:
	raise
	except Exception as e:
	raise OSError(
	f"Unexpected error checking credentials directory permissions: {e}"
	)


	def extract_office_xml_text(file_bytes: bytes, mime_type: str) -> Optional[str]:
	"""
	Very light-weight XML scraper for Word, Excel, PowerPoint files.
	Returns plain-text if something readable is found, else None.
	Uses zipfile + defusedxml.ElementTree.
	"""
	shared_strings: List[str] = []
	ns_excel_main = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"

	try:
	with zipfile.ZipFile(io.BytesIO(file_bytes)) as zf:
	targets: List[str] = []
	# Map MIME → iterable of XML files to inspect
	if (
	mime_type
	== "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
	):
	targets = ["word/document.xml"]
	elif (
	mime_type
	== "application/vnd.openxmlformats-officedocument.presentationml.presentation"
	):
	targets = [n for n in zf.namelist() if n.startswith("ppt/slides/slide")]
	elif (
	mime_type
	== "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
	):
	targets = [
	n
	for n in zf.namelist()
	if n.startswith("xl/worksheets/sheet") and "drawing" not in n
	]
	# Attempt to parse sharedStrings.xml for Excel files
	try:
	shared_strings_xml = zf.read("xl/sharedStrings.xml")
	shared_strings_root = ET.fromstring(shared_strings_xml)
	for si_element in shared_strings_root.findall(
	f"{{{ns_excel_main}}}si"
	):
	text_parts = []
	# Find all <t> elements, simple or within <r> runs, and concatenate their text
	for t_element in si_element.findall(f".//{{{ns_excel_main}}}t"):
	if t_element.text:
	text_parts.append(t_element.text)
	shared_strings.append("".join(text_parts))
	except KeyError:
	logger.info(
	"No sharedStrings.xml found in Excel file (this is optional)."
	)
	except ET.ParseError as e:
	logger.error(f"Error parsing sharedStrings.xml: {e}")
	except (
	Exception
	) as e: # Catch any other unexpected error during sharedStrings parsing
	logger.error(
	f"Unexpected error processing sharedStrings.xml: {e}",
	exc_info=True,
	)
	else:
	return None

	pieces: List[str] = []
	for member in targets:
	try:
	xml_content = zf.read(member)
	xml_root = ET.fromstring(xml_content)
	member_texts: List[str] = []

	if (
	mime_type
	== "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
	):
	for cell_element in xml_root.findall(
	f".//{{{ns_excel_main}}}c"
	): # Find all <c> elements
	value_element = cell_element.find(
	f"{{{ns_excel_main}}}v"
	) # Find <v> under <c>

	# Skip if cell has no value element or value element has no text
	if value_element is None or value_element.text is None:
	continue

	cell_type = cell_element.get("t")
	if cell_type == "s": # Shared string
	try:
	ss_idx = int(value_element.text)
	if 0 <= ss_idx < len(shared_strings):
	member_texts.append(shared_strings[ss_idx])
	else:
	logger.warning(
	f"Invalid shared string index {ss_idx} in {member}. Max index: {len(shared_strings) - 1}"
	)
	except ValueError:
	logger.warning(
	f"Non-integer shared string index: '{value_element.text}' in {member}."
	)
	else: # Direct value (number, boolean, inline string if not 's')
	member_texts.append(value_element.text)
	else: # Word or PowerPoint
	for elem in xml_root.iter():
	# For Word: <w:t> where w is "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
	# For PowerPoint: <a:t> where a is "http://schemas.openxmlformats.org/drawingml/2006/main"
	if (
	elem.tag.endswith("}t") and elem.text
	): # Check for any namespaced tag ending with 't'
	cleaned_text = elem.text.strip()
	if (
	cleaned_text
	): # Add only if there's non-whitespace text
	member_texts.append(cleaned_text)

	if member_texts:
	pieces.append(
	" ".join(member_texts)
	) # Join texts from one member with spaces

	except ET.ParseError as e:
	logger.warning(
	f"Could not parse XML in member '{member}' for {mime_type} file: {e}"
	)
	except Exception as e:
	logger.error(
	f"Error processing member '{member}' for {mime_type}: {e}",
	exc_info=True,
	)
	# continue processing other members

	if not pieces: # If no text was extracted at all
	return None

	# Join content from different members (sheets/slides) with double newlines for separation
	text = "\n\n".join(pieces).strip()
	return text or None # Ensure None is returned if text is empty after strip

	except zipfile.BadZipFile:
	logger.warning(f"File is not a valid ZIP archive (mime_type: {mime_type}).")
	return None
	except (
	ET.ParseError
	) as e: # Catch parsing errors at the top level if zipfile itself is XML-like
	logger.error(f"XML parsing error at a high level for {mime_type}: {e}")
	return None
	except Exception as e:
	logger.error(
	f"Failed to extract office XML text for {mime_type}: {e}", exc_info=True
	)
	return None


	IMAGE_MIME_TYPES = {
	"image/png",
	"image/jpeg",
	"image/gif",
	"image/webp",
	"image/bmp",
	"image/tiff",
	"image/svg+xml",
	}


	def extract_pdf_text(file_bytes: bytes) -> Optional[str]:
	"""
	Extract text from a PDF using pypdf.
	Returns plain text with pages separated by double newlines, or None on failure.
	"""
	try:
	from pypdf import PdfReader

	reader = PdfReader(io.BytesIO(file_bytes))
	pages = []
	for page in reader.pages:
	text = page.extract_text()
	if text:
	pages.append(text)
	if not pages:
	return None
	return "\n\n".join(pages).strip() or None
	except Exception as e:
	logger.warning(f"Failed to extract PDF text: {e}")
	return None


	def encode_image_content(file_bytes: bytes, mime_type: str) -> str:
	"""
	Base64-encode image bytes with a mime type metadata prefix.

	Args:
	file_bytes: The image file content as bytes.
	mime_type: The MIME type of the image (must start with "image/").

	Returns:
	str: Base64-encoded image with mime type prefix.

	Raises:
	ValueError: If mime_type is not an image MIME type.
	"""
	if not mime_type.startswith("image/"):
	raise ValueError(
	f"Expected image/* MIME type, got '{mime_type}'. "
	"Only image content can be base64-encoded for multimodal clients."
	)
	encoded = base64.b64encode(file_bytes).decode("ascii")
	return f"[base64_image:{mime_type}]{encoded}"


	def handle_http_errors(
	tool_name: str, is_read_only: bool = False, service_type: Optional[str] = None
	):
	"""
	A decorator to handle Google API HttpErrors and transient SSL errors in a standardized way.

	It wraps a tool function, catches HttpError, logs a detailed error message,
	and raises a generic Exception with a user-friendly message.

	If is_read_only is True, it will also catch ssl.SSLError and retry with
	exponential backoff. After exhausting retries, it raises a TransientNetworkError.

	Args:
	tool_name (str): The name of the tool being decorated (e.g., 'list_calendars').
	is_read_only (bool): If True, the operation is considered safe to retry on
	transient network errors. Defaults to False.
	service_type (str): Optional. The Google service type (e.g., 'calendar', 'gmail').
	"""

	def decorator(func):
	@functools.wraps(func)
	async def wrapper(args, *kwargs):
	max_retries = 3
	base_delay = 1

	for attempt in range(max_retries):
	try:
	return await func(args, *kwargs)
	except ssl.SSLError as e:
	if is_read_only and attempt < max_retries - 1:
	delay = base_delay * (2**attempt)
	logger.warning(
	f"SSL error in {tool_name} on attempt {attempt + 1}: {e}. Retrying in {delay} seconds..."
	)
	await asyncio.sleep(delay)
	else:
	logger.error(
	f"SSL error in {tool_name} on final attempt: {e}. Raising exception."
	)
	raise TransientNetworkError(
	f"A transient SSL error occurred in '{tool_name}' after {max_retries} attempts. "
	"This is likely a temporary network or certificate issue. Please try again shortly."
	) from e
	except UserInputError as e:
	message = f"Input error in {tool_name}: {e}"
	logger.warning(message)
	raise e
	except HttpError as error:
	user_google_email = kwargs.get("user_google_email", "N/A")
	error_details = str(error)

	# Check if this is an API not enabled error
	if (
	error.resp.status == 403
	and "accessNotConfigured" in error_details
	):
	enablement_msg = get_api_enablement_message(
	error_details, service_type
	)

	if enablement_msg:
	message = (
	f"API error in {tool_name}: {enablement_msg}\n\n"
	f"User: {user_google_email}"
	)
	else:
	message = (
	f"API error in {tool_name}: {error}. "
	f"The required API is not enabled for your project. "
	f"Please check the Google Cloud Console to enable it."
	)
	elif error.resp.status in [401, 403]:
	# Authentication/authorization errors
	if is_oauth21_enabled():
	if is_external_oauth21_provider():
	auth_hint = (
	"LLM: Ask the user to provide a valid OAuth 2.1 "
	"bearer token in the Authorization header and retry."
	)
	else:
	auth_hint = (
	"LLM: Ask the user to authenticate via their MCP "
	"client's OAuth 2.1 flow and retry."
	)
	else:
	auth_hint = (
	"LLM: Try 'start_google_auth' with the user's email "
	"and the appropriate service_name."
	)
	message = (
	f"API error in {tool_name}: {error}. "
	f"You might need to re-authenticate for user '{user_google_email}'. "
	f"{auth_hint}"
	)
	else:
	# Other HTTP errors (400 Bad Request, etc.) - don't suggest re-auth
	message = f"API error in {tool_name}: {error}"

	logger.error(f"API error in {tool_name}: {error}", exc_info=True)
	raise Exception(message) from error
	except TransientNetworkError:
	# Re-raise without wrapping to preserve the specific error type
	raise
	except GoogleAuthenticationError:
	# Re-raise authentication errors without wrapping
	raise
	except Exception as e:
	message = f"An unexpected error occurred in {tool_name}: {e}"
	logger.exception(message)
	raise Exception(message) from e

	# Propagate _required_google_scopes if present (for tool filtering)
	if hasattr(func, "_required_google_scopes"):
	wrapper._required_google_scopes = func._required_google_scopes

	return wrapper

	return decorator