dataiku_mcp/tools/utils/validation.py

# Copyright 2026 Dataiku SAS
#
# 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.

"""Shared validation helpers for tool modules."""

import re
from pathlib import PurePosixPath
from urllib.parse import urlsplit

_HEX_COLOR_PATTERN = re.compile(r"^#(?:[0-9a-fA-F]{3}|[0-9a-fA-F]{6})$")


def require_non_empty_string(value: str, field_name: str) -> str:
    try:
        cleaned = value.strip()
    except Exception as exc:
        raise ValueError(f"'{field_name}' must be a non-empty string") from exc
    if not cleaned:
        raise ValueError(f"'{field_name}' must be a non-empty string")
    return cleaned


def require_non_empty_strings(values: list[str], field_name: str) -> list[str]:
    return [
        require_non_empty_string(value, f"{field_name}[{index}]")
        for index, value in enumerate(values)
    ]


def require_positive_int(value: int, field_name: str) -> int:
    if value is True or value is False:
        raise ValueError(f"'{field_name}' must be >= 1")
    try:
        if value < 1:
            raise ValueError(f"'{field_name}' must be >= 1")
    except TypeError as exc:
        raise ValueError(f"'{field_name}' must be >= 1") from exc
    return value


def require_non_negative_int(value: int, field_name: str) -> int:
    if value is True or value is False:
        raise ValueError(f"'{field_name}' must be >= 0")
    try:
        if value < 0:
            raise ValueError(f"'{field_name}' must be >= 0")
    except TypeError as exc:
        raise ValueError(f"'{field_name}' must be >= 0") from exc
    return value


def require_int_at_least(value: int, field_name: str, minimum: int) -> int:
    cleaned = require_positive_int(value, field_name)
    if cleaned < minimum:
        raise ValueError(f"'{field_name}' must be >= {minimum}")
    return cleaned


def require_int_in_range(
    value: int, field_name: str, minimum: int, maximum: int
) -> int:
    """Require an int within an inclusive range, for bounded inline waits and caps."""
    cleaned = require_int_at_least(value, field_name, minimum)
    if cleaned > maximum:
        raise ValueError(f"'{field_name}' must be <= {maximum}")
    return cleaned


def require_non_empty_list(values: list, field_name: str) -> list:
    if len(values) < 1:
        raise ValueError(f"'{field_name}' must be a non-empty list")
    return values


def require_allowed_value(value: str, field_name: str, allowed_values: set[str]) -> str:
    if value not in allowed_values:
        allowed = sorted(allowed_values)
        raise ValueError(
            f"Invalid '{field_name}': '{value}'. Allowed values: {allowed}"
        )
    return value


def require_fraction(value: float, field_name: str) -> float:
    if value is True or value is False:
        raise ValueError(f"'{field_name}' must be > 0 and < 1")
    try:
        cleaned = float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"'{field_name}' must be > 0 and < 1") from exc
    if cleaned <= 0 or cleaned >= 1:
        raise ValueError(f"'{field_name}' must be > 0 and < 1")
    return cleaned


def require_hex_color(value: str, field_name: str) -> str:
    cleaned = require_non_empty_string(value, field_name)
    if not _HEX_COLOR_PATTERN.match(cleaned):
        raise ValueError(
            f"'{field_name}' must be a hex color in '#RGB' or '#RRGGBB' format"
        )
    return cleaned


def require_git_repository_url(value: str) -> str:
    """Accept remote Git URLs without inline secrets or local filesystem access."""
    value = require_non_empty_string(value, "repository_url")
    if any(character.isspace() for character in value):
        raise ValueError("'repository_url' must not contain whitespace")
    if re.fullmatch(r"[\w.][\w.-]*@[\w.-]+:[^?#]+", value):
        return value
    try:
        parsed = urlsplit(value)
        valid = (
            parsed.scheme in {"https", "http", "ssh", "git"}
            and parsed.hostname
            and parsed.path not in {"", "/"}
            and not parsed.query
            and not parsed.fragment
            and parsed.password is None
            and (parsed.username is None or parsed.scheme == "ssh")
        )
    except ValueError:
        valid = False
    if not valid:
        raise ValueError(
            "'repository_url' must be a remote Git URL without passwords, tokens, "
            "query parameters, or fragments. Configure credentials in Dataiku."
        )
    return value


def require_relative_posix_path(value: str, field_name: str) -> str:
    value = require_non_empty_string(value, field_name)
    path = PurePosixPath(value)
    if path.is_absolute() or ".." in path.parts or "\\" in value:
        raise ValueError(f"'{field_name}' must be a relative path without '..'")
    return value