atlas_agent.py

#!/usr/bin/env python3
"""Atlas Agent Host - güvenli sandbox ortamında çalışan komut aracısı.

Modelin ürettiği tool çağrılarını alır, güvenlik kurallarına göre filtreler
ve yalnızca yetkili işlemleri çalıştırır. Tüm eylemler audit log'a yazılır.

Güvenlik kuralları:
  - Yalnızca ~/atlas-workspace dizini içinde işlem yapılabilir
  - Sistem yolları yasaktır: /etc, /usr, /bin, /boot, /proc, /sys, /dev, /var, /root
  - root/sudo komutları çalıştırılamaz
  - Tehlikeli komutlar (rm -rf /, dd, mkfs, :(){:|:&};:) engellenir
  - Çalıştırılan her komut kullanıcı onayına sunulur (onay modu)
"""

import argparse
import json
import os
import re
import shlex
import subprocess
import sys
import time
import urllib.request
from datetime import datetime
from pathlib import Path

WORKSPACE = Path(os.path.expanduser("~/atlas-workspace")).resolve()
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434/api/chat")
MODEL = os.environ.get("ATLAS_MODEL", "atlas-q4km")
AUDIT_LOG = Path(os.path.expanduser("~/atlas-workspace/audit.log"))

FORBIDDEN_PATHS = [
    "/etc", "/usr", "/bin", "/boot", "/proc", "/sys", "/dev", "/var",
    "/root", "/run", "/sbin", "/lib", "/opt",
]
FORBIDDEN_CMDS = [
    r"\brm\s+-rf\s+/", r"\brm\s+-rf\s+~", r"\bdd\b", r"\bmkfs", r"\bchmod\s+777\s+/",
    r"\{\s*:\|\s*:\s*&\s*\}\s*;", r"\b:\(\)\s*\{", r"\bmount\b", r"\bumount\b",
    r"\bkillall", r"\bkill\b", r"\bshutdown", r"\breboot", r"\bhalt\b",
    r"\bwget.*/dev/sd", r"\bcurl.*/dev/sd", r"\bsudo\b", r"\bsu\b",
]

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_files",
            "description": "Belirtilen dizindeki dosyaları listeler",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "Dizin yolu (yalnızca ~/atlas-workspace içinde)"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Bir dosyanın içeriğini okur",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "Dosya yolu (yalnızca ~/atlas-workspace içinde)"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Bir dosyaya içerik yazar (varsa üzerine yazar)",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Dosya yolu"},
                    "content": {"type": "string", "description": "Yazılacak içerik"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "append_file",
            "description": "Bir dosyanın sonuna içerik ekler",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Dosya yolu"},
                    "content": {"type": "string", "description": "Eklenecek içerik"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "make_dir",
            "description": "Yeni bir dizin oluşturur",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "Oluşturulacak dizin yolu"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "Güvenli bir terminal komutu çalıştırır (salt okunur komutlar önerilir: ls, cat, grep, find, date, uptime, du, df, whoami)",
            "parameters": {
                "type": "object",
                "properties": {"cmd": {"type": "string", "description": "Çalıştırılacak komut"}},
                "required": ["cmd"],
            },
        },
    },
]


def audit(action, detail=""):
    AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    line = f"[{ts}] {action} {detail}"
    with open(AUDIT_LOG, "a", encoding="utf-8") as f:
        f.write(line + "\n")
    print(line)


def resolve_path(path):
    p = Path(os.path.expanduser(path))
    if not p.is_absolute():
        p = WORKSPACE / p
    p = p.resolve()
    if not str(p).startswith(str(WORKSPACE)):
        raise PermissionError(f"ERİŞİM REDDEDİLDİ: '{path}' çalışma alanı dışında")
    for fp in FORBIDDEN_PATHS:
        if str(p) == fp or str(p).startswith(fp + "/"):
            raise PermissionError(f"ERİŞİM REDDEDİLDİ: '{path}' sistem yolu")
    return p


def check_command(cmd):
    for pat in FORBIDDEN_CMDS:
        if re.search(pat, cmd):
            raise PermissionError(f"KOMUT REDDEDİLDİ: '{cmd}' yasaklı kalıp '{pat}' içeriyor")
    for token in shlex.split(cmd):
        tok = token.replace("'", "").replace('"', "").replace("$", "").replace("`", "")
        if tok.startswith("/"):
            if any(tok == fp or tok.startswith(fp + "/") for fp in FORBIDDEN_PATHS):
                raise PermissionError(f"KOMUT REDDEDİLDİ: '{token}' sistem yolu içeriyor")


def run_tool(name, args, confirm):
    try:
        if name == "list_files":
            p = resolve_path(args.get("path", "."))
            entries = sorted(os.listdir(p))
            detail = "\n".join(entries) if entries else "(boş)"
            return {"result": detail}
        elif name == "read_file":
            p = resolve_path(args.get("path", ""))
            if not p.is_file():
                return {"result": f"DOSYA YOK: {p}"}
            content = p.read_text(encoding="utf-8", errors="replace")
            return {"result": content[:4000]}
        elif name == "write_file":
            p = resolve_path(args.get("path", ""))
            if confirm and not ask(f"Dosya yazılacak: {p} - onaylıyor musun?"):
                return {"result": "Kullanıcı reddetti"}
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_text(args.get("content", ""), encoding="utf-8")
            audit("WRITE", str(p))
            return {"result": "Dosya yazıldı"}
        elif name == "append_file":
            p = resolve_path(args.get("path", ""))
            if confirm and not ask(f"Dosyaya eklenecek: {p} - onaylıyor musun?"):
                return {"result": "Kullanıcı reddetti"}
            p.parent.mkdir(parents=True, exist_ok=True)
            with open(p, "a", encoding="utf-8") as f:
                f.write(args.get("content", "") + "\n")
            audit("APPEND", str(p))
            return {"result": "İçerik eklendi"}
        elif name == "make_dir":
            p = resolve_path(args.get("path", ""))
            p.mkdir(parents=True, exist_ok=True)
            audit("MKDIR", str(p))
            return {"result": f"Dizin oluşturuldu: {p}"}
        elif name == "run_command":
            cmd = args.get("cmd", "")
            check_command(cmd)
            if confirm and not ask(f"Komut çalıştırılacak: {cmd} - onaylıyor musun?"):
                return {"result": "Kullanıcı reddetti"}
            audit("CMD", cmd)
            r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30, cwd=WORKSPACE)
            out = (r.stdout or "") + (r.stderr or "")
            return {"result": out[:4000] if out else f"(çıktı yok, exit={r.returncode})"}
        else:
            return {"result": f"Bilinmeyen tool: {name}"}
    except PermissionError as e:
        audit("BLOCK", str(e))
        return {"result": str(e)}
    except Exception as e:
        return {"result": f"HATA: {e}"}


def ask(question):
    try:
        ans = input(f"\n? {question} [y/N]: ").strip().lower()
        return ans in ("y", "yes")
    except (EOFError, KeyboardInterrupt):
        return False


def call_model(messages):
    body = {
        "model": MODEL,
        "messages": messages,
        "stream": False,
        "think": False,
        "tools": TOOLS,
    }
    req = urllib.request.Request(
        OLLAMA_URL,
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=180) as resp:
        return json.loads(resp.read())


def main():
    ap = argparse.ArgumentParser(description="Atlas Agent Host")
    ap.add_argument("istek", nargs="?", help="Tek seferlik istek (interaktif mod kapalıyken)")
    ap.add_argument("--interactive", "-i", action="store_true", help="İnteraktif sohbet modu")
    ap.add_argument("--no-confirm", action="store_true", help="Komut onayı isteme (varsayılan: ister)")
    args = ap.parse_args()

    WORKSPACE.mkdir(parents=True, exist_ok=True)
    audit("START", f"model={MODEL} workspace={WORKSPACE}")

    if args.interactive or not args.istek:
        print("Atlas Agent Host - çıkmak için 'exit' yazın")
        print(f"Çalışma alanı: {WORKSPACE}")
        while True:
            try:
                q = input("\nSen: ")
            except (EOFError, KeyboardInterrupt):
                break
            if q.strip().lower() in ("exit", "quit", "q"):
                break
            messages = [{"role": "user", "content": q}]
            for _ in range(10):
                resp = call_model(messages)
                m = resp.get("message", {})
                tools = m.get("tool_calls") or []
                if not tools:
                    print("\nAtlas:", m.get("content", "(boş)"))
                    break
                messages.append({"role": "assistant", "content": m.get("content") or "", "tool_calls": tools})
                for tc in tools:
                    fn = tc.get("function", {})
                    name = fn.get("name", "")
                    fargs = fn.get("arguments", {})
                    if isinstance(fargs, str):
                        try:
                            fargs = json.loads(fargs)
                        except Exception:
                            fargs = {}
                    result = run_tool(name, fargs, not args.no_confirm)
                    messages.append({"role": "tool", "name": name, "content": json.dumps(result, ensure_ascii=False)})
    else:
        messages = [{"role": "user", "content": args.istek}]
        for _ in range(10):
            resp = call_model(messages)
            m = resp.get("message", {})
            tools = m.get("tool_calls") or []
            if not tools:
                print(m.get("content", "(boş)"))
                break
            messages.append({"role": "assistant", "content": m.get("content") or "", "tool_calls": tools})
            for tc in tools:
                fn = tc.get("function", {})
                name = fn.get("name", "")
                fargs = fn.get("arguments", {})
                if isinstance(fargs, str):
                    try:
                        fargs = json.loads(fargs)
                    except Exception:
                        fargs = {}
                result = run_tool(name, fargs, not args.no_confirm)
                messages.append({"role": "tool", "name": name, "content": json.dumps(result, ensure_ascii=False)})


if __name__ == "__main__":
    main()