mirror of
https://github.com/open-webui/open-terminal.git
synced 2026-07-09 16:09:14 +00:00
68f0fa1cf5
Some checks are pending
Build Docker Image / build (linux/amd64, map[file:Dockerfile.slim name:slim short_tag:slim suffix:-slim]) (push) Waiting to run
Build Docker Image / build (linux/amd64, map[file:Dockerfile name:default short_tag:latest suffix:]) (push) Waiting to run
Build Docker Image / build (linux/amd64, map[file:Dockerfile.alpine name:alpine short_tag:alpine suffix:-alpine]) (push) Waiting to run
Build Docker Image / build (linux/arm64, map[file:Dockerfile name:default short_tag:latest suffix:]) (push) Waiting to run
Build Docker Image / build (linux/arm64, map[file:Dockerfile.alpine name:alpine short_tag:alpine suffix:-alpine]) (push) Waiting to run
Build Docker Image / build (linux/arm64, map[file:Dockerfile.slim name:slim short_tag:slim suffix:-slim]) (push) Waiting to run
Build Docker Image / merge (map[name:alpine short_tag:alpine suffix:-alpine]) (push) Blocked by required conditions
Build Docker Image / merge (map[name:default short_tag:latest suffix:]) (push) Blocked by required conditions
Build Docker Image / merge (map[name:slim short_tag:slim suffix:-slim]) (push) Blocked by required conditions
1834 lines
62 KiB
Python
1834 lines
62 KiB
Python
import asyncio
|
|
import hmac
|
|
from importlib.metadata import version as _pkg_version
|
|
import fnmatch
|
|
import json
|
|
|
|
import aiofiles
|
|
import aiofiles.os
|
|
import os
|
|
import platform
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile, WebSocket, WebSocketDisconnect
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse, Response
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel, Field
|
|
|
|
from open_terminal.env import API_KEY, BINARY_FILE_MIME_PREFIXES, CORS_ALLOWED_ORIGINS, ENABLE_NOTEBOOKS, ENABLE_SYSTEM_PROMPT, ENABLE_TERMINAL, EXECUTE_DESCRIPTION, EXECUTE_TIMEOUT, FILE_BROWSER_ROOT, LOG_DIR, MAX_TERMINAL_SESSIONS, MULTI_USER, OPEN_TERMINAL_INFO, PROCESS_LOG_RETENTION, SESSION_CWD_TTL, SYSTEM_PROMPT, TERMINAL_TERM
|
|
from open_terminal.utils.runner import PipeRunner, ProcessRunner, create_runner
|
|
from open_terminal.utils.fs import UserFS
|
|
|
|
if MULTI_USER:
|
|
from open_terminal.utils.user_isolation import check_environment, resolve_user
|
|
check_environment()
|
|
|
|
if not API_KEY:
|
|
raise SystemExit(
|
|
"\n\033[91m"
|
|
" OPEN_TERMINAL_API_KEY is required.\n"
|
|
" Set via environment variable or --api-key flag.\n"
|
|
"\033[0m"
|
|
)
|
|
|
|
try:
|
|
import fcntl
|
|
import pty
|
|
import struct
|
|
import subprocess
|
|
import termios
|
|
|
|
_PTY_AVAILABLE = True
|
|
except ImportError:
|
|
_PTY_AVAILABLE = False # Windows
|
|
|
|
|
|
def get_system_info() -> str:
|
|
"""Gather runtime system metadata for the OpenAPI description."""
|
|
shell = os.environ.get("SHELL", "/bin/sh")
|
|
user_part = f" as user '{os.getenv('USER', 'unknown')}'" if not MULTI_USER else ""
|
|
return (
|
|
f"This system is running {platform.system()} {platform.release()} ({platform.machine()}) "
|
|
f"on {socket.gethostname()}{user_part} with {shell}. "
|
|
f"Python {sys.version.split()[0]} is available."
|
|
)
|
|
|
|
|
|
def _system_prompt_variables() -> dict[str, str]:
|
|
return {
|
|
"os": platform.system(),
|
|
"kernel": platform.release(),
|
|
"arch": platform.machine(),
|
|
"hostname": socket.gethostname(),
|
|
"user": "" if MULTI_USER else os.getenv("USER", "unknown"),
|
|
"shell": os.environ.get("SHELL", "/bin/sh"),
|
|
"python_version": sys.version.split()[0],
|
|
"home": os.path.expanduser("~"),
|
|
}
|
|
|
|
|
|
def _expand_system_prompt_template(template: str) -> str:
|
|
values = _system_prompt_variables()
|
|
|
|
def replace(match: re.Match) -> str:
|
|
key = match.group(1).strip()
|
|
return values.get(key, match.group(0))
|
|
|
|
return re.sub(r"\{\{\s*([a-zA-Z0-9_]+)\s*\}\}", replace, template)
|
|
|
|
|
|
def get_system_prompt() -> str:
|
|
"""Build a default system prompt for LLM integration."""
|
|
if SYSTEM_PROMPT:
|
|
return _expand_system_prompt_template(SYSTEM_PROMPT)
|
|
|
|
shell = os.environ.get("SHELL", "/bin/sh")
|
|
user_part = f" as user '{os.getenv('USER', 'unknown')}'" if not MULTI_USER else ""
|
|
|
|
prompt = (
|
|
f"You have access to a computer running {platform.system()} {platform.release()} ({platform.machine()}) "
|
|
f'on host "{socket.gethostname()}"{user_part} with {shell}. '
|
|
f"Python {sys.version.split()[0]} is available.\n\n"
|
|
"Use your tools to directly interact with the system \u2014 run commands, read and write files, "
|
|
"and search the filesystem. "
|
|
"Prefer verifying the current state before making changes. "
|
|
"When running commands, check the output to confirm success. "
|
|
"If a command produces no output, that typically means it succeeded."
|
|
)
|
|
|
|
if OPEN_TERMINAL_INFO:
|
|
prompt += f"\n\n{OPEN_TERMINAL_INFO}"
|
|
|
|
return prompt
|
|
|
|
|
|
_EXECUTE_DESCRIPTION = (
|
|
"Run a shell command in the background and return a command ID.\n\n"
|
|
+ get_system_info()
|
|
)
|
|
if EXECUTE_DESCRIPTION:
|
|
_EXECUTE_DESCRIPTION += "\n\n" + EXECUTE_DESCRIPTION
|
|
|
|
bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def verify_api_key(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme),
|
|
):
|
|
if not API_KEY:
|
|
return
|
|
if not credentials or not hmac.compare_digest(credentials.credentials, API_KEY):
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
|
|
def get_filesystem(request: Request) -> UserFS:
|
|
"""Build a :class:`UserFS` scoped to the requesting user.
|
|
|
|
When multi-user mode is active and the ``X-User-Id`` header is present,
|
|
returns a ``UserFS`` that routes all I/O through ``sudo -u``.
|
|
Otherwise returns a plain ``UserFS`` using stdlib.
|
|
"""
|
|
if not MULTI_USER:
|
|
return UserFS()
|
|
user_id = request.headers.get("x-user-id")
|
|
if not user_id:
|
|
return UserFS()
|
|
username, home = resolve_user(user_id)
|
|
return UserFS(username=username, home=home)
|
|
|
|
|
|
def get_file_browser_root(fs: UserFS) -> dict[str, str] | None:
|
|
configured = FILE_BROWSER_ROOT.strip()
|
|
mode = configured.lower()
|
|
if mode in ("", "home"):
|
|
return {"path": fs.home, "label": "Home"}
|
|
if mode == "filesystem":
|
|
return None
|
|
|
|
path = configured.replace("{{home}}", fs.home)
|
|
if path == "~":
|
|
path = fs.home
|
|
elif path.startswith("~/"):
|
|
path = os.path.join(fs.home, path[2:])
|
|
|
|
root_path = fs.resolve_path(path)
|
|
label = "Home" if root_path == fs.home else os.path.basename(root_path) or root_path
|
|
return {"path": root_path, "label": label}
|
|
|
|
|
|
app = FastAPI(
|
|
title="Open Terminal",
|
|
description="A remote terminal API.",
|
|
version=_pkg_version("open-terminal"),
|
|
)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=[o.strip() for o in CORS_ALLOWED_ORIGINS.split(",")],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.exception_handler(PermissionError)
|
|
async def permission_error_handler(request: Request, exc: PermissionError):
|
|
return JSONResponse(status_code=403, content={"detail": str(exc)})
|
|
|
|
|
|
@app.middleware("http")
|
|
async def normalize_null_query_params(request: Request, call_next):
|
|
"""Strip query parameters whose value is the literal string 'null'."""
|
|
from urllib.parse import urlencode
|
|
|
|
raw_params = request.query_params.multi_items()
|
|
cleaned = [(k, v) for k, v in raw_params if v.lower() != "null"]
|
|
if len(cleaned) != len(raw_params):
|
|
request.scope["query_string"] = urlencode(cleaned).encode("utf-8")
|
|
return await call_next(request)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ExecRequest(BaseModel):
|
|
command: str = Field(
|
|
...,
|
|
description="Shell command to execute. Supports chaining (&&, ||, ;), pipes (|), and redirections.",
|
|
json_schema_extra={"examples": ["echo hello", "ls -la && whoami"]},
|
|
)
|
|
cwd: Optional[str] = Field(
|
|
None,
|
|
description="Working directory for the command. Defaults to the server's current directory if not set.",
|
|
)
|
|
env: Optional[dict[str, str]] = Field(
|
|
None,
|
|
description="Extra environment variables merged into the subprocess environment.",
|
|
)
|
|
|
|
|
|
class InputRequest(BaseModel):
|
|
input: str = Field(
|
|
...,
|
|
description="Text to send to the process's stdin. Include newline characters as needed.",
|
|
)
|
|
|
|
|
|
class WriteRequest(BaseModel):
|
|
path: str = Field(
|
|
...,
|
|
description="Absolute or relative path to write to. Parent directories are created automatically.",
|
|
)
|
|
content: str = Field(
|
|
...,
|
|
description="Text content to write to the file.",
|
|
)
|
|
|
|
|
|
class ReplacementChunk(BaseModel):
|
|
target: str = Field(
|
|
...,
|
|
description="Exact string to find. Must match precisely, including whitespace.",
|
|
)
|
|
replacement: str = Field(
|
|
...,
|
|
description="Content to replace the target with.",
|
|
)
|
|
start_line: Optional[int] = Field(
|
|
None,
|
|
description="Narrow the search to lines at or after this (1-indexed).",
|
|
ge=1,
|
|
)
|
|
end_line: Optional[int] = Field(
|
|
None,
|
|
description="Narrow the search to lines at or before this (1-indexed).",
|
|
ge=1,
|
|
)
|
|
allow_multiple: bool = Field(
|
|
False,
|
|
description="If true, replaces all occurrences. If false, errors when multiple matches are found.",
|
|
)
|
|
|
|
|
|
class MkdirRequest(BaseModel):
|
|
path: str = Field(
|
|
...,
|
|
description="Directory path to create. Parent directories are created automatically.",
|
|
)
|
|
|
|
|
|
class MoveRequest(BaseModel):
|
|
source: str = Field(
|
|
...,
|
|
description="Path to the file or directory to move.",
|
|
)
|
|
destination: str = Field(
|
|
...,
|
|
description="Destination path (new location).",
|
|
)
|
|
|
|
|
|
class ReplaceRequest(BaseModel):
|
|
path: str = Field(
|
|
...,
|
|
description="Path to the file to modify.",
|
|
)
|
|
replacements: list[ReplacementChunk] = Field(
|
|
...,
|
|
description="List of find-and-replace operations to apply sequentially.",
|
|
)
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background process management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class BackgroundProcess:
|
|
id: str
|
|
command: str
|
|
runner: ProcessRunner
|
|
status: str = "running"
|
|
exit_code: Optional[int] = None
|
|
log_task: Optional[asyncio.Task] = field(default=None, repr=False)
|
|
finished_at: Optional[float] = field(default=None, repr=False)
|
|
log_path: Optional[str] = field(default=None, repr=False)
|
|
|
|
|
|
_processes: dict[str, BackgroundProcess] = {}
|
|
_EXPIRY_SECONDS = 300 # auto-clean finished processes after 5 min
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-session working directory tracking
|
|
# ---------------------------------------------------------------------------
|
|
# Maps session_id → (absolute_cwd_path, last_accessed_timestamp).
|
|
# Replaces the old os.chdir() approach which was process-global and unsafe
|
|
# with concurrent sessions.
|
|
_session_cwds: dict[str, tuple[str, float]] = {}
|
|
|
|
|
|
|
|
def _expire_session_cwds():
|
|
"""Remove session cwd entries that haven't been accessed within the TTL."""
|
|
now = time.time()
|
|
expired = [sid for sid, (_, ts) in _session_cwds.items() if now - ts > SESSION_CWD_TTL]
|
|
for sid in expired:
|
|
del _session_cwds[sid]
|
|
|
|
|
|
def _get_session_cwd(session_id: str | None, fs: "UserFS") -> str:
|
|
"""Return the tracked cwd for *session_id*, or ``fs.home`` as default."""
|
|
_expire_session_cwds()
|
|
if session_id and session_id in _session_cwds:
|
|
cwd, _ = _session_cwds[session_id]
|
|
_session_cwds[session_id] = (cwd, time.time()) # refresh TTL
|
|
return cwd
|
|
return fs.home
|
|
|
|
|
|
def _set_session_cwd(session_id: str | None, path: str):
|
|
"""Store a session's cwd. No-op if *session_id* is ``None``."""
|
|
if session_id:
|
|
_session_cwds[session_id] = (path, time.time())
|
|
|
|
|
|
from open_terminal.utils.log import log_process, read_log
|
|
|
|
|
|
|
|
|
|
def _cleanup_expired():
|
|
"""Remove finished processes that have expired.
|
|
|
|
Also deletes log files older than *LOG_RETENTION_SECONDS*.
|
|
"""
|
|
now = time.time()
|
|
expired = [
|
|
process_id
|
|
for process_id, background_process in _processes.items()
|
|
if background_process.finished_at
|
|
and now - background_process.finished_at > _EXPIRY_SECONDS
|
|
]
|
|
for process_id in expired:
|
|
bp = _processes.pop(process_id)
|
|
# Delete the log file if it has exceeded the retention period.
|
|
if (
|
|
bp.log_path
|
|
and bp.finished_at
|
|
and now - bp.finished_at > PROCESS_LOG_RETENTION
|
|
):
|
|
try:
|
|
os.remove(bp.log_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _get_process(process_id: str) -> BackgroundProcess:
|
|
_cleanup_expired()
|
|
background_process = _processes.get(process_id)
|
|
if not background_process:
|
|
raise HTTPException(status_code=404, detail="Process not found")
|
|
return background_process
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Health
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get(
|
|
"/health",
|
|
include_in_schema=False,
|
|
operation_id="health_check",
|
|
summary="Health check",
|
|
description="Returns service status. No authentication required.",
|
|
)
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config (capability discovery)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get(
|
|
"/api/config",
|
|
include_in_schema=False,
|
|
)
|
|
async def get_config():
|
|
"""Return server feature flags for client-side discovery."""
|
|
return {
|
|
"features": {
|
|
"terminal": ENABLE_TERMINAL,
|
|
"notebooks": ENABLE_NOTEBOOKS,
|
|
"system": ENABLE_SYSTEM_PROMPT,
|
|
},
|
|
}
|
|
|
|
|
|
if ENABLE_SYSTEM_PROMPT:
|
|
|
|
@app.get(
|
|
"/system",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def get_system():
|
|
"""Return a system prompt for LLM integration."""
|
|
return {"prompt": get_system_prompt()}
|
|
|
|
|
|
if OPEN_TERMINAL_INFO:
|
|
|
|
@app.get(
|
|
"/info",
|
|
operation_id="get_info",
|
|
summary="Get environment info",
|
|
description="Return operator-provided information about this environment. Use this to understand the system you are working with.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def get_info():
|
|
return {"info": OPEN_TERMINAL_INFO}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Files
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get(
|
|
"/files/cwd",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def get_cwd(
|
|
http_request: Request,
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
response = {
|
|
"cwd": _get_session_cwd(session_id, fs),
|
|
"home": fs.home,
|
|
}
|
|
root = get_file_browser_root(fs)
|
|
if root:
|
|
response["root"] = root
|
|
return response
|
|
|
|
|
|
@app.post(
|
|
"/files/cwd",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def set_cwd(
|
|
http_request: Request,
|
|
request: MkdirRequest,
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
target = fs.resolve_path(request.path)
|
|
if not fs.username and not await fs.isdir(target):
|
|
raise HTTPException(status_code=404, detail="Directory not found")
|
|
_set_session_cwd(session_id, target)
|
|
return {"cwd": target}
|
|
|
|
|
|
@app.get(
|
|
"/files/list",
|
|
operation_id="list_files",
|
|
summary="List directory contents",
|
|
description="Return a structured listing of files and directories at the given path.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Directory not found."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def list_files(
|
|
http_request: Request,
|
|
directory: str = Query(".", description="Directory path to list."),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(directory, cwd=session_cwd)
|
|
if not await fs.isdir(target):
|
|
raise HTTPException(status_code=404, detail="Directory not found")
|
|
entries = await fs.listdir(target)
|
|
return {"dir": target, "entries": entries}
|
|
|
|
|
|
@app.get(
|
|
"/files/read",
|
|
operation_id="read_file",
|
|
summary="Read a file",
|
|
description="Read a file and return its contents. Supports text files and images (PNG, JPEG, WebP, etc.). For text files you can optionally request a specific line range. Images are returned as binary so you can view and analyze them directly. Use display_file to show a file to the user.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "File not found."},
|
|
415: {"description": "Unsupported binary file type."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def read_file(
|
|
http_request: Request,
|
|
path: str = Query(..., description="Path to the file to read."),
|
|
start_line: Optional[int] = Query(
|
|
None, description="First line to return (1-indexed, inclusive). Defaults to the beginning of the file.", ge=1
|
|
),
|
|
end_line: Optional[int] = Query(
|
|
None, description="Last line to return (1-indexed, inclusive). Defaults to the end of the file.", ge=1
|
|
),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(path, cwd=session_cwd)
|
|
if not await fs.isfile(target):
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
|
|
try:
|
|
content = await fs.read_text(target)
|
|
lines = content.splitlines(keepends=True)
|
|
except (UnicodeDecodeError, ValueError):
|
|
import mimetypes
|
|
|
|
raw = await fs.read(target)
|
|
mime, _ = mimetypes.guess_type(target)
|
|
mime = mime or "application/octet-stream"
|
|
|
|
# Try document text extraction (PDF, Office, OpenDocument, etc.)
|
|
from open_terminal.utils.documents import EXTRACTORS
|
|
|
|
for ext_mime, ext_suffix, extractor in EXTRACTORS:
|
|
if (ext_mime and mime == ext_mime) or (
|
|
ext_suffix and target.lower().endswith(ext_suffix)
|
|
):
|
|
text = await asyncio.to_thread(extractor, target)
|
|
lines = text.splitlines(keepends=True)
|
|
start = (start_line or 1) - 1
|
|
end = end_line or len(lines)
|
|
return {
|
|
"path": target,
|
|
"total_lines": len(lines),
|
|
"content": "".join(lines[start:end]),
|
|
}
|
|
|
|
# Return raw binary for allowed mime type prefixes (e.g. image/*)
|
|
if any(mime.startswith(prefix) for prefix in BINARY_FILE_MIME_PREFIXES):
|
|
return Response(content=raw, media_type=mime)
|
|
|
|
# Other binary files: reject (LLMs can't interpret raw bytes)
|
|
raise HTTPException(
|
|
status_code=415,
|
|
detail=f"Unsupported binary file type: {mime} ({len(raw)} bytes)",
|
|
)
|
|
|
|
start = (start_line or 1) - 1
|
|
end = end_line or len(lines)
|
|
return {
|
|
"path": target,
|
|
"total_lines": len(lines),
|
|
"content": "".join(lines[start:end]),
|
|
}
|
|
|
|
|
|
@app.get(
|
|
"/files/display",
|
|
operation_id="display_file",
|
|
summary="Display a file to the user",
|
|
description="Open a file in the user's file viewer so they can see it. Use this when the user wants to view or look at a file. This does not return file content to you — use read_file if you need to read the content yourself.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def display_file(
|
|
http_request: Request,
|
|
path: str = Query(..., description="Absolute path to the file to display."),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
"""Signal that a file should be displayed to the user.
|
|
|
|
This endpoint does not serve file content itself. It returns the resolved
|
|
path and whether the file exists. The consuming client is responsible for
|
|
intercepting this response and presenting the file in its own UI (e.g.
|
|
opening a preview pane, launching a viewer, etc.).
|
|
"""
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(path, cwd=session_cwd)
|
|
exists = await fs.isfile(target)
|
|
return {"path": target, "exists": exists}
|
|
|
|
|
|
@app.get(
|
|
"/files/view",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def view_file(
|
|
path: str = Query(..., description="Path to the file to view."),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
"""Return raw file bytes with the appropriate Content-Type.
|
|
|
|
Unlike read_file (which is designed for LLM consumption and restricts
|
|
binary types), this endpoint serves any file as-is for UI previewing.
|
|
"""
|
|
target = fs.resolve_path(path)
|
|
if not await fs.isfile(target):
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
|
|
import mimetypes
|
|
|
|
mime, _ = mimetypes.guess_type(target)
|
|
mime = mime or "application/octet-stream"
|
|
raw = await fs.read(target)
|
|
return Response(content=raw, media_type=mime)
|
|
|
|
|
|
@app.get(
|
|
"/files/serve/{path:path}",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def serve_file(path: str, fs: UserFS = Depends(get_filesystem)):
|
|
"""Path-based alias for view_file — enables relative URL resolution in iframes."""
|
|
return await view_file(path=f"/{path}", fs=fs)
|
|
|
|
|
|
@app.post(
|
|
"/files/write",
|
|
operation_id="write_file",
|
|
summary="Write a file",
|
|
description="Write text content to a file. Creates parent directories automatically. Overwrites if the file already exists.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def write_file(http_request: Request, request: WriteRequest, fs: UserFS = Depends(get_filesystem)):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(request.path, cwd=session_cwd)
|
|
try:
|
|
await fs.write(target, request.content)
|
|
except (OSError, subprocess.CalledProcessError) as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return {"path": target, "size": len(request.content.encode())}
|
|
|
|
|
|
@app.post(
|
|
"/files/mkdir",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def mkdir(request: MkdirRequest, fs: UserFS = Depends(get_filesystem)):
|
|
target = fs.resolve_path(request.path)
|
|
try:
|
|
await fs.mkdir(target)
|
|
except (OSError, subprocess.CalledProcessError) as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return {"path": target}
|
|
|
|
|
|
@app.delete(
|
|
"/files/delete",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def delete_entry(
|
|
path: str = Query(..., description="Path to delete."),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
target = fs.resolve_path(path)
|
|
if not await fs.exists(target):
|
|
raise HTTPException(status_code=404, detail="Path not found")
|
|
is_dir = await fs.isdir(target)
|
|
try:
|
|
await fs.remove(target)
|
|
except (OSError, subprocess.CalledProcessError) as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return {"path": target, "type": "directory" if is_dir else "file"}
|
|
|
|
|
|
@app.post(
|
|
"/files/move",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def move_entry(request: MoveRequest, fs: UserFS = Depends(get_filesystem)):
|
|
source = fs.resolve_path(request.source)
|
|
destination = fs.resolve_path(request.destination)
|
|
|
|
if not await fs.exists(source):
|
|
raise HTTPException(status_code=404, detail="Source path not found")
|
|
|
|
dest_parent = os.path.dirname(destination)
|
|
if not await fs.isdir(dest_parent):
|
|
raise HTTPException(status_code=400, detail="Destination parent directory not found")
|
|
|
|
if await fs.exists(destination):
|
|
raise HTTPException(status_code=409, detail="Destination already exists")
|
|
|
|
try:
|
|
await fs.move(source, destination)
|
|
except (OSError, subprocess.CalledProcessError) as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return {"source": source, "destination": destination}
|
|
|
|
|
|
@app.post(
|
|
"/files/replace",
|
|
operation_id="replace_file_content",
|
|
summary="Replace content in a file",
|
|
description="Find and replace exact strings in a file. Supports multiple replacements in one call with optional line range narrowing.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "File not found."},
|
|
400: {"description": "Target string not found or ambiguous match."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def replace_file_content(http_request: Request, request: ReplaceRequest, fs: UserFS = Depends(get_filesystem)):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(request.path, cwd=session_cwd)
|
|
if not await fs.isfile(target):
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
|
|
try:
|
|
content = await fs.read_text(target)
|
|
except OSError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
for chunk in request.replacements:
|
|
if chunk.start_line or chunk.end_line:
|
|
lines = content.splitlines(keepends=True)
|
|
start = (chunk.start_line or 1) - 1
|
|
end = chunk.end_line or len(lines)
|
|
search_region = "".join(lines[start:end])
|
|
else:
|
|
search_region = content
|
|
|
|
count = search_region.count(chunk.target)
|
|
if count == 0:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Target string not found: {chunk.target[:100]!r}",
|
|
)
|
|
if count > 1 and not chunk.allow_multiple:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Found {count} occurrences of target string but allow_multiple is false",
|
|
)
|
|
|
|
if chunk.start_line or chunk.end_line:
|
|
new_region = search_region.replace(chunk.target, chunk.replacement)
|
|
lines[start:end] = [new_region]
|
|
content = "".join(lines)
|
|
else:
|
|
content = content.replace(chunk.target, chunk.replacement)
|
|
|
|
try:
|
|
await fs.write(target, content)
|
|
except OSError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
return {"path": target, "size": len(content.encode())}
|
|
|
|
|
|
@app.get(
|
|
"/files/grep",
|
|
operation_id="grep_search",
|
|
summary="Search file contents",
|
|
description="Search for a text pattern across files in a directory. Returns structured matches with file paths, line numbers, and matching lines. Skips binary files.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Search path not found."},
|
|
400: {"description": "Invalid regex pattern."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def grep_search(
|
|
http_request: Request,
|
|
query: str = Query(..., description="Text or regex pattern to search for."),
|
|
path: str = Query(".", description="Directory or file to search in."),
|
|
regex: bool = Query(True, description="Use regex. Set false for literal search."),
|
|
case_insensitive: bool = Query(
|
|
False, description="Perform case-insensitive matching."
|
|
),
|
|
include: Optional[list[str]] = Query(
|
|
None,
|
|
description="Glob patterns to filter files (e.g. '*.py'). Files must match at least one pattern.",
|
|
),
|
|
match_per_line: bool = Query(
|
|
True,
|
|
description="If true, return each matching line with line numbers. If false, return only the names of matching files.",
|
|
),
|
|
max_results: int = Query(
|
|
50, description="Maximum number of matches to return.", ge=1, le=500
|
|
),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(path, cwd=session_cwd)
|
|
if not await aiofiles.os.path.exists(target):
|
|
raise HTTPException(status_code=404, detail="Search path not found")
|
|
|
|
flags = re.IGNORECASE if case_insensitive else 0
|
|
if regex:
|
|
try:
|
|
pattern = re.compile(query, flags)
|
|
except re.error as exc:
|
|
raise HTTPException(status_code=400, detail=f"Invalid regex: {exc}")
|
|
else:
|
|
pattern = re.compile(re.escape(query), flags)
|
|
|
|
def _search_sync():
|
|
def _matches_include(filename: str) -> bool:
|
|
if not include:
|
|
return True
|
|
return any(fnmatch.fnmatch(filename, glob) for glob in include)
|
|
|
|
matches = []
|
|
truncated = False
|
|
|
|
def _search_file(file_path: str):
|
|
nonlocal truncated
|
|
if truncated:
|
|
return
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8", errors="strict") as f:
|
|
for line_number, line in enumerate(f, 1):
|
|
if pattern.search(line):
|
|
if match_per_line:
|
|
matches.append(
|
|
{
|
|
"file": file_path,
|
|
"line": line_number,
|
|
"content": line.rstrip("\n\r"),
|
|
}
|
|
)
|
|
if len(matches) >= max_results:
|
|
truncated = True
|
|
return
|
|
else:
|
|
matches.append({"file": file_path})
|
|
if len(matches) >= max_results:
|
|
truncated = True
|
|
return # one match per file is enough
|
|
except (UnicodeDecodeError, ValueError, OSError):
|
|
pass # skip binary or unreadable files
|
|
|
|
if os.path.isfile(target):
|
|
_search_file(target)
|
|
else:
|
|
for dirpath, dirnames, filenames in os.walk(target):
|
|
# Prune directories belonging to other users.
|
|
dirnames[:] = [
|
|
d for d in dirnames
|
|
if fs.is_path_allowed(os.path.join(dirpath, d))
|
|
]
|
|
if truncated:
|
|
break
|
|
for filename in sorted(filenames):
|
|
if not _matches_include(filename):
|
|
continue
|
|
full = os.path.join(dirpath, filename)
|
|
if not fs.is_path_allowed(full):
|
|
continue
|
|
_search_file(full)
|
|
|
|
return matches, truncated
|
|
|
|
matches, truncated = await asyncio.to_thread(_search_sync)
|
|
return {
|
|
"query": query,
|
|
"path": target,
|
|
"matches": matches,
|
|
"truncated": truncated,
|
|
}
|
|
|
|
|
|
@app.get(
|
|
"/files/glob",
|
|
operation_id="glob_search",
|
|
summary="Search files by name",
|
|
description="Search for files and subdirectories by name within a specified directory using glob patterns. Results will include the relative path, type, size, and modification time.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Search directory not found."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def glob_search(
|
|
http_request: Request,
|
|
pattern: str = Query(..., description="Glob pattern to search for (e.g. '*.py')."),
|
|
path: str = Query(".", description="Directory to search within."),
|
|
exclude: Optional[list[str]] = Query(
|
|
None, description="Glob patterns to exclude from search results."
|
|
),
|
|
type: Optional[str] = Query(
|
|
"any",
|
|
description="Type filter: 'file', 'directory', or 'any'.",
|
|
pattern="^(file|directory|any)$",
|
|
),
|
|
max_results: int = Query(
|
|
50, description="Maximum number of matches to return.", ge=1, le=500
|
|
),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
target = fs.resolve_path(path, cwd=session_cwd)
|
|
if not await aiofiles.os.path.isdir(target):
|
|
raise HTTPException(status_code=404, detail="Search directory not found")
|
|
|
|
def _glob_sync():
|
|
matches = []
|
|
truncated = False
|
|
|
|
for dirpath, dirnames, filenames in os.walk(target):
|
|
if truncated:
|
|
break
|
|
|
|
# Prune directories belonging to other users.
|
|
dirnames[:] = [
|
|
d for d in dirnames
|
|
if fs.is_path_allowed(os.path.join(dirpath, d))
|
|
]
|
|
|
|
entries = []
|
|
if type in ("any", "directory"):
|
|
entries.extend([(d, "directory") for d in dirnames])
|
|
if type in ("any", "file"):
|
|
entries.extend([(f, "file") for f in filenames])
|
|
|
|
for name, entry_type in sorted(entries, key=lambda x: x[0]):
|
|
if truncated:
|
|
break
|
|
|
|
full_path = os.path.join(dirpath, name)
|
|
rel_path = os.path.relpath(full_path, target)
|
|
|
|
# Check inclusion pattern
|
|
if not fnmatch.fnmatch(name, pattern) and not fnmatch.fnmatch(
|
|
rel_path, pattern
|
|
):
|
|
continue
|
|
|
|
# Check exclusion patterns
|
|
if exclude and any(
|
|
fnmatch.fnmatch(name, excl) or fnmatch.fnmatch(rel_path, excl)
|
|
for excl in exclude
|
|
):
|
|
continue
|
|
|
|
try:
|
|
file_stat = os.stat(full_path)
|
|
matches.append(
|
|
{
|
|
"path": rel_path,
|
|
"type": entry_type,
|
|
"size": file_stat.st_size,
|
|
"modified": file_stat.st_mtime,
|
|
}
|
|
)
|
|
|
|
if len(matches) >= max_results:
|
|
truncated = True
|
|
break
|
|
except OSError:
|
|
pass
|
|
|
|
return matches, truncated
|
|
|
|
matches, truncated = await asyncio.to_thread(_glob_sync)
|
|
return {
|
|
"pattern": pattern,
|
|
"path": target,
|
|
"matches": matches,
|
|
"truncated": truncated,
|
|
}
|
|
|
|
|
|
|
|
|
|
@app.post(
|
|
"/files/upload",
|
|
include_in_schema=False,
|
|
operation_id="upload_file",
|
|
summary="Upload a file",
|
|
description="Save a file to the specified path via multipart form data.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def upload_file(
|
|
directory: str = Query(..., description="Destination directory for the file."),
|
|
file: UploadFile = File(
|
|
..., description="The file to upload."
|
|
),
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
content = await file.read()
|
|
filename = os.path.basename(file.filename or "upload")
|
|
|
|
directory = fs.resolve_path(directory)
|
|
path = os.path.normpath(os.path.join(directory, filename))
|
|
|
|
try:
|
|
await fs.mkdir(directory)
|
|
await fs.write_bytes(path, content)
|
|
except PermissionError as e:
|
|
raise HTTPException(status_code=403, detail=str(e))
|
|
except OSError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
return {"path": path, "size": len(content)}
|
|
|
|
|
|
class ArchiveRequest(BaseModel):
|
|
paths: list[str] = Field(
|
|
...,
|
|
description="List of file or directory paths to include in the ZIP archive.",
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/files/archive",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def archive_paths(
|
|
request: ArchiveRequest,
|
|
fs: UserFS = Depends(get_filesystem),
|
|
):
|
|
"""Bundle files and/or directories into a single ZIP archive."""
|
|
import io
|
|
import zipfile
|
|
|
|
if not request.paths:
|
|
raise HTTPException(status_code=400, detail="No paths provided")
|
|
|
|
resolved = []
|
|
for p in request.paths:
|
|
target = fs.resolve_path(p)
|
|
if not await fs.exists(target):
|
|
raise HTTPException(status_code=404, detail=f"Path not found: {p}")
|
|
resolved.append(target)
|
|
|
|
# Derive a meaningful archive name from the input paths.
|
|
if len(resolved) == 1:
|
|
archive_name = os.path.basename(resolved[0].rstrip("/\\")) or "archive"
|
|
else:
|
|
archive_name = "download"
|
|
|
|
def _build_zip() -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for target in resolved:
|
|
if os.path.isfile(target):
|
|
zf.write(target, os.path.basename(target))
|
|
elif os.path.isdir(target):
|
|
dirname = os.path.basename(target.rstrip("/\\")) or "dir"
|
|
for dirpath, dirnames, filenames in os.walk(target):
|
|
dirnames[:] = [
|
|
d for d in dirnames
|
|
if fs.is_path_allowed(os.path.join(dirpath, d))
|
|
]
|
|
for fname in filenames:
|
|
full = os.path.join(dirpath, fname)
|
|
if not fs.is_path_allowed(full):
|
|
continue
|
|
arcname = os.path.join(
|
|
dirname, os.path.relpath(full, target)
|
|
)
|
|
zf.write(full, arcname)
|
|
return buf.getvalue()
|
|
|
|
data = await asyncio.to_thread(_build_zip)
|
|
return Response(
|
|
content=data,
|
|
media_type="application/zip",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{archive_name}.zip"',
|
|
},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get(
|
|
"/execute",
|
|
operation_id="list_processes",
|
|
summary="List running commands",
|
|
description="Returns a list of all tracked background processes, including running, done, and killed.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def list_processes():
|
|
_cleanup_expired()
|
|
return [
|
|
{
|
|
"id": background_process.id,
|
|
"command": background_process.command,
|
|
"status": background_process.status,
|
|
"exit_code": background_process.exit_code,
|
|
"log_path": background_process.log_path,
|
|
}
|
|
for background_process in _processes.values()
|
|
]
|
|
|
|
|
|
@app.post(
|
|
"/execute",
|
|
operation_id="run_command",
|
|
summary="Execute a command",
|
|
description=_EXECUTE_DESCRIPTION,
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def execute(
|
|
http_request: Request,
|
|
request: ExecRequest,
|
|
wait: Optional[float] = Query(
|
|
None,
|
|
description="Seconds to wait for the command to finish before returning. If the command completes in time, output is included inline. Null to return immediately.",
|
|
ge=0,
|
|
le=300,
|
|
),
|
|
tail: Optional[int] = Query(
|
|
None,
|
|
description="Return only the last N output entries. Useful to limit response size when only recent output matters.",
|
|
ge=1,
|
|
),
|
|
):
|
|
fs = get_filesystem(http_request)
|
|
session_id = http_request.headers.get("x-session-id")
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
cwd = fs.resolve_path(request.cwd, cwd=session_cwd) if request.cwd else (session_cwd or fs.home)
|
|
|
|
subprocess_env = {**os.environ, **request.env} if request.env else None
|
|
runner = await create_runner(
|
|
request.command, cwd, subprocess_env, run_as_user=fs.username
|
|
)
|
|
|
|
process_id = time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6]
|
|
log_path = os.path.join(LOG_DIR, "processes", f"{process_id}.jsonl")
|
|
background_process = BackgroundProcess(
|
|
id=process_id, command=request.command, runner=runner, log_path=log_path
|
|
)
|
|
background_process.log_task = asyncio.create_task(log_process(background_process))
|
|
_processes[process_id] = background_process
|
|
|
|
if wait is None and EXECUTE_TIMEOUT:
|
|
wait = EXECUTE_TIMEOUT
|
|
if wait is not None:
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.shield(background_process.log_task), timeout=wait
|
|
)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
output, next_offset, truncated = await read_log(
|
|
background_process.log_path, offset=0, tail=tail
|
|
)
|
|
|
|
return {
|
|
"id": process_id,
|
|
"command": request.command,
|
|
"status": background_process.status,
|
|
"exit_code": background_process.exit_code,
|
|
"output": output,
|
|
"truncated": truncated,
|
|
"next_offset": next_offset,
|
|
"log_path": background_process.log_path,
|
|
}
|
|
|
|
|
|
@app.get(
|
|
"/execute/{process_id}/status",
|
|
operation_id="get_process_status",
|
|
summary="Get command status and output",
|
|
description="Returns new output since the last poll, process status, and exit code. Output is drained on read to keep memory bounded.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Process not found."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def get_status(
|
|
process_id: str,
|
|
wait: Optional[float] = Query(
|
|
None,
|
|
description="Seconds to wait for the process to finish before returning. Returns early if the process exits. Null to return immediately.",
|
|
ge=0,
|
|
le=300,
|
|
),
|
|
offset: int = Query(
|
|
0,
|
|
description="Number of output entries to skip. Use next_offset from the previous response to get only new output.",
|
|
ge=0,
|
|
),
|
|
tail: Optional[int] = Query(
|
|
None,
|
|
description="Return only the last N output entries. Useful to limit response size when only recent output matters.",
|
|
ge=1,
|
|
),
|
|
):
|
|
background_process = _get_process(process_id)
|
|
|
|
if wait is None and EXECUTE_TIMEOUT:
|
|
wait = EXECUTE_TIMEOUT
|
|
if wait is not None and background_process.status == "running":
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.shield(background_process.log_task), timeout=wait
|
|
)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
output, next_offset, truncated = await read_log(
|
|
background_process.log_path, offset=offset, tail=tail
|
|
)
|
|
|
|
return {
|
|
"id": background_process.id,
|
|
"command": background_process.command,
|
|
"status": background_process.status,
|
|
"exit_code": background_process.exit_code,
|
|
"output": output,
|
|
"truncated": truncated,
|
|
"next_offset": next_offset,
|
|
"log_path": background_process.log_path,
|
|
}
|
|
|
|
|
|
@app.post(
|
|
"/execute/{process_id}/input",
|
|
operation_id="send_process_input",
|
|
summary="Send input to a running command",
|
|
description="Write text to the process's stdin. Include newline characters as needed.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Process not found."},
|
|
400: {"description": "Process has already exited or stdin is closed."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def send_input(process_id: str, body: InputRequest):
|
|
background_process = _get_process(process_id)
|
|
if background_process.status != "running":
|
|
raise HTTPException(status_code=400, detail="Process has already exited")
|
|
|
|
# Convert literal escape sequences (\n, \x03 for Ctrl-C, etc.) into real
|
|
# characters — LLMs often emit these as literal strings.
|
|
text = body.input.encode("raw_unicode_escape").decode("unicode_escape")
|
|
|
|
try:
|
|
background_process.runner.write_input(text.encode())
|
|
if isinstance(background_process.runner, PipeRunner):
|
|
await background_process.runner.drain_input()
|
|
except (BrokenPipeError, ConnectionResetError, OSError):
|
|
raise HTTPException(status_code=400, detail="Process stdin is closed")
|
|
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.delete(
|
|
"/execute/{process_id}",
|
|
operation_id="kill_process",
|
|
summary="Kill a running command",
|
|
description="Terminate the process. Sends SIGTERM by default for graceful shutdown. Use force=true to send SIGKILL.",
|
|
dependencies=[Depends(verify_api_key)],
|
|
responses={
|
|
404: {"description": "Process not found."},
|
|
401: {"description": "Invalid or missing API key."},
|
|
},
|
|
)
|
|
async def kill_process(
|
|
process_id: str,
|
|
force: bool = Query(False, description="Send SIGKILL instead of SIGTERM."),
|
|
):
|
|
background_process = _get_process(process_id)
|
|
if background_process.status == "running":
|
|
background_process.runner.kill(force=force)
|
|
exit_code = await background_process.runner.wait()
|
|
background_process.runner.close()
|
|
background_process.status = "killed"
|
|
background_process.exit_code = exit_code
|
|
del _processes[process_id]
|
|
return {"status": "killed"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Port detection & proxy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from open_terminal.utils.port import detect_listening_ports, get_descendant_pids
|
|
|
|
@app.get(
|
|
"/ports",
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def list_ports(request: Request):
|
|
"""Return TCP ports currently listening on localhost.
|
|
|
|
In multi-user mode, only shows ports owned by the requesting user.
|
|
In single-user mode, shows ports owned by descendant processes.
|
|
"""
|
|
all_ports = await asyncio.to_thread(detect_listening_ports)
|
|
|
|
try:
|
|
fs = get_filesystem(request)
|
|
except Exception:
|
|
# User provisioning failed (e.g. useradd rejected in restricted
|
|
# container runtimes). An unprovisioned user has no ports.
|
|
return {"ports": []}
|
|
|
|
if fs.username:
|
|
# Filter by user UID
|
|
import pwd
|
|
try:
|
|
user_uid = pwd.getpwnam(fs.username).pw_uid
|
|
all_ports = [p for p in all_ports if p.get("uid") == user_uid]
|
|
except KeyError:
|
|
all_ports = []
|
|
else:
|
|
own_pid = os.getpid()
|
|
descendant_pids = await asyncio.to_thread(get_descendant_pids, own_pid)
|
|
all_ports = [p for p in all_ports if p.get("pid") in descendant_pids]
|
|
|
|
# Strip uid from response (internal detail)
|
|
for p in all_ports:
|
|
p.pop("uid", None)
|
|
|
|
return {"ports": all_ports}
|
|
|
|
|
|
# -- Port proxy client (reused across requests) --
|
|
_port_proxy_client = None
|
|
|
|
|
|
async def _get_port_proxy_client():
|
|
global _port_proxy_client
|
|
if _port_proxy_client is None:
|
|
import httpx
|
|
_port_proxy_client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(300.0, connect=5.0),
|
|
follow_redirects=False,
|
|
)
|
|
return _port_proxy_client
|
|
|
|
|
|
@app.api_route(
|
|
"/proxy/{port}/{path:path}",
|
|
methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
|
|
include_in_schema=False,
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
async def port_proxy(port: int, path: str, request: Request):
|
|
"""Reverse-proxy a request to localhost:{port}/{path}."""
|
|
if port < 1 or port > 65535:
|
|
raise HTTPException(status_code=422, detail="Port must be between 1 and 65535")
|
|
|
|
target_url = f"http://localhost:{port}/{path}"
|
|
if request.query_params:
|
|
target_url += f"?{request.query_params}"
|
|
|
|
# Forward headers, stripping hop-by-hop and host.
|
|
headers = dict(request.headers)
|
|
for h in ("host", "transfer-encoding", "connection", "authorization"):
|
|
headers.pop(h, None)
|
|
|
|
body = await request.body()
|
|
|
|
import httpx
|
|
|
|
client = await _get_port_proxy_client()
|
|
try:
|
|
upstream = await client.request(
|
|
method=request.method,
|
|
url=target_url,
|
|
headers=headers,
|
|
content=body or None,
|
|
)
|
|
except httpx.ConnectError:
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail=f"Connection refused: localhost:{port}",
|
|
)
|
|
except httpx.TimeoutException:
|
|
raise HTTPException(
|
|
status_code=504,
|
|
detail=f"Timeout connecting to localhost:{port}",
|
|
)
|
|
|
|
response_headers = dict(upstream.headers)
|
|
for h in ("transfer-encoding", "connection", "content-encoding", "content-length"):
|
|
response_headers.pop(h, None)
|
|
|
|
return Response(
|
|
content=upstream.content,
|
|
status_code=upstream.status_code,
|
|
headers=response_headers,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interactive terminal sessions (resource-oriented API)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if ENABLE_TERMINAL:
|
|
|
|
import uuid as _uuid
|
|
from datetime import datetime as _datetime
|
|
from fastapi.responses import JSONResponse
|
|
|
|
try:
|
|
import select as _select
|
|
except ImportError:
|
|
_select = None # Not available on all platforms in all contexts
|
|
|
|
# Determine terminal backend: prefer Unix PTY, then pywinpty, else None
|
|
if _PTY_AVAILABLE:
|
|
_TERMINAL_BACKEND = "pty"
|
|
else:
|
|
try:
|
|
from winpty import PtyProcess as _WinPtyProcess
|
|
|
|
_TERMINAL_BACKEND = "winpty"
|
|
except ImportError:
|
|
_TERMINAL_BACKEND = None
|
|
|
|
# Active terminal sessions: {id: {...}}
|
|
_terminal_sessions: dict[str, dict] = {}
|
|
|
|
|
|
def _cleanup_session(session_id: str):
|
|
"""Clean up a terminal session's resources.
|
|
|
|
For PTY sessions the shell is spawned with ``start_new_session=True``,
|
|
giving it a dedicated process group. We signal the *entire* group so
|
|
that background jobs started inside the terminal (e.g. ``sleep 999 &``)
|
|
are also reaped, and always call ``process.wait()`` to avoid zombies.
|
|
"""
|
|
session = _terminal_sessions.pop(session_id, None)
|
|
if session is None:
|
|
return
|
|
|
|
backend = session.get("backend")
|
|
|
|
if backend == "pty":
|
|
try:
|
|
os.close(session["master_fd"])
|
|
except OSError:
|
|
pass
|
|
|
|
process = session["process"]
|
|
if process.poll() is None:
|
|
# Signal the whole process group first (graceful).
|
|
try:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
try:
|
|
process.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
# Forceful kill of the entire group.
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
process.wait()
|
|
|
|
elif backend == "winpty":
|
|
pty_proc = session["pty_process"]
|
|
if pty_proc.isalive():
|
|
pty_proc.terminate()
|
|
|
|
|
|
@app.post("/api/terminals", dependencies=[Depends(verify_api_key)], include_in_schema=False)
|
|
async def create_terminal(request: Request):
|
|
"""Create a new terminal session and return its ID."""
|
|
if _TERMINAL_BACKEND is None:
|
|
return JSONResponse(
|
|
{"error": "PTY not available on this platform (install pywinpty on Windows)"},
|
|
status_code=503,
|
|
)
|
|
|
|
# Prune dead sessions before checking limit
|
|
if _TERMINAL_BACKEND == "pty":
|
|
dead = [sid for sid, s in _terminal_sessions.items() if s["process"].poll() is not None]
|
|
else:
|
|
dead = [sid for sid, s in _terminal_sessions.items() if not s["pty_process"].isalive()]
|
|
for sid in dead:
|
|
_cleanup_session(sid)
|
|
|
|
if len(_terminal_sessions) >= MAX_TERMINAL_SESSIONS:
|
|
return JSONResponse(
|
|
{"error": f"Maximum number of terminal sessions ({MAX_TERMINAL_SESSIONS}) reached"},
|
|
status_code=429,
|
|
)
|
|
|
|
session_id = str(_uuid.uuid4())[:8]
|
|
|
|
if _TERMINAL_BACKEND == "pty":
|
|
try:
|
|
master_fd, slave_fd = pty.openpty()
|
|
except OSError:
|
|
return JSONResponse(
|
|
{"error": "Out of PTY devices — too many active terminals or processes"},
|
|
status_code=503,
|
|
)
|
|
|
|
try:
|
|
fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0))
|
|
|
|
fs = get_filesystem(request)
|
|
|
|
# Use per-session cwd if available, else fall back to home
|
|
session_id = request.headers.get("x-session-id", session_id)
|
|
session_cwd = _get_session_cwd(session_id, fs) if session_id else None
|
|
|
|
if fs.username:
|
|
shell_cmd = [
|
|
"script", "-qc",
|
|
f"sudo -i -u {fs.username}",
|
|
"/dev/null",
|
|
]
|
|
cwd = session_cwd or fs.home
|
|
else:
|
|
shell_cmd = [os.environ.get("SHELL", "/bin/sh")]
|
|
cwd = session_cwd or os.getcwd()
|
|
|
|
spawn_env = os.environ.copy()
|
|
spawn_env.setdefault("TERM", TERMINAL_TERM)
|
|
process = subprocess.Popen(
|
|
shell_cmd,
|
|
stdin=slave_fd,
|
|
stdout=slave_fd,
|
|
stderr=slave_fd,
|
|
cwd=cwd,
|
|
env=spawn_env,
|
|
start_new_session=True,
|
|
)
|
|
except Exception:
|
|
os.close(slave_fd)
|
|
os.close(master_fd)
|
|
raise
|
|
os.close(slave_fd)
|
|
|
|
# Set non-blocking
|
|
flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
|
|
fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
|
|
|
|
_terminal_sessions[session_id] = {
|
|
"backend": "pty",
|
|
"master_fd": master_fd,
|
|
"process": process,
|
|
"created_at": _datetime.utcnow().isoformat() + "Z",
|
|
"pid": process.pid,
|
|
}
|
|
|
|
else: # winpty
|
|
shell = os.environ.get("COMSPEC", "cmd.exe")
|
|
spawn_env = os.environ.copy()
|
|
spawn_env.setdefault("TERM", TERMINAL_TERM)
|
|
pty_proc = _WinPtyProcess.spawn(
|
|
[shell],
|
|
cwd=os.getcwd(),
|
|
env=spawn_env,
|
|
dimensions=(24, 80),
|
|
)
|
|
_terminal_sessions[session_id] = {
|
|
"backend": "winpty",
|
|
"pty_process": pty_proc,
|
|
"created_at": _datetime.utcnow().isoformat() + "Z",
|
|
"pid": pty_proc.pid,
|
|
}
|
|
|
|
session = _terminal_sessions[session_id]
|
|
return {
|
|
"id": session_id,
|
|
"created_at": session["created_at"],
|
|
"pid": session["pid"],
|
|
}
|
|
|
|
|
|
def _session_is_alive(session: dict) -> bool:
|
|
"""Check if a terminal session's process is still running."""
|
|
if session["backend"] == "pty":
|
|
return session["process"].poll() is None
|
|
else:
|
|
return session["pty_process"].isalive()
|
|
|
|
|
|
@app.get("/api/terminals", dependencies=[Depends(verify_api_key)], include_in_schema=False)
|
|
async def list_terminals(request: Request):
|
|
"""List active terminal sessions."""
|
|
result = []
|
|
to_remove = []
|
|
for sid, session in _terminal_sessions.items():
|
|
if not _session_is_alive(session):
|
|
to_remove.append(sid)
|
|
continue
|
|
result.append({
|
|
"id": sid,
|
|
"created_at": session["created_at"],
|
|
"pid": session["pid"],
|
|
})
|
|
for sid in to_remove:
|
|
_cleanup_session(sid)
|
|
return result
|
|
|
|
|
|
@app.get("/api/terminals/{session_id}", dependencies=[Depends(verify_api_key)], include_in_schema=False)
|
|
async def get_terminal(session_id: str, request: Request):
|
|
"""Get info about a terminal session."""
|
|
session = _terminal_sessions.get(session_id)
|
|
if session is None:
|
|
return JSONResponse({"error": "Session not found"}, status_code=404)
|
|
if not _session_is_alive(session):
|
|
_cleanup_session(session_id)
|
|
return JSONResponse({"error": "Session not found"}, status_code=404)
|
|
return {
|
|
"id": session_id,
|
|
"created_at": session["created_at"],
|
|
"pid": session["pid"],
|
|
}
|
|
|
|
|
|
@app.delete("/api/terminals/{session_id}", dependencies=[Depends(verify_api_key)], include_in_schema=False)
|
|
async def delete_terminal(session_id: str, request: Request):
|
|
"""Kill and remove a terminal session."""
|
|
if session_id not in _terminal_sessions:
|
|
return JSONResponse({"error": "Session not found"}, status_code=404)
|
|
_cleanup_session(session_id)
|
|
return {"status": "deleted"}
|
|
|
|
|
|
@app.websocket("/api/terminals/{session_id}")
|
|
async def ws_terminal(ws: WebSocket, session_id: str):
|
|
"""Attach to an existing terminal session via WebSocket.
|
|
|
|
Authentication is via **first-message auth**: after connecting, the client
|
|
must send a JSON text frame as its first message::
|
|
|
|
{"type": "auth", "token": "<api_key>"}
|
|
|
|
The server validates the token and closes the connection if invalid.
|
|
After authentication, the client sends keystrokes as **binary** frames
|
|
and receives PTY output as binary frames.
|
|
|
|
To resize, send a **text** JSON frame::
|
|
|
|
{"type": "resize", "cols": 120, "rows": 40}
|
|
"""
|
|
session = _terminal_sessions.get(session_id)
|
|
if session is None:
|
|
await ws.close(code=4004, reason="Session not found")
|
|
return
|
|
|
|
if not _session_is_alive(session):
|
|
_cleanup_session(session_id)
|
|
await ws.close(code=4004, reason="Session has ended")
|
|
return
|
|
|
|
await ws.accept()
|
|
|
|
# First-message authentication
|
|
if API_KEY:
|
|
try:
|
|
msg = await asyncio.wait_for(ws.receive_text(), timeout=10.0)
|
|
payload = json.loads(msg)
|
|
if payload.get("type") != "auth" or not hmac.compare_digest(payload.get("token", ""), API_KEY):
|
|
await ws.close(code=4001, reason="Invalid API key")
|
|
return
|
|
except (asyncio.TimeoutError, json.JSONDecodeError, Exception):
|
|
await ws.close(code=4001, reason="Auth timeout or invalid payload")
|
|
return
|
|
|
|
backend = session["backend"]
|
|
loop = asyncio.get_event_loop()
|
|
stop_event = asyncio.Event()
|
|
|
|
# --- Platform-specific read/write/resize helpers ---
|
|
|
|
if backend == "pty":
|
|
master_fd = session["master_fd"]
|
|
process = session["process"]
|
|
|
|
def _blocking_read():
|
|
"""Read from PTY using select() so we don't block forever."""
|
|
while not stop_event.is_set():
|
|
try:
|
|
rlist, _, _ = _select.select([master_fd], [], [], 0.1)
|
|
if rlist:
|
|
return os.read(master_fd, 4096)
|
|
except (OSError, ValueError):
|
|
return b""
|
|
return b""
|
|
|
|
def _check_alive():
|
|
return process.poll() is None
|
|
|
|
def _write_data(data: bytes):
|
|
os.write(master_fd, data)
|
|
|
|
def _do_resize(rows: int, cols: int):
|
|
fcntl.ioctl(
|
|
master_fd,
|
|
termios.TIOCSWINSZ,
|
|
struct.pack("HHHH", rows, cols, 0, 0),
|
|
)
|
|
|
|
else: # winpty
|
|
pty_proc = session["pty_process"]
|
|
|
|
def _blocking_read():
|
|
"""Read from WinPTY process."""
|
|
try:
|
|
data = pty_proc.read(4096)
|
|
return data.encode(errors="replace") if data else b""
|
|
except EOFError:
|
|
return b""
|
|
except Exception:
|
|
return b""
|
|
|
|
def _check_alive():
|
|
return pty_proc.isalive()
|
|
|
|
def _write_data(data: bytes):
|
|
pty_proc.write(data.decode(errors="replace"))
|
|
|
|
def _do_resize(rows: int, cols: int):
|
|
pty_proc.setwinsize(rows, cols)
|
|
|
|
# --- Reader / writer tasks ---
|
|
|
|
async def _pty_reader():
|
|
"""Forward PTY output -> WebSocket."""
|
|
try:
|
|
while not stop_event.is_set():
|
|
data = await loop.run_in_executor(None, _blocking_read)
|
|
if not data:
|
|
if stop_event.is_set():
|
|
break
|
|
if not _check_alive():
|
|
break
|
|
continue
|
|
try:
|
|
await ws.send_bytes(data)
|
|
except Exception:
|
|
break
|
|
finally:
|
|
pass
|
|
|
|
reader_task = asyncio.create_task(_pty_reader())
|
|
|
|
try:
|
|
while True:
|
|
msg = await ws.receive()
|
|
if msg["type"] == "websocket.disconnect":
|
|
break
|
|
elif "bytes" in msg and msg["bytes"]:
|
|
await loop.run_in_executor(None, _write_data, msg["bytes"])
|
|
elif "text" in msg and msg["text"]:
|
|
try:
|
|
payload = json.loads(msg["text"])
|
|
if payload.get("type") == "resize":
|
|
cols = payload.get("cols", 80)
|
|
rows = payload.get("rows", 24)
|
|
_do_resize(rows, cols)
|
|
except (json.JSONDecodeError, KeyError):
|
|
pass
|
|
except WebSocketDisconnect:
|
|
pass
|
|
finally:
|
|
stop_event.set()
|
|
reader_task.cancel()
|
|
try:
|
|
await reader_task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
# Clean up session on disconnect
|
|
_cleanup_session(session_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Notebook execution (optional)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if ENABLE_NOTEBOOKS:
|
|
from open_terminal.utils.notebooks import create_notebooks_router
|
|
|
|
app.include_router(create_notebooks_router(verify_api_key))
|