| |
| """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() |
