Spaces:

pedrorezende
/

multi-agent-sdlc

Running

File size: 7,514 Bytes

# This will require some refactoring in the future, e.g. moving some content to "centralized" files for code reuse.

from typing import Annotated
from multi_agent_sdlc.runtime.paths import normalise_relative_path
import re
from pydantic import AfterValidator, Field, StringConstraints

ENTRY_POINT_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*")

MODULE_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*$")

DEPENDENCY_PATTERN = re.compile(
    r"^[A-Za-z0-9][A-Za-z0-9._-]*"
    r"(?:\[[A-Za-z0-9._,-]+\])?"
    r"(?:(?:===|==|~=|!=|<=|>=|<|>)[A-Za-z0-9.*+!_-]+)?$"
)

REMOTE_DEPENDENCY_PREFIXES = (
    "http://",
    "https://",
    "git+",
    "ssh://",
    "file:",
)


TEST_DIRECTORY_NAMES = {
    "test",
    "tests",
    "__tests__",
    "spec",
    "specs",
}

PROHIBITED_ENTRY_POINTS = {
    "pytest",
    "ruff",
    "mypy",
    "flake8",
    "coverage",
    "coverage3",
    "pip",
    "pip3",
    "bash",
    "sh",
    "zsh",
    "fish",
    "powershell",
    "pwsh",
}

PROHIBITED_PYTHON_MODULES = {
    "pytest",
    "unittest",
    "coverage",
    "pip",
    "ensurepip",
    "venv",
    "subprocess",
}

PROHIBITED_RUNTIME_PACKAGES = {
    "pytest",
    "pytest-cov",
    "coverage",
    "ruff",
    "mypy",
    "flake8",
    "tox",
    "nox",
}


def validate_entry_point(entry_point: str) -> str:
    """Validate and return a project entry-point name."""

    cleaned = entry_point.strip()

    if not cleaned:
        raise ValueError("Entry point cannot be empty.")

    if cleaned.lower() in PROHIBITED_ENTRY_POINTS:
        raise PermissionError(f"The Coder cannot execute `{cleaned}`.")

    return cleaned


def validate_application_arguments(
    arguments: list[str],
) -> list[str]:
    """Validate arguments passed to an application or module."""

    validated: list[str] = []

    for argument in arguments:
        if not isinstance(argument, str):
            raise TypeError("Every application argument must be a string.")

        if "\n" in argument or "\r" in argument:
            raise PermissionError("Multiline application arguments are not allowed.")

        if "\x00" in argument:
            raise PermissionError(
                "Null bytes are not allowed in application arguments."
            )

        validated.append(argument)

    return validated


def validate_module_name(module: str) -> str:
    """Validate and return an application module name."""

    cleaned = module.strip()

    if not cleaned:
        raise ValueError("Module cannot be empty.")

    if not MODULE_PATTERN.fullmatch(cleaned):
        raise ValueError(f"Invalid Python module name: {cleaned}")

    root_module = cleaned.split(".", maxsplit=1)[0].lower()

    if root_module in PROHIBITED_PYTHON_MODULES:
        raise PermissionError(f"The Coder cannot execute Python module `{cleaned}`.")

    return cleaned


def validate_runtime_dependency(package: str) -> str:
    """Validate and return a runtime dependency specification."""
    cleaned = package.strip()

    if not cleaned:
        raise ValueError("Dependency specification cannot be empty.")

    lowered = cleaned.lower()

    if lowered.startswith(REMOTE_DEPENDENCY_PREFIXES):
        raise PermissionError(
            "Git, URL, SSH, and local-file dependencies are not allowed: " f"{cleaned}"
        )

    if cleaned.startswith("-"):
        raise PermissionError(
            f"Dependency command-line options are not allowed: {cleaned}"
        )

    if "@" in cleaned:
        raise PermissionError(
            f"Direct-reference dependencies are not allowed: {cleaned}"
        )

    if not DEPENDENCY_PATTERN.fullmatch(cleaned):
        raise ValueError(f"Invalid runtime dependency specification: {cleaned}")

    dependency_name = normalise_dependency_name(cleaned)

    if dependency_name in PROHIBITED_RUNTIME_PACKAGES:
        raise PermissionError(
            f"`{cleaned}` is a testing or development dependency. "
            "The Coder may add runtime dependencies only."
        )

    return cleaned


def is_test_related_path(path: str) -> bool:
    """Return True when a path includes a blocked test directory."""
    candidate = normalise_relative_path(path)

    return any(part.lower() in TEST_DIRECTORY_NAMES for part in candidate.parts)


def reject_coder_test_path(path: str) -> None:
    """Prevent the Coder from writing into test-related directories."""
    if is_test_related_path(path):
        raise PermissionError(
            "The Coder cannot create or modify files or directories "
            f"inside test-related paths: {path}. "
            "Test implementation belongs to the Tester. "
            "Do not retry this operation using another tool or path."
        )


def validate_project_relative_path(path: str) -> str:
    candidate = normalise_relative_path(path.strip())
    return candidate.as_posix()


def normalise_dependency_name(specification: str) -> str:
    """Extract the normalised package name from a dependency specification."""
    match = re.match(
        r"^[A-Za-z0-9][A-Za-z0-9._-]*",
        specification,
    )

    if match is None:
        return ""

    return match.group(0).lower().replace("_", "-")


def validate_file_content(content: str) -> str:
    if "\x00" in content:
        raise ValueError("File content cannot contain null bytes.")

    return content


EntryPoint = Annotated[
    str,
    StringConstraints(
        strip_whitespace=True,
        min_length=1,
        max_length=100,
        pattern=r"[A-Za-z0-9][A-Za-z0-9._-]*",
    ),
    AfterValidator(validate_entry_point),
]

StandardInput = Annotated[
    str,
    Field(
        min_length=1,
        max_length=10_000,
        description=(
            "Text sent to the application's standard input. Separate "
            "interactive responses with newline characters. Omit this "
            "argument when no standard input is required."
        ),
    ),
]

ApplicationArguments = Annotated[
    list[str],
    Field(
        max_length=50,
        description="Arguments passed directly to the application.",
    ),
    AfterValidator(validate_application_arguments),
]

ExecutionTimeout = Annotated[
    int,
    Field(
        ge=1,
        le=200,
        description="Maximum application execution time in seconds.",
    ),
]

PythonModuleName = Annotated[
    str,
    Field(
        min_length=1,
        max_length=200,
        description=(
            "A dotted Python module belonging to the generated application. "
            "Testing, package-management, environment-management, and "
            "process-execution modules are prohibited."
        ),
    ),
    AfterValidator(validate_module_name),
]

RuntimeDependency = Annotated[
    str,
    AfterValidator(validate_runtime_dependency),
]


RuntimeDependencies = Annotated[
    list[RuntimeDependency],
    Field(
        min_length=1,
        max_length=20,
        description="Runtime dependencies to add with uv.",
    ),
]

ProjectRelativePath = Annotated[
    str,
    Field(
        min_length=1,
        max_length=500,
        description=(
            "A path relative to the project directory. "
            "Absolute paths and parent-directory traversal are prohibited."
        ),
    ),
    AfterValidator(validate_project_relative_path),
]

FileContent = Annotated[
    str,
    Field(
        max_length=500_000,
        description="Complete UTF-8 text content to write to the file.",
    ),
    AfterValidator(validate_file_content),
]