Blame for tool_call.py

"""ReAct stili tool cagri ayriciligi + dispatcher.



Akis (genel)

------------

1. Model ureten bir metin doner. Genellikle su sablon:



       Thought: hedefin portlarini taramaliyim.

       Action: run_cmd

       Action Input: nmap -sV -p- 10.0.0.5



   ya da bitirirken:



       Thought: yeterli bilgi var.

       Final Answer: Acik portlar: 22/ssh, 80/http ...



2. `parse_react()` bu metni `ParsedResponse`'a cevirir.

3. agent.py birden fazla aksiyon varsa kullaniciya sorar (proje karari).

4. `ToolDispatcher.dispatch(call)` -> safety -> KaliSSH -> Observation string.

5. Observation modele "Observation: ..." prefix'i ile geri verilir.



Bu modul sadece parsing ve dispatch'i yapar; LLM cagrisi agent.py'da.



Modelfile uyumu

---------------

Modelfile'da `Observation:` ve `Tool Output:` stop token'lari var. Bu sayede

model `Action Input: ...` yazdiktan sonra durur ve biz observation'i

ekleyene kadar uretmez. parse_react bu sayede tek seferde bir aksiyona

odaklanir.

"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Callable, Optional

from safety import SafetyChecker, Verdict
from ssh_client import KaliSSH, SSHConnectionError

if TYPE_CHECKING:
    from logger import SessionLogger


# -----------------------------------------------------------------------------
# Veri tipleri
# -----------------------------------------------------------------------------


@dataclass
class ToolCall:
    """Model tarafindan istenen tek bir tool cagrisi."""

    tool: str
    args: dict[str, str]
    raw: str = ""  # ham metin (debug)

    def short(self, n: int = 120) -> str:
        body = self.args.get("cmd") or json.dumps(self.args, ensure_ascii=False)
        body = body.replace("\n", " ")
        return f"{self.tool}({body if len(body) <= n else body[: n - 3] + '...'})"


@dataclass
class ParsedResponse:
    """Modelin tek bir uretiminden cikarilmis yapisal bilgi."""

    thought: Optional[str]
    actions: list[ToolCall] = field(default_factory=list)
    final_answer: Optional[str] = None
    raw: str = ""

    @property
    def has_action(self) -> bool:
        return bool(self.actions)

    @property
    def is_final(self) -> bool:
        return self.final_answer is not None and not self.actions


# -----------------------------------------------------------------------------
# Parser
# -----------------------------------------------------------------------------

# Bir bolumun bittigini gosteren etiketler. Lookahead'de kullanilir.
_SECTION_LOOKAHEAD = (
    r"(?=\n\s*(?:Thought\s*:|Action\s*:|Action\s*Input\s*:|"
    r"Observation\s*:|Tool\s*Output\s*:|Final\s*Answer\s*:)|\Z)"
)

_THOUGHT_RE = re.compile(
    r"Thought\s*:\s*(?P<body>.*?)" + _SECTION_LOOKAHEAD,
    re.DOTALL | re.IGNORECASE,
)

# Klasik ReAct format:
#   Action: run_cmd
#   Action Input: nmap -sV 1.2.3.4
_ACTION_BLOCK_RE = re.compile(
    r"Action\s*:\s*(?P<tool>[A-Za-z_][\w\-]*)\s*\n"
    r"\s*Action\s*Input\s*:\s*(?P<input>.*?)" + _SECTION_LOOKAHEAD,
    re.DOTALL | re.IGNORECASE,
)

# Fonksiyon-cagri stili (model bu sekilde de yazabilir):
#   Action: run_cmd(nmap -sV 1.2.3.4)
# Parantez ici icerigi tolere etmek icin nested 1 seviye paren'e izin veriyoruz.
# Onemli: bu regex \n veya  '(' ile baslayan satirlari yakalamali, klasik
# formati da yanlis match etmesin diye lookahead ile parantez sart.
_ACTION_FUNCTION_RE = re.compile(
    r"Action\s*:\s*(?P<tool>[A-Za-z_][\w\-]*)\s*"
    r"\(\s*(?P<input>(?:[^()]|\([^()]*\))*?)\s*\)",
    re.DOTALL | re.IGNORECASE,
)

_FINAL_RE = re.compile(
    r"Final\s*Answer\s*:\s*(?P<body>.*?)\Z",
    re.DOTALL | re.IGNORECASE,
)


def _strip_input(raw: str) -> str:
    """Action Input'u temizle: kod blogu / tirnak isaretlerini sok."""
    text = raw.strip()
    if not text:
        return ""

    # ```bash\n...\n``` veya ```\n...\n```
    fenced = re.match(r"^```(?:[a-zA-Z0-9_+-]*)\s*\n(?P<body>.*?)\n```\s*$", text, re.DOTALL)
    if fenced:
        text = fenced.group("body").strip()
        return text

    # `tek-satir`
    if text.startswith("`") and text.endswith("`") and text.count("`") == 2:
        text = text[1:-1].strip()
        return text

    # "cift tirnak" veya 'tek tirnak' (sarmalayici)
    if (text.startswith('"') and text.endswith('"')) or (
        text.startswith("'") and text.endswith("'")
    ):
        text = text[1:-1].strip()

    return text


def parse_react(text: str) -> ParsedResponse:
    """ReAct stilinde uretilmis metni yapisal hale getir.



    Notlar:

        * Birden fazla Action blogu varsa hepsi listeye girer.

        * Final Answer varsa actions yine de toplanir; agent.py karari verir.

        * Buyuk/kucuk harf duyarsiz; etiketlerin once gelen bosluklari onemsiz.

    """
    raw = text or ""
    thought_match = _THOUGHT_RE.search(raw)
    thought = thought_match.group("body").strip() if thought_match else None

    # Iki farkli format'i da yakaliyoruz; pozisyona gore sirala ki model'in
    # uretim sirasi korunsun. Ayni pozisyonda klasik bulunduysa fonksiyon
    # cagrisi versiyonunu yoksay (cunku zaten klasik match ediyor demektir).
    raw_matches: list[tuple[int, str, str, str]] = []  # (pos, tool, input, raw_text)

    classic_spans: list[tuple[int, int]] = []
    for m in _ACTION_BLOCK_RE.finditer(raw):
        classic_spans.append((m.start(), m.end()))
        raw_matches.append(
            (m.start(), m.group("tool").strip(), _strip_input(m.group("input")), m.group(0))
        )

    def _overlaps_classic(start: int, end: int) -> bool:
        return any(not (end <= s or start >= e) for s, e in classic_spans)

    for m in _ACTION_FUNCTION_RE.finditer(raw):
        if _overlaps_classic(m.start(), m.end()):
            continue
        raw_matches.append(
            (m.start(), m.group("tool").strip(), _strip_input(m.group("input")), m.group(0))
        )

    raw_matches.sort(key=lambda t: t[0])

    actions: list[ToolCall] = [
        ToolCall(tool=tool, args={"cmd": cmd_input}, raw=raw_text)
        for _, tool, cmd_input, raw_text in raw_matches
    ]

    final_match = _FINAL_RE.search(raw)
    final_answer = final_match.group("body").strip() if final_match else None
    # Final Answer eger ayni cikti icinde Action'dan SONRA geliyorsa, ReAct
    # mantiginda Action'i once isleriz. Bu yuzden eger actions varsa
    # final_answer'i bu turda yoksayariz; agent.py loop'u bir sonraki turda
    # alir. Aksi durumda final_answer dogru kalir.
    if actions and final_answer is not None:
        # Eger final_answer son Action'dan sonra geliyorsa, bos olmadigi
        # surece bilgi olarak tutalim ama is_final dondermeyiz.
        all_action_ends = [
            m.end() for m in _ACTION_BLOCK_RE.finditer(raw)
        ] + [
            m.end() for m in _ACTION_FUNCTION_RE.finditer(raw)
        ]
        last_action_end = max(all_action_ends) if all_action_ends else 0
        final_pos = raw.rfind("Final Answer")
        if final_pos < last_action_end:
            final_answer = None

    return ParsedResponse(
        thought=thought,
        actions=actions,
        final_answer=final_answer,
        raw=raw,
    )


# -----------------------------------------------------------------------------
# Dispatcher
# -----------------------------------------------------------------------------


ToolFn = Callable[[dict[str, str]], str]


class UnknownToolError(RuntimeError):
    pass


class ToolDispatcher:
    """ToolCall -> Observation string. SafetyChecker araligi burada."""

    def __init__(

        self,

        kali: KaliSSH,

        safety: SafetyChecker,

        *,

        logger: Optional["SessionLogger"] = None,

    ):
        self.kali = kali
        self.safety = safety
        self.logger = logger
        self._tools: dict[str, ToolFn] = {
            "run_cmd": self._tool_run_cmd,
        }

    @property
    def available_tools(self) -> tuple[str, ...]:
        return tuple(sorted(self._tools.keys()))

    def register(self, name: str, fn: ToolFn) -> None:
        self._tools[name] = fn

    def dispatch(self, call: ToolCall) -> str:
        if self.logger:
            self.logger.tool_call(call.tool, call.args)
        fn = self._tools.get(call.tool)
        if fn is None:
            msg = (
                f"BLOCKED: bilinmeyen tool '{call.tool}'. "
                f"Kullanilabilir: {', '.join(self.available_tools)}"
            )
            if self.logger:
                self.logger.error("unknown_tool", tool=call.tool)
            return msg
        try:
            return fn(call.args)
        except Exception as exc:  # noqa: BLE001
            if self.logger:
                self.logger.error("tool_exception", tool=call.tool, detail=str(exc))
            return f"ERROR: tool '{call.tool}' istisna firlatti: {exc}"

    # --- tools ----------------------------------------------------------
    def _tool_run_cmd(self, args: dict[str, str]) -> str:
        cmd = (args.get("cmd") or "").strip()
        if not cmd:
            return "BLOCKED: bos komut."

        verdict: Verdict = self.safety.check(cmd)
        if verdict.blocked:
            # Modele net mesaj donelim ki yeniden dene-mesin / siklasmasin.
            return verdict.to_user_message()

        try:
            result = self.kali.run(cmd)
        except SSHConnectionError as exc:
            return f"ERROR: SSH/komut hatasi: {exc}"

        body = result.combined_output()
        prefix = ""
        if verdict.severity == "warn":
            prefix = verdict.to_user_message() + "\n"
        return f"{prefix}exit={result.exit_code} ({result.duration_s:.1f}s)\n{body}"


# -----------------------------------------------------------------------------
# Yardimci: observation'i model girdisine donustur
# -----------------------------------------------------------------------------


def format_observation(obs: str) -> str:
    """Modele Observation: olarak verilecek bicimi olustur.



    `Observation:` Modelfile'da stop token oldugu icin model bunu yazmaya

    calismaz; bizim eklememiz gerekir. Sonuna bos satir koymak modelin

    devam etme niyetini desteklemek icin.

    """
    return f"Observation: {obs.rstrip()}\n"


# -----------------------------------------------------------------------------
# Self-test
# -----------------------------------------------------------------------------


def _self_test() -> int:
    samples = [
        # 0: tek action
        (
            "tek_action",
            "Thought: portlari taramaliyim.\n"
            "Action: run_cmd\n"
            "Action Input: nmap -sV 10.0.0.1\n",
            {"actions": 1, "final": False, "tool": "run_cmd",
             "cmd": "nmap -sV 10.0.0.1"},
        ),
        # 1: kod bloklu input
        (
            "kod_bloklu",
            "Action: run_cmd\n"
            "Action Input: ```bash\nnmap -sV 10.0.0.1\n```\n",
            {"actions": 1, "cmd": "nmap -sV 10.0.0.1"},
        ),
        # 2: tirnakli input
        (
            "tirnakli",
            'Action: run_cmd\nAction Input: "whoami"\n',
            {"actions": 1, "cmd": "whoami"},
        ),
        # 3: final answer
        (
            "final",
            "Thought: bitti.\nFinal Answer: 22/ssh, 80/http acik.\n",
            {"actions": 0, "final": True},
        ),
        # 4: coklu action
        (
            "coklu_action",
            "Action: run_cmd\nAction Input: whoami\n"
            "Action: run_cmd\nAction Input: hostname\n",
            {"actions": 2},
        ),
        # 5: action sonrasi final (agent loop bir sonraki tura biraktirir)
        (
            "action_sonra_final",
            "Action: run_cmd\nAction Input: ls\nFinal Answer: tamam\n",
            {"actions": 1, "final": False},  # is_final False olmali
        ),
        # 6: fonksiyon-cagri stili (model bu sekilde de yazabilir)
        (
            "fonksiyon_cagri",
            "Thought: scan.\nAction: run_cmd(nmap -p- 1.2.3.4 | grep open)\n",
            {"actions": 1, "tool": "run_cmd",
             "cmd": "nmap -p- 1.2.3.4 | grep open"},
        ),
        # 7: fonksiyon-cagri + final (action calismali, final_answer yoksayilir)
        (
            "fonksiyon_sonra_final",
            "Action: run_cmd(whoami)\nFinal Answer: ben kullaniciyim\n",
            {"actions": 1, "final": False, "cmd": "whoami"},
        ),
        # 8: ic parantezli komut (process substitution)
        (
            "ic_paren",
            "Action: run_cmd(echo $(date) > /tmp/x)\n",
            {"actions": 1, "cmd": "echo $(date) > /tmp/x"},
        ),
    ]

    failed = 0
    for name, text, expected in samples:
        p = parse_react(text)
        ok = True
        if "actions" in expected and len(p.actions) != expected["actions"]:
            ok = False
        if "tool" in expected and (not p.actions or p.actions[0].tool != expected["tool"]):
            ok = False
        if "cmd" in expected and (not p.actions or p.actions[0].args.get("cmd") != expected["cmd"]):
            ok = False
        if "final" in expected and bool(p.is_final) != bool(expected["final"]):
            ok = False
        marker = "OK  " if ok else "FAIL"
        print(f"[{marker}] {name:<20} actions={len(p.actions)} "
              f"final={p.is_final} thought={'Y' if p.thought else 'N'}")
        if not ok:
            failed += 1
            print(f"        beklenen: {expected}")
            print(f"        bulunan : actions={[a.short() for a in p.actions]} "
                  f"final={p.final_answer!r}")
    print()
    print(f"{len(samples) - failed}/{len(samples)} parse testi gecti.")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    import sys

    sys.exit(_self_test())