tools/_shell_safety.py

"""backend/tools/_shell_safety.py — Shell execution safety (single source of truth)

Chiude GAP-11 (registry.py) e GAP-12 (exec.py) con un'unica implementazione.
Importare da qui. Non duplicare la logica altrove.

Algoritmo:
1. Blocca metacaratteri shell (previene bypass allowlist)
2. shlex.split() — tokenizzazione sicura, rileva quoting anomalo
3. frozenset argv[0] — allowlist letterale, non regex di prefisso
4. Regole per git (subcmd), python -m e curl/wget (solo https://)
5. create_subprocess_exec / subprocess.run con lista argv — NO shell=True
"""
from __future__ import annotations
import asyncio, os, re, shlex, subprocess
from typing import Optional

_METACHAR_RE = re.compile(r'[;&|`<>\n\r]|\$[\(\{]')
_ALLOWED: frozenset = frozenset({
    "ls", "cat", "echo", "pwd", "whoami", "date", "uname",
    "python3", "python", "node", "npm", "pnpm",
    "grep", "find", "head", "tail", "wc", "sort", "uniq", "diff",
    "mkdir", "touch", "cp", "mv", "chmod", "git", "curl", "wget",
})
_GIT_OK: frozenset = frozenset({"status", "log", "diff", "show", "branch", "remote"})
_PYTHON_MODULES_BLOCKED: frozenset = frozenset({"pip", "pip3", "ensurepip"})


def safe_shell_env() -> dict:
    """Env pulita — nessun secret del processo padre ereditato."""
    return {
        "HOME": "/tmp", "TMPDIR": "/tmp",
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
        "LANG": os.environ.get("LANG", "en_US.UTF-8"),
        "TERM": "xterm-256color",
    }


# SEC-FS-JAIL: comandi che accettano path come argomenti — devono restare
# confinati alla stessa root di _safe_fs_path in registry.py, altrimenti
# l'allowlist shell diventa un bypass per leggere/scrivere file arbitrari
# (es. 'cat /etc/passwd', 'cp /run/secrets/x /tmp/y' erano permessi prima).
_PATH_TAKING_CMDS = frozenset({
    "cat", "cp", "mv", "touch", "mkdir", "chmod", "find", "head", "tail", "diff",
})

def _fs_jail_root() -> str:
    return os.path.realpath(os.getenv("FS_TOOL_ROOT", os.getcwd()))

def _path_is_jailed(candidate: str, root: str) -> bool:
    _abs = candidate if os.path.isabs(candidate) else os.path.join(root, candidate)
    _resolved = os.path.realpath(_abs)
    return _resolved == root or _resolved.startswith(root + os.sep)

def validate_shell_command(command: str) -> Optional[str]:
    """
    Valida command. Ritorna None se OK, stringa di errore se rifiutato.
    Thread-safe, sincrona, zero I/O.
    """
    cmd = command.strip()
    if not cmd:
        return "comando vuoto"
    if _METACHAR_RE.search(cmd):
        return "metacaratteri non permessi (; & | ` < > newline $() ${})"
    try:
        argv = shlex.split(cmd)
    except ValueError as e:
        return f"parsing fallito: {e}"
    if not argv:
        return "comando vuoto dopo parsing"
    exe = argv[0].lower()
    if exe not in _ALLOWED:
        return f"comando non permesso: '{exe}' (ammessi: {', '.join(sorted(_ALLOWED))})"
    if exe == "git":
        if len(argv) < 2 or argv[1].lower() not in _GIT_OK:
            return f"git sub-comando non permesso (ammessi: {', '.join(sorted(_GIT_OK))})"
    elif exe in ("python", "python3") and len(argv) >= 3 and argv[1] == "-m":
        if argv[2].lower() in _PYTHON_MODULES_BLOCKED:
            return f"modulo Python non permesso: '{argv[2]}'"
    elif exe in ("curl", "wget"):
        if not any(a.startswith("https://") for a in argv[1:]):
            return f"{exe}: solo URL https://"
    if exe in _PATH_TAKING_CMDS:
        _root = _fs_jail_root()
        for _arg in argv[1:]:
            if _arg.startswith("-"):
                continue  # flag, non un path
            if not _path_is_jailed(_arg, _root):
                return f"{exe}: path fuori dalla sandbox consentita ('{_arg}')"
    return None


async def run_shell_safe(command: str, cwd: Optional[str] = None, timeout: int = 30) -> dict:
    """Asincrona, NO shell=True. Per exec.py / endpoint HTTP."""
    err = validate_shell_command(command)
    if err:
        return {"ok": False, "stdout": "", "stderr": "", "error": err, "code": -1}
    argv = shlex.split(command.strip())
    try:
        proc = await asyncio.create_subprocess_exec(
            *argv, stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE, cwd=cwd, env=safe_shell_env())
        out, er2 = await asyncio.wait_for(proc.communicate(), timeout=float(timeout))
        return {"ok": proc.returncode == 0,
                "stdout": out.decode("utf-8", errors="replace")[:8192],
                "stderr": er2.decode("utf-8", errors="replace")[:2048],
                "error": None, "code": proc.returncode}
    except asyncio.TimeoutError:
        try:
            proc.kill()
        except Exception:
            pass
        return {"ok": False, "stdout": "", "stderr": "", "error": f"timeout ({timeout}s)", "code": -1}
    except Exception as exc:
        return {"ok": False, "stdout": "", "stderr": "", "error": str(exc), "code": -1}


def run_shell_safe_sync(command: str, cwd: Optional[str] = None, timeout: int = 30) -> dict:
    """Sincrona, NO shell=True. Per fallback subprocess in registry.py."""
    err = validate_shell_command(command)
    if err:
        return {"ok": False, "stdout": "", "stderr": "", "error": err, "code": -1}
    argv = shlex.split(command.strip())
    try:
        r = subprocess.run(argv, capture_output=True, text=True,
                           timeout=min(timeout, 120), cwd=cwd, env=safe_shell_env())
        return {"ok": r.returncode == 0, "stdout": r.stdout[:8192],
                "stderr": r.stderr[:2048], "error": None, "code": r.returncode}
    except subprocess.TimeoutExpired:
        return {"ok": False, "stdout": "", "stderr": "", "error": f"timeout ({timeout}s)", "code": -1}
    except Exception as exc:
        return {"ok": False, "stdout": "", "stderr": "", "error": str(exc), "code": -1}