nexus/tools/crontab_tool.py

"""
Crontab Tool - Read/write user crontab via `crontab` CLI.
Author: Hieu Louis (2026)
"""
from __future__ import annotations

import os
import subprocess
from typing import Dict, Any, List, Optional

from .base import Tool, ToolResult, ToolContext, ToolCategory, ToolSafety


# Operations được hỗ trợ
CRONTAB_OPERATIONS = {"list", "add", "remove", "replace", "show"}

# Read-only ops
READONLY_OPS = {"list", "show"}

# Write ops (cần confirmation + dry_run)
WRITE_OPS = {"add", "remove", "replace"}


class CrontabTool(Tool):
    """Đọc/ghi user crontab via `crontab` CLI."""
    category = ToolCategory.SYSTEM
    safety = ToolSafety.DANGEROUS
    requires_confirmation = True

    @property
    def name(self) -> str:
        return "crontab"

    @property
    def description(self) -> str:
        return (
            "Read/write user crontab: list (`crontab -l`), add entry, "
            "remove all (`crontab -r`), replace from file/text (`crontab -`). "
            "Có dry_run và backup trước khi ghi."
        )

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "operation": {
                    "type": "string",
                    "enum": sorted(CRONTAB_OPERATIONS),
                },
                "entry": {
                    "type": "string",
                    "description": "Cron entry to add (e.g. '0 3 * * * /opt/backup.sh')",
                },
                "content": {
                    "type": "string",
                    "description": "Full crontab content for 'replace'",
                },
                "user": {"type": "string", "description": "Target user (-u). Cần root."},
                "backup": {"type": "boolean", "default": True, "description": "Backup current crontab trước khi ghi"},
            },
            "required": ["operation"],
        }

    def validate_args(self, args: Dict[str, Any]) -> Optional[str]:
        op = args.get("operation")
        if not op:
            return "Missing required arg: operation"
        if op not in CRONTAB_OPERATIONS:
            return f"Unsupported operation: {op}"
        if op == "add" and not args.get("entry"):
            return "Operation 'add' requires 'entry' arg"
        if op == "replace" and not args.get("content"):
            return "Operation 'replace' requires 'content' arg"
        return None

    def _list_current(self, user: Optional[str], timeout: int) -> tuple:
        """Return (stdout, returncode). Trả về chuỗi rỗng nếu chưa có crontab."""
        cmd = ["crontab", "-l"]
        if user:
            cmd = ["crontab", "-u", user, "-l"]
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                check=False,
            )
            return result.stdout, result.returncode
        except FileNotFoundError:
            raise
        except subprocess.TimeoutExpired:
            raise

    def _write_crontab(
        self,
        content: str,
        user: Optional[str],
        timeout: int,
        env: Dict[str, str],
    ) -> subprocess.CompletedProcess:
        cmd = ["crontab", "-"]
        if user:
            cmd = ["crontab", "-u", user, "-"]
        return subprocess.run(
            cmd,
            input=content,
            capture_output=True,
            text=True,
            timeout=timeout,
            env=env,
            check=False,
        )

    def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult:
        op = args["operation"]
        user = args.get("user")
        env = dict(os.environ)
        env.update(context.env)

        # ---- list / show ----
        if op in {"list", "show"}:
            try:
                stdout, rc = self._list_current(user, context.timeout)
                if rc != 0 and "no crontab for" not in (stdout + "").lower():
                    return ToolResult(
                        success=False,
                        error=stdout or "crontab -l failed",
                        return_code=rc,
                    )
                return ToolResult(
                    success=True,
                    output=stdout or "(no crontab)",
                    return_code=0,
                    metadata={
                        "operation": op,
                        "user": user,
                        "empty": not stdout.strip(),
                    },
                )
            except FileNotFoundError:
                return ToolResult(
                    success=False,
                    error="crontab CLI not found.",
                    return_code=127,
                )
            except subprocess.TimeoutExpired:
                return ToolResult(success=False, error="crontab -l timed out", return_code=124)

        # ---- Dry-run for write ops ----
        if context.dry_run:
            preview: str
            if op == "add":
                preview = f"[dry-run] Would add entry: {args['entry']}"
            elif op == "remove":
                preview = "[dry-run] Would remove all crontab entries (crontab -r)"
            else:  # replace
                preview = f"[dry-run] Would replace crontab with:\n{args['content']}"
            return ToolResult(
                success=True,
                output=preview,
                metadata={
                    "dry_run": True,
                    "operation": op,
                    "user": user,
                },
            )

        # ---- Read current for backup/append ----
        try:
            current, _ = self._list_current(user, context.timeout)
        except FileNotFoundError:
            return ToolResult(
                success=False,
                error="crontab CLI not found.",
                return_code=127,
            )
        except subprocess.TimeoutExpired:
            return ToolResult(success=False, error="crontab -l timed out", return_code=124)

        backup_path: Optional[str] = None
        if args.get("backup", True) and current.strip():
            backup_path = os.path.join(
                context.working_dir,
                f".crontab.backup.{os.getpid()}.txt",
            )
            try:
                with open(backup_path, "w", encoding="utf-8") as f:
                    f.write(current)
            except OSError as e:
                return ToolResult(
                    success=False,
                    error=f"Backup failed: {e}",
                    return_code=1,
                )

        # ---- Compose new content ----
        if op == "add":
            new_content = current.rstrip() + "\n" + args["entry"] + "\n"
        elif op == "remove":
            new_content = ""
        else:  # replace
            new_content = args["content"] + (
                "" if args["content"].endswith("\n") else "\n"
            )

        # ---- Write ----
        try:
            result = self._write_crontab(new_content, user, context.timeout, env)
            return ToolResult(
                success=(result.returncode == 0),
                output=new_content,
                error=result.stderr or None,
                return_code=result.returncode,
                artifacts=[backup_path] if backup_path else [],
                metadata={
                    "operation": op,
                    "user": user,
                    "backup_path": backup_path,
                    "entry_count": len([l for l in new_content.splitlines() if l.strip() and not l.strip().startswith("#")]),
                    "dry_run": False,
                },
            )
        except FileNotFoundError:
            return ToolResult(success=False, error="crontab CLI not found.", return_code=127)
        except subprocess.TimeoutExpired:
            return ToolResult(success=False, error="crontab - timed out", return_code=124)
        except Exception as e:
            return ToolResult(success=False, error=str(e), return_code=1)