diff --git a/Dockerfile b/Dockerfile index c0a3acb..de5d65f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,6 +29,11 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates gnupg apt-transport-https \ # Capabilities (needed for setcap on Python binary) libcap2-bin \ + # Virtual desktop ("Computer Use") + xvfb x11vnc novnc openbox xdotool scrot xauth \ + xterm x11-xserver-utils \ + fonts-liberation fonts-noto-color-emoji \ + dmz-cursor-theme \ && rm -rf /var/lib/apt/lists/* # Node.js (LTS) @@ -36,6 +41,10 @@ RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \ && apt-get install -y --no-install-recommends nodejs \ && rm -rf /var/lib/apt/lists/* +# Chromium (for headful browser automation via the virtual desktop) +RUN apt-get update && apt-get install -y --no-install-recommends chromium \ + && rm -rf /var/lib/apt/lists/* + # Docker CLI + Compose + Buildx (mount socket at runtime for access) RUN curl -fsSL https://get.docker.com | sh @@ -68,12 +77,22 @@ RUN pip install --no-cache-dir . \ RUN useradd -m -s /bin/bash user && echo 'user ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers +# Chromium needs a writable /dev/shm for shared memory. When running +# without --no-sandbox (the default inside Docker) we need at least 64 MB. +# Some container runtimes mount /dev/shm as 64 MB which is too small for +# Chromium; the kernel will silently OOM the renderer. We work around +# this by creating a small tmpfs in the user's home directory. +RUN echo "kernel.shmmax = 268435456" >> /etc/sysctl.conf || true + +RUN printf '#!/bin/sh\nexport CHROMIUM_FLAGS="$CHROMIUM_FLAGS --no-sandbox --disable-gpu --disable-software-rasterizer"\n' \ + > /etc/chromium.d/00-container + USER user ENV SHELL=/bin/bash ENV PATH="/home/user/.local/bin:${PATH}" WORKDIR /home/user -EXPOSE 8000 +EXPOSE 8000 6080 COPY entrypoint.sh /app/entrypoint.sh diff --git a/entrypoint.sh b/entrypoint.sh index 8afbba5..3b9a9ef 100755 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -81,6 +81,32 @@ if [ -n "${OPEN_TERMINAL_NPM_PACKAGES:-}" ]; then fi fi +# ----------------------------------------------------------------------- +# Virtual Desktop ("Computer Use") +# +# When OPEN_TERMINAL_ENABLE_DESKTOP is true, start Xvfb, x11vnc, and +# noVNC so that the agent has a virtual display for GUI interaction. +# The Python app will manage the actual lifecycle (start/stop) via the +# DesktopManager, but we pre-seed DISPLAY and clean up stale lock files +# so the first API call starts faster. +# ----------------------------------------------------------------------- +if [ "${OPEN_TERMINAL_ENABLE_DESKTOP:-false}" = "true" ]; then + DISPLAY="${OPEN_TERMINAL_DESKTOP_DISPLAY:-:0}" + SCREEN="${OPEN_TERMINAL_DESKTOP_SCREEN_SIZE:-1280x720x24}" + VNC_PORT="${OPEN_TERMINAL_DESKTOP_VNC_PORT:-5900}" + NOVNC_PORT="${OPEN_TERMINAL_DESKTOP_NOVNC_PORT:-6080}" + + # Clean up stale lock/pid files from previous runs + DISPLAY_NUM="${DISPLAY#:}" + DISPLAY_NUM="${DISPLAY_NUM%%.*}" + rm -f "/tmp/.X${DISPLAY_NUM}-lock" "/tmp/.X11-unix/X${DISPLAY_NUM}" 2>/dev/null || true + + export DISPLAY + echo "Virtual desktop configured: display=${DISPLAY} screen=${SCREEN}" + echo " VNC port: ${VNC_PORT} | noVNC port: ${NOVNC_PORT}" + echo " Access noVNC at: http://localhost:${NOVNC_PORT}/vnc.html" +fi + # ----------------------------------------------------------------------- # Network egress filtering via DNS whitelist + iptables + capability drop # diff --git a/open_terminal/env.py b/open_terminal/env.py index da0aecb..bec7374 100644 --- a/open_terminal/env.py +++ b/open_terminal/env.py @@ -170,4 +170,37 @@ def _resolve_file_env(var: str, default: str = "") -> str: ) ) +# --------------------------------------------------------------------------- +# Virtual Desktop ("Computer Use") +# --------------------------------------------------------------------------- + +ENABLE_DESKTOP = os.environ.get( + "OPEN_TERMINAL_ENABLE_DESKTOP", + str(config.get("enable_desktop", False)), +).lower() not in ("false", "0", "no", "") + +DESKTOP_DISPLAY = os.environ.get( + "OPEN_TERMINAL_DESKTOP_DISPLAY", + config.get("desktop_display", ":0"), +) + +DESKTOP_SCREEN_SIZE = os.environ.get( + "OPEN_TERMINAL_DESKTOP_SCREEN_SIZE", + config.get("desktop_screen_size", "1280x720x24"), +) + +DESKTOP_VNC_PORT = int( + os.environ.get( + "OPEN_TERMINAL_DESKTOP_VNC_PORT", + config.get("desktop_vnc_port", "5900"), + ) +) + +DESKTOP_NOVNC_PORT = int( + os.environ.get( + "OPEN_TERMINAL_DESKTOP_NOVNC_PORT", + config.get("desktop_novnc_port", "6080"), + ) +) + diff --git a/open_terminal/main.py b/open_terminal/main.py index 5a65ff9..e8064c1 100644 --- a/open_terminal/main.py +++ b/open_terminal/main.py @@ -24,20 +24,36 @@ 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, LOG_DIR, MAX_TERMINAL_SESSIONS, MULTI_USER, OPEN_TERMINAL_INFO, PROCESS_LOG_RETENTION, SESSION_CWD_TTL, SYSTEM_PROMPT, TERMINAL_TERM +from open_terminal.env import ( + API_KEY, + BINARY_FILE_MIME_PREFIXES, + CORS_ALLOWED_ORIGINS, + ENABLE_DESKTOP, + ENABLE_NOTEBOOKS, + ENABLE_SYSTEM_PROMPT, + ENABLE_TERMINAL, + EXECUTE_DESCRIPTION, + EXECUTE_TIMEOUT, + 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" + '\n\033[91m OPEN_TERMINAL_API_KEY is required.\n Set via environment variable or --api-key flag.\n\033[0m' ) try: @@ -54,12 +70,12 @@ 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 "" + 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." + 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.' ) @@ -68,32 +84,40 @@ def get_system_prompt() -> str: if SYSTEM_PROMPT: return SYSTEM_PROMPT - shell = os.environ.get("SHELL", "/bin/sh") - user_part = f" as user '{os.getenv('USER', 'unknown')}'" if not MULTI_USER else "" + 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'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." + 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}" + prompt += f'\n\n{OPEN_TERMINAL_INFO}' + + if ENABLE_DESKTOP: + prompt += ( + '\n\nThis environment has a virtual desktop with a graphical display. ' + 'You can use the desktop tools to take screenshots, click, type, and ' + 'interact with GUI applications — enabling "computer use" capabilities. ' + 'Start the desktop with POST /desktop/start before using desktop tools.\n\n' + 'IMPORTANT: Always call desktop_locate to find UI elements before interacting with them. ' + 'The workflow is: desktop_locate(description="...") → get bounding box with coordinates → ' + 'desktop_click(x=..., y=...). Never guess coordinates — always locate first.' + ) return prompt -_EXECUTE_DESCRIPTION = ( - "Run a shell command in the background and return a command ID.\n\n" - + get_system_info() -) +_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 + _EXECUTE_DESCRIPTION += '\n\n' + EXECUTE_DESCRIPTION bearer_scheme = HTTPBearer(auto_error=False) @@ -104,7 +128,7 @@ async def verify_api_key( 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") + raise HTTPException(status_code=401, detail='Invalid API key') def get_filesystem(request: Request) -> UserFS: @@ -116,7 +140,7 @@ def get_filesystem(request: Request) -> UserFS: """ if not MULTI_USER: return UserFS() - user_id = request.headers.get("x-user-id") + user_id = request.headers.get('x-user-id') if not user_id: return UserFS() username, home = resolve_user(user_id) @@ -124,33 +148,33 @@ def get_filesystem(request: Request) -> UserFS: app = FastAPI( - title="Open Terminal", - description="A remote terminal API.", - version=_pkg_version("open-terminal"), + 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_origins=[o.strip() for o in CORS_ALLOWED_ORIGINS.split(',')], allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + 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)}) + return JSONResponse(status_code=403, content={'detail': str(exc)}) -@app.middleware("http") +@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"] + 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") + request.scope['query_string'] = urlencode(cleaned).encode('utf-8') return await call_next(request) @@ -162,8 +186,8 @@ async def normalize_null_query_params(request: Request, call_next): 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"]}, + description='Shell command to execute. Supports chaining (&&, ||, ;), pipes (|), and redirections.', + json_schema_extra={'examples': ['echo hello', 'ls -la && whoami']}, ) cwd: Optional[str] = Field( None, @@ -171,7 +195,7 @@ class ExecRequest(BaseModel): ) env: Optional[dict[str, str]] = Field( None, - description="Extra environment variables merged into the subprocess environment.", + description='Extra environment variables merged into the subprocess environment.', ) @@ -185,69 +209,68 @@ class InputRequest(BaseModel): class WriteRequest(BaseModel): path: str = Field( ..., - description="Absolute or relative path to write to. Parent directories are created automatically.", + description='Absolute or relative path to write to. Parent directories are created automatically.', ) content: str = Field( ..., - description="Text content to write to the file.", + description='Text content to write to the file.', ) class ReplacementChunk(BaseModel): target: str = Field( ..., - description="Exact string to find. Must match precisely, including whitespace.", + description='Exact string to find. Must match precisely, including whitespace.', ) replacement: str = Field( ..., - description="Content to replace the target with.", + 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).", + 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).", + 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.", + 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.", + 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.", + description='Path to the file or directory to move.', ) destination: str = Field( ..., - description="Destination path (new location).", + description='Destination path (new location).', ) class ReplaceRequest(BaseModel): path: str = Field( ..., - description="Path to the file to modify.", + description='Path to the file to modify.', ) replacements: list[ReplacementChunk] = Field( ..., - description="List of find-and-replace operations to apply sequentially.", + description='List of find-and-replace operations to apply sequentially.', ) - # --------------------------------------------------------------------------- # Background process management # --------------------------------------------------------------------------- @@ -258,7 +281,7 @@ class BackgroundProcess: id: str command: str runner: ProcessRunner - status: str = "running" + 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) @@ -278,7 +301,6 @@ class BackgroundProcess: _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() @@ -287,7 +309,7 @@ def _expire_session_cwds(): del _session_cwds[sid] -def _get_session_cwd(session_id: str | None, fs: "UserFS") -> str: +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: @@ -306,8 +328,6 @@ def _set_session_cwd(session_id: str | None, path: str): from open_terminal.utils.log import log_process, read_log - - def _cleanup_expired(): """Remove finished processes that have expired. @@ -317,17 +337,12 @@ def _cleanup_expired(): expired = [ process_id for process_id, background_process in _processes.items() - if background_process.finished_at - and now - background_process.finished_at > _EXPIRY_SECONDS + 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 - ): + if bp.log_path and bp.finished_at and now - bp.finished_at > PROCESS_LOG_RETENTION: try: os.remove(bp.log_path) except OSError: @@ -338,7 +353,7 @@ 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") + raise HTTPException(status_code=404, detail='Process not found') return background_process @@ -348,13 +363,13 @@ def _get_process(process_id: str) -> BackgroundProcess: @app.get( - "/health", - operation_id="health_check", - summary="Health check", - description="Returns service status. No authentication required.", + '/health', + operation_id='health_check', + summary='Health check', + description='Returns service status. No authentication required.', ) async def health(): - return {"status": "ok"} + return {'status': 'ok'} # --------------------------------------------------------------------------- @@ -363,16 +378,17 @@ async def health(): @app.get( - "/api/config", + '/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, + 'features': { + 'terminal': ENABLE_TERMINAL, + 'notebooks': ENABLE_NOTEBOOKS, + 'system': ENABLE_SYSTEM_PROMPT, + 'desktop': ENABLE_DESKTOP, }, } @@ -380,26 +396,26 @@ async def get_config(): if ENABLE_SYSTEM_PROMPT: @app.get( - "/system", + '/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()} + 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.", + '/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} + return {'info': OPEN_TERMINAL_INFO} # --------------------------------------------------------------------------- @@ -408,7 +424,7 @@ async def get_info(): @app.get( - "/files/cwd", + '/files/cwd', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -416,12 +432,12 @@ async def get_cwd( http_request: Request, fs: UserFS = Depends(get_filesystem), ): - session_id = http_request.headers.get("x-session-id") - return {"cwd": _get_session_cwd(session_id, fs)} + session_id = http_request.headers.get('x-session-id') + return {'cwd': _get_session_cwd(session_id, fs)} @app.post( - "/files/cwd", + '/files/cwd', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -430,67 +446,67 @@ async def set_cwd( request: MkdirRequest, fs: UserFS = Depends(get_filesystem), ): - session_id = http_request.headers.get("x-session-id") + 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") + raise HTTPException(status_code=404, detail='Directory not found') _set_session_cwd(session_id, target) - return {"cwd": 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.", + '/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."}, + 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."), + directory: str = Query('.', description='Directory path to list.'), fs: UserFS = Depends(get_filesystem), ): - session_id = http_request.headers.get("x-session-id") + 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") + raise HTTPException(status_code=404, detail='Directory not found') entries = await fs.listdir(target) - return {"dir": target, "entries": entries} + 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.", + '/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."}, + 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."), + 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 + 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 + 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_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") + raise HTTPException(status_code=404, detail='File not found') try: content = await fs.read_text(target) @@ -500,23 +516,21 @@ async def read_file( raw = await fs.read(target) mime, _ = mimetypes.guess_type(target) - mime = mime or "application/octet-stream" + 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) - ): + 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]), + 'path': target, + 'total_lines': len(lines), + 'content': ''.join(lines[start:end]), } # Return raw binary for allowed mime type prefixes (e.g. image/*) @@ -526,31 +540,31 @@ async def read_file( # 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)", + 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]), + '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", + '/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."}, + 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."), + 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. @@ -560,20 +574,20 @@ async def display_file( 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_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} + return {'path': target, 'exists': exists} @app.get( - "/files/view", + '/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."), + path: str = Query(..., description='Path to the file to view.'), fs: UserFS = Depends(get_filesystem), ): """Return raw file bytes with the appropriate Content-Type. @@ -583,39 +597,39 @@ async def view_file( """ target = fs.resolve_path(path) if not await fs.isfile(target): - raise HTTPException(status_code=404, detail="File not found") + raise HTTPException(status_code=404, detail='File not found') import mimetypes mime, _ = mimetypes.guess_type(target) - mime = mime or "application/octet-stream" + mime = mime or 'application/octet-stream' raw = await fs.read(target) return Response(content=raw, media_type=mime) @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.", + '/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."}, + 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_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())} + return {'path': target, 'size': len(request.content.encode())} @app.post( - "/files/mkdir", + '/files/mkdir', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -625,31 +639,31 @@ async def mkdir(request: MkdirRequest, fs: UserFS = Depends(get_filesystem)): await fs.mkdir(target) except (OSError, subprocess.CalledProcessError) as e: raise HTTPException(status_code=400, detail=str(e)) - return {"path": target} + return {'path': target} @app.delete( - "/files/delete", + '/files/delete', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) async def delete_entry( - path: str = Query(..., description="Path to delete."), + 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") + 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"} + return {'path': target, 'type': 'directory' if is_dir else 'file'} @app.post( - "/files/move", + '/files/move', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -658,40 +672,40 @@ async def move_entry(request: MoveRequest, fs: UserFS = Depends(get_filesystem)) destination = fs.resolve_path(request.destination) if not await fs.exists(source): - raise HTTPException(status_code=404, detail="Source path not found") + 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") + 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") + 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} + 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.", + '/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."}, + 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_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") + raise HTTPException(status_code=404, detail='File not found') try: content = await fs.read_text(target) @@ -703,7 +717,7 @@ async def replace_file_content(http_request: Request, request: ReplaceRequest, f 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]) + search_region = ''.join(lines[start:end]) else: search_region = content @@ -711,18 +725,18 @@ async def replace_file_content(http_request: Request, request: ReplaceRequest, f if count == 0: raise HTTPException( status_code=400, - detail=f"Target string not found: {chunk.target[:100]!r}", + 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", + 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) + content = ''.join(lines) else: content = content.replace(chunk.target, chunk.replacement) @@ -731,54 +745,50 @@ async def replace_file_content(http_request: Request, request: ReplaceRequest, f except OSError as e: raise HTTPException(status_code=400, detail=str(e)) - return {"path": target, "size": len(content.encode())} + 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.", + '/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."}, + 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(False, description="Treat query as a regex pattern."), - case_insensitive: bool = Query( - False, description="Perform case-insensitive matching." - ), + query: str = Query(..., description='Text or regex pattern to search for.'), + path: str = Query('.', description='Directory or file to search in.'), + regex: bool = Query(False, description='Treat query as a regex pattern.'), + 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 + 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_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") + 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}") + raise HTTPException(status_code=400, detail=f'Invalid regex: {exc}') else: pattern = re.compile(re.escape(query), flags) @@ -796,22 +806,22 @@ def _search_file(file_path: str): if truncated: return try: - with open(file_path, "r", encoding="utf-8", errors="strict") as f: + 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"), + '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}) + matches.append({'file': file_path}) if len(matches) >= max_results: truncated = True return # one match per file is enough @@ -823,10 +833,7 @@ def _search_file(file_path: str): 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)) - ] + dirnames[:] = [d for d in dirnames if fs.is_path_allowed(os.path.join(dirpath, d))] if truncated: break for filename in sorted(filenames): @@ -841,46 +848,42 @@ def _search_file(file_path: str): matches, truncated = await asyncio.to_thread(_search_sync) return { - "query": query, - "path": target, - "matches": matches, - "truncated": truncated, + '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.", + '/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."}, + 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." - ), + 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", + '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 + 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_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") + raise HTTPException(status_code=404, detail='Search directory not found') def _glob_sync(): matches = [] @@ -891,16 +894,13 @@ def _glob_sync(): break # Prune directories belonging to other users. - dirnames[:] = [ - d for d in dirnames - if fs.is_path_allowed(os.path.join(dirpath, d)) - ] + 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]) + 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: @@ -910,26 +910,21 @@ def _glob_sync(): rel_path = os.path.relpath(full_path, target) # Check inclusion pattern - if not fnmatch.fnmatch(name, pattern) and not fnmatch.fnmatch( - rel_path, 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 - ): + 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, + 'path': rel_path, + 'type': entry_type, + 'size': file_stat.st_size, + 'modified': file_stat.st_mtime, } ) @@ -943,35 +938,31 @@ def _glob_sync(): matches, truncated = await asyncio.to_thread(_glob_sync) return { - "pattern": pattern, - "path": target, - "matches": matches, - "truncated": truncated, + 'pattern': pattern, + 'path': target, + 'matches': matches, + 'truncated': truncated, } - - @app.post( - "/files/upload", + '/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.", + 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."}, + 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." - ), + 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") + filename = os.path.basename(file.filename or 'upload') directory = fs.resolve_path(directory) path = os.path.normpath(os.path.join(directory, filename)) @@ -983,18 +974,18 @@ async def upload_file( 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)} + 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.", + description='List of file or directory paths to include in the ZIP archive.', ) @app.post( - "/files/archive", + '/files/archive', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -1007,50 +998,45 @@ async def archive_paths( import zipfile if not request.paths: - raise HTTPException(status_code=400, detail="No paths provided") + 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}") + 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" + archive_name = os.path.basename(resolved[0].rstrip('/\\')) or 'archive' else: - archive_name = "download" + archive_name = 'download' def _build_zip() -> bytes: buf = io.BytesIO() - with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + 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" + 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)) - ] + 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) - ) + 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", + media_type='application/zip', headers={ - "Content-Disposition": f'attachment; filename="{archive_name}.zip"', + 'Content-Disposition': f'attachment; filename="{archive_name}.zip"', }, ) @@ -1061,37 +1047,37 @@ def _build_zip() -> bytes: @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.", + '/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."}, + 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, + '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", + '/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."}, + 401: {'description': 'Invalid or missing API key.'}, }, ) async def execute( @@ -1099,31 +1085,27 @@ async def execute( 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.", + 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.", + 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_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 - ) + 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 - ) + 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 @@ -1131,55 +1113,51 @@ async def execute( wait = EXECUTE_TIMEOUT if wait is not None: try: - await asyncio.wait_for( - asyncio.shield(background_process.log_task), timeout=wait - ) + 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 - ) + 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, + '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.", + '/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."}, + 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.", + 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.", + 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.", + description='Return only the last N output entries. Useful to limit response size when only recent output matters.', ge=1, ), ): @@ -1187,85 +1165,81 @@ async def get_status( if wait is None and EXECUTE_TIMEOUT: wait = EXECUTE_TIMEOUT - if wait is not None and background_process.status == "running": + if wait is not None and background_process.status == 'running': try: - await asyncio.wait_for( - asyncio.shield(background_process.log_task), timeout=wait - ) + 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 - ) + 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, + '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", + '/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."}, + 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") + 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") + 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") + raise HTTPException(status_code=400, detail='Process stdin is closed') - return {"status": "ok"} + 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.", + '/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."}, + 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."), + force: bool = Query(False, description='Send SIGKILL instead of SIGTERM.'), ): background_process = _get_process(process_id) - if background_process.status == "running": + 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.status = 'killed' background_process.exit_code = exit_code del _processes[process_id] - return {"status": "killed"} + return {'status': 'killed'} # --------------------------------------------------------------------------- @@ -1274,8 +1248,9 @@ async def kill_process( from open_terminal.utils.port import detect_listening_ports, get_descendant_pids + @app.get( - "/ports", + '/ports', include_in_schema=False, dependencies=[Depends(verify_api_key)], ) @@ -1292,26 +1267,27 @@ async def list_ports(request: Request): except Exception: # User provisioning failed (e.g. useradd rejected in restricted # container runtimes). An unprovisioned user has no ports. - return {"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] + 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] + 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) + p.pop('uid', None) - return {"ports": all_ports} + return {'ports': all_ports} # -- Port proxy client (reused across requests) -- @@ -1322,6 +1298,7 @@ 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, @@ -1330,23 +1307,23 @@ async def _get_port_proxy_client(): @app.api_route( - "/proxy/{port}/{path:path}", - methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"], + '/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") + raise HTTPException(status_code=422, detail='Port must be between 1 and 65535') - target_url = f"http://localhost:{port}/{path}" + target_url = f'http://localhost:{port}/{path}' if request.query_params: - target_url += f"?{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"): + for h in ('host', 'transfer-encoding', 'connection', 'authorization'): headers.pop(h, None) body = await request.body() @@ -1364,16 +1341,16 @@ async def port_proxy(port: int, path: str, request: Request): except httpx.ConnectError: raise HTTPException( status_code=502, - detail=f"Connection refused: localhost:{port}", + detail=f'Connection refused: localhost:{port}', ) except httpx.TimeoutException: raise HTTPException( status_code=504, - detail=f"Timeout connecting to localhost:{port}", + detail=f'Timeout connecting to localhost:{port}', ) response_headers = dict(upstream.headers) - for h in ("transfer-encoding", "connection", "content-encoding", "content-length"): + for h in ('transfer-encoding', 'connection', 'content-encoding', 'content-length'): response_headers.pop(h, None) return Response( @@ -1388,7 +1365,6 @@ async def port_proxy(port: int, path: str, request: Request): # --------------------------------------------------------------------------- if ENABLE_TERMINAL: - import uuid as _uuid from datetime import datetime as _datetime from fastapi.responses import JSONResponse @@ -1400,19 +1376,18 @@ async def port_proxy(port: int, path: str, request: Request): # Determine terminal backend: prefer Unix PTY, then pywinpty, else None if _PTY_AVAILABLE: - _TERMINAL_BACKEND = "pty" + _TERMINAL_BACKEND = 'pty' else: try: from winpty import PtyProcess as _WinPtyProcess - _TERMINAL_BACKEND = "winpty" + _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. @@ -1425,15 +1400,15 @@ def _cleanup_session(session_id: str): if session is None: return - backend = session.get("backend") + backend = session.get('backend') - if backend == "pty": + if backend == 'pty': try: - os.close(session["master_fd"]) + os.close(session['master_fd']) except OSError: pass - process = session["process"] + process = session['process'] if process.poll() is None: # Signal the whole process group first (graceful). try: @@ -1450,68 +1425,68 @@ def _cleanup_session(session_id: str): pass process.wait() - elif backend == "winpty": - pty_proc = session["pty_process"] + 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) + @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)"}, + {'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] + 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()] + 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"}, + {'error': f'Maximum number of terminal sessions ({MAX_TERMINAL_SESSIONS}) reached'}, status_code=429, ) session_id = str(_uuid.uuid4())[:8] - if _TERMINAL_BACKEND == "pty": + 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"}, + {'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)) + 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_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", + '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")] + 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) + spawn_env.setdefault('TERM', TERMINAL_TERM) process = subprocess.Popen( shell_cmd, stdin=slave_fd, @@ -1532,17 +1507,17 @@ async def create_terminal(request: Request): 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, + '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") + shell = os.environ.get('COMSPEC', 'cmd.exe') spawn_env = os.environ.copy() - spawn_env.setdefault("TERM", TERMINAL_TERM) + spawn_env.setdefault('TERM', TERMINAL_TERM) pty_proc = _WinPtyProcess.spawn( [shell], cwd=os.getcwd(), @@ -1550,29 +1525,27 @@ async def create_terminal(request: Request): dimensions=(24, 80), ) _terminal_sessions[session_id] = { - "backend": "winpty", - "pty_process": pty_proc, - "created_at": _datetime.utcnow().isoformat() + "Z", - "pid": pty_proc.pid, + '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"], + '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 + if session['backend'] == 'pty': + return session['process'].poll() is None else: - return session["pty_process"].isalive() + return session['pty_process'].isalive() - - @app.get("/api/terminals", dependencies=[Depends(verify_api_key)], include_in_schema=False) + @app.get('/api/terminals', dependencies=[Depends(verify_api_key)], include_in_schema=False) async def list_terminals(request: Request): """List active terminal sessions.""" result = [] @@ -1581,42 +1554,41 @@ async def list_terminals(request: Request): if not _session_is_alive(session): to_remove.append(sid) continue - result.append({ - "id": sid, - "created_at": session["created_at"], - "pid": session["pid"], - }) + 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) + @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) + 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 JSONResponse({'error': 'Session not found'}, status_code=404) return { - "id": session_id, - "created_at": session["created_at"], - "pid": session["pid"], + '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) + @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) + return JSONResponse({'error': 'Session not found'}, status_code=404) _cleanup_session(session_id) - return {"status": "deleted"} + return {'status': 'deleted'} - - @app.websocket("/api/terminals/{session_id}") + @app.websocket('/api/terminals/{session_id}') async def ws_terminal(ws: WebSocket, session_id: str): """Attach to an existing terminal session via WebSocket. @@ -1635,12 +1607,12 @@ async def ws_terminal(ws: WebSocket, session_id: str): """ session = _terminal_sessions.get(session_id) if session is None: - await ws.close(code=4004, reason="Session not found") + 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") + await ws.close(code=4004, reason='Session has ended') return await ws.accept() @@ -1650,22 +1622,22 @@ async def ws_terminal(ws: WebSocket, session_id: str): 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") + 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") + await ws.close(code=4001, reason='Auth timeout or invalid payload') return - backend = session["backend"] + 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"] + 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.""" @@ -1675,8 +1647,8 @@ def _blocking_read(): if rlist: return os.read(master_fd, 4096) except (OSError, ValueError): - return b"" - return b"" + return b'' + return b'' def _check_alive(): return process.poll() is None @@ -1688,27 +1660,27 @@ def _do_resize(rows: int, cols: int): fcntl.ioctl( master_fd, termios.TIOCSWINSZ, - struct.pack("HHHH", rows, cols, 0, 0), + struct.pack('HHHH', rows, cols, 0, 0), ) else: # winpty - pty_proc = session["pty_process"] + 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"" + return data.encode(errors='replace') if data else b'' except EOFError: - return b"" + return b'' except Exception: - return b"" + return b'' def _check_alive(): return pty_proc.isalive() def _write_data(data: bytes): - pty_proc.write(data.decode(errors="replace")) + pty_proc.write(data.decode(errors='replace')) def _do_resize(rows: int, cols: int): pty_proc.setwinsize(rows, cols) @@ -1738,16 +1710,16 @@ async def _pty_reader(): try: while True: msg = await ws.receive() - if msg["type"] == "websocket.disconnect": + 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"]: + 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) + 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 @@ -1773,3 +1745,353 @@ async def _pty_reader(): app.include_router(create_notebooks_router(verify_api_key)) + +# --------------------------------------------------------------------------- +# Virtual Desktop ("Computer Use") +# --------------------------------------------------------------------------- + +if ENABLE_DESKTOP: + import base64 + + from open_terminal.utils.desktop import DesktopNotRunningError, get_desktop + + class ClickRequest(BaseModel): + x: int = Field( + 0, + description='X coordinate in pixels from left edge of screen.', + ) + y: int = Field( + 0, + description='Y coordinate in pixels from top edge of screen.', + ) + button: int = Field( + 1, + description='Mouse button: 1 = left (default), 2 = middle, 3 = right.', + ) + + class MouseMoveRequest(BaseModel): + x: int = Field(..., description='X coordinate to move to.') + y: int = Field(..., description='Y coordinate to move to.') + + class DragRequest(BaseModel): + start_x: int = Field(..., description='Starting X coordinate.') + start_y: int = Field(..., description='Starting Y coordinate.') + end_x: int = Field(..., description='Ending X coordinate.') + end_y: int = Field(..., description='Ending Y coordinate.') + button: int = Field( + 1, + description='Mouse button to hold: 1 = left, 2 = middle, 3 = right.', + ) + + class TypeRequest(BaseModel): + text: str = Field(..., description='Text to type into the active window.') + human_like: bool = Field( + True, + description='When true (default), add a randomized, human-like delay between keystrokes.', + ) + + class KeyPressRequest(BaseModel): + key: str = Field( + ..., + description=('Key or key combination to press. Examples: "Return", "Escape", "ctrl+c", "alt+F4", "super".'), + ) + + class ScrollRequest(BaseModel): + x: int = Field(..., description='X coordinate to scroll at.') + y: int = Field(..., description='Y coordinate to scroll at.') + direction: str = Field( + 'down', + description="Scroll direction: 'up' or 'down'.", + pattern='^(up|down)$', + ) + amount: int = Field( + 5, + description='Number of scroll clicks (each ~3 lines).', + ge=1, + le=50, + ) + + async def _ensure_desktop(): + dm = get_desktop() + if not dm.is_running: + await dm.async_start() + + @app.exception_handler(DesktopNotRunningError) + async def desktop_not_running_handler(request: Request, exc: DesktopNotRunningError): + return JSONResponse(status_code=503, content={'detail': str(exc)}) + + async def _auto_screenshot(): + await asyncio.sleep(1.0) + dm = get_desktop() + jpg_bytes, width, height = await dm.async_annotated_screenshot() + mx, my = await asyncio.to_thread(dm.get_mouse_location) + + b64 = base64.b64encode(jpg_bytes).decode('ascii') + image_data = f'data:image/jpeg;base64,{b64}' + + return { + 'image': image_data, + 'cursor': {'x': mx, 'y': my}, + 'context': ( + f'Desktop {width}x{height} after the action. ' + f'Cursor at ({mx},{my}). ' + f'Use desktop_locate to find elements, then use the returned coordinates with desktop_click.' + ), + 'width': width, + 'height': height, + } + + @app.get( + '/desktop', + operation_id='desktop_status', + summary='Get desktop status', + description='Returns the current state of the virtual desktop: whether it is running, the display dimensions, and the VNC/noVNC ports.', + dependencies=[Depends(verify_api_key)], + responses={ + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_status(): + dm = get_desktop() + return dm.status() + + @app.post( + '/desktop/start', + operation_id='desktop_start', + summary='Start the virtual desktop', + description=( + 'Start Xvfb, x11vnc, noVNC, and a window manager. Idempotent — returns immediately if already running.' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_start(): + dm = get_desktop() + await dm.async_start() + return dm.status() + + @app.post( + '/desktop/stop', + operation_id='desktop_stop', + summary='Stop the virtual desktop', + description='Terminate all desktop processes (Xvfb, x11vnc, noVNC, window manager).', + dependencies=[Depends(verify_api_key)], + responses={ + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_stop(): + dm = get_desktop() + await dm.async_stop() + return dm.status() + + @app.post( + '/desktop/screenshot', + operation_id='desktop_screenshot', + summary='DEPRECATED: Use desktop_locate instead', + description=( + 'DEPRECATED — do NOT call this tool. ' + 'Use desktop_locate(description="...") instead to see and locate UI elements on the desktop. ' + 'desktop_locate takes a screenshot automatically and returns bounding boxes with coordinates. ' + 'Calling this tool will only return a text reminder to use desktop_locate.' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 200: { + 'description': 'Screenshot with cursor position and context.', + }, + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_screenshot(request: Request): + await _ensure_desktop() + result = await _auto_screenshot_without_action() + accept = request.headers.get('accept', '') + if 'image/png' in accept or 'image/jpeg' in accept: + dm = get_desktop() + jpg_bytes, width, height = await dm.async_annotated_screenshot() + return Response(content=jpg_bytes, media_type='image/jpeg') + return result + + async def _auto_screenshot_without_action(): + return await _auto_screenshot() + + @app.post( + '/desktop/click', + operation_id='desktop_click', + summary='Mouse click', + description=( + 'Click on the desktop. You MUST call desktop_locate first to find the correct coordinates. ' + 'After clicking, call desktop_locate again to verify the result. ' + 'Button: 1=left, 2=middle, 3=right.' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_click(request: ClickRequest): + await _ensure_desktop() + dm = get_desktop() + x, y = request.x, request.y + await asyncio.to_thread(dm.mouse_click, x, y, request.button) + return {'status': 'ok', 'x': x, 'y': y} + + @app.post( + '/desktop/mouse_move', + operation_id='desktop_mouse_move', + summary='Move the mouse', + description='Move the mouse cursor to the specified screen pixel coordinates without clicking. Use desktop_locate first to find coordinates.', + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_mouse_move(request: MouseMoveRequest): + await _ensure_desktop() + dm = get_desktop() + await asyncio.to_thread(dm.mouse_move, request.x, request.y) + return {'status': 'ok', 'x': request.x, 'y': request.y} + + @app.post( + '/desktop/mouse_location', + operation_id='desktop_mouse_location', + summary='Get mouse location', + description='Return the current mouse cursor position as (x, y) in screen pixel coordinates.', + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_mouse_location(): + await _ensure_desktop() + dm = get_desktop() + x, y = await asyncio.to_thread(dm.get_mouse_location) + return {'x': x, 'y': y} + + @app.post( + '/desktop/drag', + operation_id='desktop_drag', + summary='Drag (mouse down, move, mouse up)', + description='Press and hold a mouse button at (start_x, start_y), drag to (end_x, end_y), then release. Use desktop_locate first to find coordinates.', + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_drag(request: DragRequest): + await _ensure_desktop() + dm = get_desktop() + await asyncio.to_thread( + dm.mouse_drag, + request.start_x, + request.start_y, + request.end_x, + request.end_y, + request.button, + ) + return {'status': 'ok'} + + @app.post( + '/desktop/type', + operation_id='desktop_type', + summary='Type text', + description='Type text into the currently focused window, as if entered on a keyboard.', + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_type(request: TypeRequest): + await _ensure_desktop() + dm = get_desktop() + await asyncio.to_thread(dm.type_text, request.text, request.human_like) + return {'status': 'ok'} + + @app.post( + '/desktop/key', + operation_id='desktop_key', + summary='Press a key or key combination', + description=( + 'Press a key or key combination. Examples: "Return", "Escape", "ctrl+c", "alt+F4", "super", "ctrl+shift+t".' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_key(request: KeyPressRequest): + await _ensure_desktop() + dm = get_desktop() + await asyncio.to_thread(dm.key_press, request.key) + return {'status': 'ok'} + + @app.post( + '/desktop/scroll', + operation_id='desktop_scroll', + summary='Scroll at a position', + description='Move to (x, y) and scroll up or down by the given amount. Use desktop_locate first to find coordinates.', + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_scroll(request: ScrollRequest): + await _ensure_desktop() + dm = get_desktop() + await asyncio.to_thread(dm.scroll, request.x, request.y, request.direction, request.amount) + return {'status': 'ok'} + + @app.post( + '/desktop/windows', + operation_id='desktop_windows', + summary='List all visible windows', + description=( + 'List all visible windows on the desktop. Returns each window\'s ID, title, ' + 'position (x, y), size (width, height), PID, and whether it is the active (focused) window. ' + 'Use this to understand what applications are open and find a specific window to interact with.' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_windows(): + await _ensure_desktop() + dm = get_desktop() + windows = await dm.async_list_windows() + return {'windows': windows} + + class WindowFocusRequest(BaseModel): + window_id: str = Field(..., description='Window ID to focus (from desktop_windows).') + + @app.post( + '/desktop/window_focus', + operation_id='desktop_window_focus', + summary='Focus (bring to front) a window', + description=( + 'Activate and bring a window to the foreground by its window ID. ' + 'Use desktop_windows first to get the window ID.' + ), + dependencies=[Depends(verify_api_key)], + responses={ + 503: {'description': 'Desktop is not running.'}, + 401: {'description': 'Invalid or missing API key.'}, + }, + ) + async def desktop_window_focus(request: WindowFocusRequest): + await _ensure_desktop() + dm = get_desktop() + await dm.async_focus_window(request.window_id) + return {'status': 'ok', 'window_id': request.window_id} diff --git a/open_terminal/utils/desktop.py b/open_terminal/utils/desktop.py new file mode 100644 index 0000000..38aba06 --- /dev/null +++ b/open_terminal/utils/desktop.py @@ -0,0 +1,701 @@ +"""Virtual desktop manager — Xvfb + x11vnc + noVNC + xdotool. + +Provides a programmatic "Computer Use" capability: the agent can see the +screen (screenshots) and interact with it (mouse clicks, keyboard input) +while a human observes in real-time via the embedded noVNC web client. +""" + +import asyncio +import base64 +import io +import logging +import os +import shutil +import subprocess +import tempfile +import time + +logger = logging.getLogger(__name__) + + +class DesktopNotRunningError(RuntimeError): + pass + + +class DesktopManager: + """Manages the lifecycle of a virtual desktop environment. + + Components started (all as subprocesses of the current process): + + 1. **Xvfb** — headless X server on a virtual display. + 2. **Window manager** — ``openbox`` (or whatever is available) so windows + are usable. + 3. **x11vnc** — VNC server exposing the virtual display. + 4. **websockify + noVNC** — WebSocket proxy + embedded web VNC client so + users can view the desktop in a browser. + """ + + def __init__( + self, + display: str = ':0', + screen: str = '1280x720x24', + vnc_port: int = 5900, + novnc_port: int = 6080, + ): + self._display = display + self._screen = screen + self._vnc_port = vnc_port + self._novnc_port = novnc_port + self._xvfb: subprocess.Popen | None = None + self._x11vnc: subprocess.Popen | None = None + self._novnc: subprocess.Popen | None = None + self._wm: subprocess.Popen | None = None + self._lock = asyncio.Lock() + self._cursor_overlay = None + self._cursor_xdisplay = None + + @property + def display(self) -> str: + return self._display + + @property + def vnc_port(self) -> int: + return self._vnc_port + + @property + def novnc_port(self) -> int: + return self._novnc_port + + def _env(self) -> dict: + env = os.environ.copy() + env['DISPLAY'] = self._display + return env + + def start(self) -> None: + """Start all desktop components. + + Safe to call multiple times — subsequent calls are no-ops when the + desktop is already running. + """ + if self.is_running: + return + + env = self._env() + + self._xvfb = subprocess.Popen( + [ + 'Xvfb', + self._display, + '-screen', + '0', + self._screen, + '-ac', + '+extension', + 'GLX', + '-nolisten', + 'tcp', + ], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + time.sleep(0.5) + if self._xvfb.poll() is not None: + raise RuntimeError('Xvfb failed to start') + + os.environ['DISPLAY'] = self._display + + if os.path.isdir('/usr/share/icons/DMZ-White'): + subprocess.run( + ['xsetroot', '-cursor_name', 'left_ptr'], + env=env, + capture_output=True, + timeout=5, + ) + + wm = shutil.which('openbox') or shutil.which('xfwm4') or shutil.which('metacity') + if wm: + self._wm = subprocess.Popen( + [wm], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + time.sleep(0.3) + + self._x11vnc = subprocess.Popen( + [ + 'x11vnc', + '-display', + self._display, + '-rfbport', + str(self._vnc_port), + '-nopw', + '-listen', + '127.0.0.1', + '-shared', + '-forever', + '-noxdamage', + '-dontdisconnect', + '-cursor', + 'X', + ], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + time.sleep(0.3) + + novnc_path = '/usr/share/novnc' + if os.path.isdir(novnc_path): + self._novnc = subprocess.Popen( + [ + 'websockify', + '--web', + novnc_path, + str(self._novnc_port), + f'localhost:{self._vnc_port}', + ], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + logger.info( + 'Desktop started: display=%s vnc=%d novnc=%d', + self._display, + self._vnc_port, + self._novnc_port, + ) + + async def async_start(self) -> None: + """Async wrapper around :meth:`start`.""" + async with self._lock: + await asyncio.to_thread(self.start) + + def stop(self) -> None: + """Terminate all desktop subprocesses.""" + self._destroy_cursor_overlay() + for proc in (self._novnc, self._x11vnc, self._wm, self._xvfb): + if proc and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=2) + self._novnc = None + self._x11vnc = None + self._wm = None + self._xvfb = None + os.environ.pop('DISPLAY', None) + + async def async_stop(self) -> None: + """Async wrapper around :meth:`stop`.""" + async with self._lock: + await asyncio.to_thread(self.stop) + + @property + def is_running(self) -> bool: + return self._xvfb is not None and self._xvfb.poll() is None + + def _require_running(self) -> None: + if not self.is_running: + raise DesktopNotRunningError('Desktop is not running. Start it with POST /desktop/start.') + + def get_screen_size(self) -> tuple[int, int]: + """Return ``(width, height)`` of the virtual display.""" + self._require_running() + result = subprocess.run( + ['xdotool', 'getdisplaygeometry'], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + parts = result.stdout.strip().split() + if len(parts) >= 2: + return int(parts[0]), int(parts[1]) + parts = self._screen.split('x') + return int(parts[0]), int(parts[1]) + + def get_mouse_location(self) -> tuple[int, int]: + """Return ``(x, y)`` of the mouse in screen coordinates.""" + self._require_running() + result = subprocess.run( + ['xdotool', 'getmouselocation'], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + x = y = 0 + for part in result.stdout.strip().split(): + if part.startswith('x:'): + x = int(part[2:]) + elif part.startswith('y:'): + y = int(part[2:]) + return x, y + + def screenshot(self) -> tuple[bytes, int, int]: + """Capture a PNG screenshot. + + Returns ``(png_bytes, width, height)``. + """ + self._require_running() + with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp: + tmp_path = tmp.name + try: + subprocess.run( + [ + 'scrot', + '-o', + '-q', + '90', + tmp_path, + ], + env=self._env(), + check=True, + capture_output=True, + timeout=10, + ) + with open(tmp_path, 'rb') as f: + data = f.read() + w, h = self.get_screen_size() + return data, w, h + except FileNotFoundError: + subprocess.run( + ['import', '-window', 'root', '-quality', '90', tmp_path], + env=self._env(), + check=True, + capture_output=True, + timeout=10, + ) + with open(tmp_path, 'rb') as f: + data = f.read() + w, h = self.get_screen_size() + return data, w, h + finally: + try: + os.unlink(tmp_path) + except OSError: + pass + + async def async_screenshot(self) -> tuple[bytes, int, int]: + return await asyncio.to_thread(self.screenshot) + + def annotated_screenshot(self) -> tuple[bytes, int, int]: + png_bytes, real_w, real_h = self.screenshot() + from PIL import Image, ImageDraw + + img = Image.open(io.BytesIO(png_bytes)).convert('RGB') + draw = ImageDraw.Draw(img) + + mx, my = self.get_mouse_location() + draw.ellipse([mx - 5, my - 5, mx + 5, my + 5], fill='red') + draw.ellipse([mx - 3, my - 3, mx + 3, my + 3], fill='white') + + buf = io.BytesIO() + img.save(buf, format='JPEG', quality=85) + return buf.getvalue(), real_w, real_h + + async def async_annotated_screenshot(self) -> tuple[bytes, int, int]: + return await asyncio.to_thread(self.annotated_screenshot) + + def mouse_move(self, x: int, y: int) -> None: + self._require_running() + subprocess.run( + ['xdotool', 'mousemove', str(x), str(y)], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + self._update_cursor_overlay(x, y) + + def mouse_click(self, x: int, y: int, button: int = 1) -> None: + self._require_running() + self._hide_cursor_overlay() + subprocess.run( + [ + 'xdotool', + 'mousemove', + str(x), + str(y), + 'click', + str(button), + ], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + self._update_cursor_overlay(x, y) + time.sleep(0.3) + + def _update_cursor_overlay(self, x: int, y: int) -> None: + from Xlib import X + from Xlib.display import Display as XDisplay + from Xlib.ext import shape + + try: + if self._cursor_xdisplay is None: + self._cursor_xdisplay = XDisplay(self._display) + + disp = self._cursor_xdisplay + screen = disp.screen() + + size = 13 + half = size // 2 + win_w = size + 4 + win_h = size + 4 + cx = half + 2 + cy = half + 2 + + if self._cursor_overlay is not None: + try: + self._cursor_overlay.unmap() + self._cursor_overlay.destroy() + except Exception: + pass + + overlay = screen.root.create_window( + x - half - 2, + y - half - 2, + win_w, + win_h, + 0, + screen.root_depth, + X.InputOutput, + screen.root_visual, + background_pixel=0, + override_redirect=True, + event_mask=0, + ) + + mask = overlay.create_pixmap(win_w, win_h, 1) + gc0 = mask.create_gc(foreground=0) + mask.fill_rectangle(gc0, 0, 0, win_w, win_h) + gc1 = mask.create_gc(foreground=1) + mask.line(gc1, cx - half, cy, cx + half, cy) + mask.line(gc1, cx, cy - half, cx, cy + half) + shape.mask(overlay, shape.SO.Set, shape.SK.Bounding, 0, 0, mask) + + white = screen.default_colormap.alloc_color(65535, 65535, 65535).pixel + gc_w = overlay.create_gc(foreground=white, line_width=2) + overlay.line(gc_w, cx - half, cy, cx + half, cy) + overlay.line(gc_w, cx, cy - half, cx, cy + half) + + overlay.map() + disp.flush() + self._cursor_overlay = overlay + except Exception: + self._cursor_overlay = None + self._cursor_xdisplay = None + + def _hide_cursor_overlay(self) -> None: + if self._cursor_overlay is not None: + try: + self._cursor_overlay.unmap() + if self._cursor_xdisplay is not None: + self._cursor_xdisplay.flush() + except Exception: + pass + + def _destroy_cursor_overlay(self) -> None: + if self._cursor_overlay is not None: + try: + self._cursor_overlay.unmap() + self._cursor_overlay.destroy() + if self._cursor_xdisplay is not None: + self._cursor_xdisplay.flush() + except Exception: + pass + self._cursor_overlay = None + if self._cursor_xdisplay is not None: + try: + self._cursor_xdisplay.close() + except Exception: + pass + self._cursor_xdisplay = None + + def mouse_down(self, x: int, y: int, button: int = 1) -> None: + self._require_running() + self._hide_cursor_overlay() + subprocess.run( + ['xdotool', 'mousemove', str(x), str(y)], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + subprocess.run( + ['xdotool', 'mousedown', str(button)], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + + def mouse_up(self, x: int, y: int, button: int = 1) -> None: + self._require_running() + self._hide_cursor_overlay() + subprocess.run( + ['xdotool', 'mousemove', str(x), str(y)], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + subprocess.run( + ['xdotool', 'mouseup', str(button)], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + self._update_cursor_overlay(x, y) + + def mouse_drag( + self, + start_x: int, + start_y: int, + end_x: int, + end_y: int, + button: int = 1, + ) -> None: + self._require_running() + env = self._env() + self._hide_cursor_overlay() + subprocess.run( + ['xdotool', 'mousemove', str(start_x), str(start_y)], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + subprocess.run( + ['xdotool', 'mousedown', str(button)], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + subprocess.run( + [ + 'xdotool', + 'mousemove', + '--sync', + str(end_x), + str(end_y), + ], + env=env, + check=True, + capture_output=True, + timeout=10, + ) + subprocess.run( + ['xdotool', 'mouseup', str(button)], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + self._update_cursor_overlay(end_x, end_y) + + def type_text(self, text: str, human_like: bool = True) -> None: + self._require_running() + env = self._env() + if human_like: + for char in text: + delay = max(5, min(120, int(30 + 25 * ((ord(char) * 7 % 11) / 10.0)))) + subprocess.run( + ['xdotool', 'type', '--clearmodifiers', '--delay', str(delay), char], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + time.sleep(delay / 1000.0 + 0.005 * (ord(char) % 5)) + else: + subprocess.run( + ['xdotool', 'type', '--clearmodifiers', '--delay', '12', text], + env=env, + check=True, + capture_output=True, + timeout=max(10, len(text) // 5), + ) + + def key_press(self, key: str) -> None: + self._require_running() + subprocess.run( + ['xdotool', 'key', '--clearmodifiers', key], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + + def scroll(self, x: int, y: int, direction: str, amount: int = 5) -> None: + """Scroll at ``(x, y)``. *direction* is ``"up"`` or ``"down"``.""" + self._require_running() + env = self._env() + self._hide_cursor_overlay() + subprocess.run( + ['xdotool', 'mousemove', str(x), str(y)], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + button = 5 if direction == 'down' else 4 + for _ in range(amount): + subprocess.run( + ['xdotool', 'click', str(button)], + env=env, + check=True, + capture_output=True, + timeout=5, + ) + time.sleep(0.02) + self._update_cursor_overlay(x, y) + + def list_windows(self) -> list[dict]: + self._require_running() + result = subprocess.run( + ['xdotool', 'search', '--onlyvisible', '--name', ''], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + window_ids = [w.strip() for w in result.stdout.strip().splitlines() if w.strip()] + if not window_ids: + return [] + + cmd = ['xprop', '-root', '_NET_CLIENT_LIST'] + root_result = subprocess.run(cmd, env=self._env(), capture_output=True, text=True, timeout=5) + client_windows = set() + for line in root_result.stdout.splitlines(): + for part in line.split(): + try: + client_windows.add(str(int(part, 16))) + except (ValueError, TypeError): + pass + try: + if part.startswith('0x'): + client_windows.add(str(int(part, 16))) + except (ValueError, TypeError): + pass + + windows = [] + for wid in window_ids: + try: + name_result = subprocess.run( + ['xdotool', 'getwindowname', wid], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + name = name_result.stdout.strip() + + geom_result = subprocess.run( + ['xdotool', 'getwindowgeometry', '--shell', wid], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + geom = {} + for line in geom_result.stdout.strip().splitlines(): + if '=' in line: + k, v = line.split('=', 1) + geom[k] = int(v) + + pid_result = subprocess.run( + ['xdotool', 'getwindowpid', wid], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + pid = pid_result.stdout.strip() + + active_result = subprocess.run( + ['xdotool', 'getactivewindow'], + env=self._env(), + capture_output=True, + text=True, + timeout=5, + ) + active = active_result.stdout.strip() == wid + + windows.append({ + 'id': wid, + 'name': name or '(untitled)', + 'x': geom.get('X', 0), + 'y': geom.get('Y', 0), + 'width': geom.get('WIDTH', 0), + 'height': geom.get('HEIGHT', 0), + 'pid': pid if pid.isdigit() else None, + 'active': active, + }) + except Exception: + continue + + return windows + + async def async_list_windows(self) -> list[dict]: + return await asyncio.to_thread(self.list_windows) + + def focus_window(self, window_id: str) -> None: + self._require_running() + subprocess.run( + ['xdotool', 'windowactivate', '--sync', window_id], + env=self._env(), + check=True, + capture_output=True, + timeout=5, + ) + + async def async_focus_window(self, window_id: str) -> None: + await asyncio.to_thread(self.focus_window, window_id) + + def status(self) -> dict: + """Return a dict describing the current desktop state.""" + running = self.is_running + info: dict = { + 'running': running, + 'display': self._display, + 'vnc_port': self._vnc_port, + 'novnc_port': self._novnc_port, + } + if running: + w, h = self.get_screen_size() + info['screen_width'] = w + info['screen_height'] = h + return info + + +_desktop: DesktopManager | None = None + + +def get_desktop() -> DesktopManager: + """Return (and lazily create) the global :class:`DesktopManager`.""" + global _desktop + if _desktop is None: + from open_terminal.env import ( + DESKTOP_DISPLAY, + DESKTOP_NOVNC_PORT, + DESKTOP_SCREEN_SIZE, + DESKTOP_VNC_PORT, + ) + + _desktop = DesktopManager( + display=DESKTOP_DISPLAY, + screen=DESKTOP_SCREEN_SIZE, + vnc_port=DESKTOP_VNC_PORT, + novnc_port=DESKTOP_NOVNC_PORT, + ) + return _desktop diff --git a/test_desktop.sh b/test_desktop.sh new file mode 100755 index 0000000..b9947f0 --- /dev/null +++ b/test_desktop.sh @@ -0,0 +1,200 @@ +#!/usr/bin/env bash +# --------------------------------------------------------------------------- +# Integration tests for the virtual desktop ("Computer Use") feature. +# +# Expects the container to be already running with: +# BASE_URL – e.g. http://localhost:8000 +# API_KEY – the bearer token +# --------------------------------------------------------------------------- +set -euo pipefail + +BASE_URL="${BASE_URL:-http://localhost:8000}" +API_KEY="${API_KEY:-test-secret-key}" + +PASS=0 +FAIL=0 + +# -- helpers ---------------------------------------------------------------- + +api() { + local method="$1" path="$2" + shift 2 + curl -sf -X "$method" \ + "${BASE_URL}${path}" \ + -H "Authorization: Bearer ${API_KEY}" \ + -H "Content-Type: application/json" \ + "$@" 2>/dev/null +} + +api_raw() { + local method="$1" path="$2" + shift 2 + curl -s -X "$method" \ + "${BASE_URL}${path}" \ + -H "Authorization: Bearer ${API_KEY}" \ + -H "Content-Type: application/json" \ + -o /dev/null -w '%{http_code}' \ + "$@" 2>/dev/null +} + +assert_eq() { + local label="$1" expected="$2" actual="$3" + if [ "$expected" = "$actual" ]; then + echo " PASS: $label" + PASS=$((PASS + 1)) + else + echo " FAIL: $label (expected=$expected actual=$actual)" + FAIL=$((FAIL + 1)) + fi +} + +assert_contains() { + local label="$1" needle="$2" haystack="$3" + if echo "$haystack" | grep -q "$needle"; then + echo " PASS: $label" + PASS=$((PASS + 1)) + else + echo " FAIL: $label (response did not contain '$needle')" + FAIL=$((FAIL + 1)) + fi +} + +assert_file_png() { + local label="$1" file="$2" + local magic + magic=$(xxd -l4 -p "$file" 2>/dev/null || echo "") + if [ "$magic" = "89504e47" ]; then + echo " PASS: $label" + PASS=$((PASS + 1)) + else + echo " FAIL: $label (not a valid PNG, magic=$magic)" + FAIL=$((FAIL + 1)) + fi +} + +# -- tests ------------------------------------------------------------------ + +echo "=== Test 1: Health check ===" +health=$(api GET /health) +assert_contains "health returns ok" '"status":"ok"' "$health" + +echo "" +echo "=== Test 2: Config includes desktop feature ===" +config=$(api GET /api/config) +assert_contains "desktop feature flag present" '"desktop":true' "$config" + +echo "" +echo "=== Test 3: Desktop status (before start) ===" +status=$(api GET /desktop) +assert_contains "desktop reports not running" '"running":false' "$status" + +echo "" +echo "=== Test 4: Screenshot fails when desktop not running ===" +code=$(api_raw POST /desktop/screenshot) +assert_eq "screenshot returns 503" "503" "$code" + +echo "" +echo "=== Test 5: Start the desktop ===" +status=$(api POST /desktop/start) +sleep 3 +assert_contains "desktop started" '"running":true' "$status" + +echo "" +echo "=== Test 6: Desktop status (after start) ===" +status=$(api GET /desktop) +assert_contains "desktop running" '"running":true' "$status" +assert_contains "screen width present" '"screen_width"' "$status" +assert_contains "screen height present" '"screen_height"' "$status" + +echo "" +echo "=== Test 7: Screenshot (base64 JSON) ===" +resp=$(api POST /desktop/screenshot) +assert_contains "screenshot has width" '"width"' "$resp" +assert_contains "screenshot has height" '"height"' "$resp" +assert_contains "screenshot has data" '"data"' "$resp" +assert_contains "screenshot format is png" '"format":"png"' "$resp" + +echo "" +echo "=== Test 8: Screenshot (raw binary PNG) ===" +tmp_raw="/tmp/desktop_screenshot_raw.png" +api POST "/desktop/screenshot?format=raw" -o "$tmp_raw" >/dev/null 2>&1 || true +assert_file_png "raw screenshot is valid PNG" "$tmp_raw" + +echo "" +echo "=== Test 9: Mouse click ===" +resp=$(api POST /desktop/click -d '{"x":100,"y":100,"button":1}') +assert_contains "click returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 10: Mouse move ===" +resp=$(api POST /desktop/mouse_move -d '{"x":200,"y":200}') +assert_contains "mouse_move returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 11: Type text ===" +resp=$(api POST /desktop/type -d '{"text":"hello world"}') +assert_contains "type returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 12: Key press ===" +resp=$(api POST /desktop/key -d '{"key":"Return"}') +assert_contains "key returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 13: Complex key combo ===" +resp=$(api POST /desktop/key -d '{"key":"ctrl+a"}') +assert_contains "key combo returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 14: Scroll ===" +resp=$(api POST /desktop/scroll -d '{"x":640,"y":360,"direction":"down","amount":3}') +assert_contains "scroll returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 15: Drag ===" +resp=$(api POST /desktop/drag -d '{"start_x":100,"start_y":100,"end_x":300,"end_y":300,"button":1}') +assert_contains "drag returns ok" '"status":"ok"' "$resp" + +echo "" +echo "=== Test 16: Second screenshot (after interactions) ===" +tmp_raw2="/tmp/desktop_screenshot_raw2.png" +api POST "/desktop/screenshot?format=raw" -o "$tmp_raw2" >/dev/null 2>&1 || true +assert_file_png "second screenshot is valid PNG" "$tmp_raw2" + +echo "" +echo "=== Test 17: Stop the desktop ===" +status=$(api POST /desktop/stop) +assert_contains "desktop stopped" '"running":false' "$status" + +echo "" +echo "=== Test 18: Desktop status after stop ===" +status=$(api GET /desktop) +assert_contains "desktop not running after stop" '"running":false' "$status" + +echo "" +echo "=== Test 19: Screenshot fails after stop ===" +code=$(api_raw POST /desktop/screenshot) +assert_eq "screenshot returns 503 after stop" "503" "$code" + +echo "" +echo "=== Test 20: Restart desktop (idempotent start) ===" +status=$(api POST /desktop/start) +sleep 3 +assert_contains "desktop restarted" '"running":true' "$status" +status2=$(api POST /desktop/start) +assert_contains "second start is idempotent" '"running":true' "$status2" + +echo "" +echo "=== Test 21: Config endpoint with desktop feature ===" +config=$(api GET /api/config) +assert_contains "desktop in config" '"desktop"' "$config" + +# -- cleanup -- +rm -f /tmp/desktop_screenshot_raw.png /tmp/desktop_screenshot_raw2.png + +echo "" +echo "===========================================" +echo " Results: $PASS passed, $FAIL failed" +echo "===========================================" + +[ "$FAIL" -eq 0 ] && exit 0 || exit 1