Merge pull request #2780 from yueguobin/mcp/node-file-streaming

feat: Node file streaming, recursive listing, file type detection, file delete, and MCP file tools
This commit is contained in:
Jeremy Grossmann 2026-06-10 19:41:30 +02:00 committed by GitHub
commit 5df385fabe
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 664 additions and 145 deletions

View File

@ -782,6 +782,81 @@ class Gns3Connector:
_url = f"{self.base_url}/projects/{project_id}/files/{encoded_path}"
self.http_call("post", _url, data=content, headers={"Content-Type": "text/plain"})
def list_node_files(
self, project_id: str, node_id: str,
path: str = "", recursive: bool = False
) -> list[dict[str, Any]]:
"""
List files in a node directory with metadata.
**Required Attributes:**
- `project_id`
- `node_id`
- `path` Subdirectory path within node directory (optional)
- `recursive` Whether to recursively list all files (optional)
**Returns:**
List of file objects with name, path, size, modified time, type, etc.
"""
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files"
_params = {}
if path:
_params["path"] = path
if recursive:
_params["recursive"] = "true"
_response_data = self.http_call("get", _url, params=_params if _params else None)
return cast(list[dict[str, Any]], _response_data.json())
def get_node_file(self, project_id: str, node_id: str, file_path: str) -> str:
"""
Get the content of a file in a node directory.
**Required Attributes:**
- `project_id`
- `node_id`
- `file_path`
**Returns**
File content as text string
"""
encoded_path = quote(file_path, safe="/")
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
_response = self.http_call("get", _url)
return _response.text
def write_node_file(self, project_id: str, node_id: str, file_path: str, content: str) -> None:
"""
Write content to a file in a node directory. Creates the file if it doesn't exist.
**Required Attributes:**
- `project_id`
- `node_id`
- `file_path`
- `content`
"""
encoded_path = quote(file_path, safe="/")
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
self.http_call("post", _url, data=content, headers={"Content-Type": "text/plain"})
def delete_node_file(self, project_id: str, node_id: str, file_path: str) -> None:
"""
Delete a file from the node directory.
**Required Attributes:**
- `project_id`
- `node_id`
- `file_path`
"""
encoded_path = quote(file_path, safe="/")
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
self.http_call("delete", _url)
def get_computes(self) -> list[dict[str, Any]]:
"""
Returns a list of computes.

View File

@ -19,13 +19,14 @@ API routes for projects.
"""
import os
import shutil
import urllib.parse
import logging
log = logging.getLogger()
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status, Query
from fastapi.encoders import jsonable_encoder
from fastapi.responses import FileResponse
from typing import List
@ -129,59 +130,6 @@ async def delete_compute_project(project: Project = Depends(dep_project)) -> Non
await project.delete()
ProjectManager.instance().remove_project(project.id)
# @Route.get(
# r"/projects/{project_id}/notifications",
# description="Receive notifications about the project",
# parameters={
# "project_id": "Project UUID",
# },
# status_codes={
# 200: "End of stream",
# 404: "The project doesn't exist"
# })
# async def notification(request, response):
#
# pm = ProjectManager.instance()
# project = pm.get_project(request.match_info["project_id"])
#
# response.content_type = "application/json"
# response.set_status(200)
# response.enable_chunked_encoding()
#
# response.start(request)
# queue = project.get_listen_queue()
# ProjectHandler._notifications_listening.setdefault(project.id, 0)
# ProjectHandler._notifications_listening[project.id] += 1
# await response.write("{}\n".format(json.dumps(ProjectHandler._getPingMessage())).encode("utf-8"))
# while True:
# try:
# (action, msg) = await asyncio.wait_for(queue.get(), 5)
# if hasattr(msg, "asdict"):
# msg = json.dumps({"action": action, "event": msg.asdict()}, sort_keys=True)
# else:
# msg = json.dumps({"action": action, "event": msg}, sort_keys=True)
# log.debug("Send notification: %s", msg)
# await response.write(("{}\n".format(msg)).encode("utf-8"))
# except asyncio.TimeoutError:
# await response.write("{}\n".format(json.dumps(ProjectHandler._getPingMessage())).encode("utf-8"))
# project.stop_listen_queue(queue)
# if project.id in ProjectHandler._notifications_listening:
# ProjectHandler._notifications_listening[project.id] -= 1
# def _getPingMessage(cls):
# """
# Ping messages are regularly sent to the client to
# keep the connection open. We send with it some information about server load.
#
# :returns: hash
# """
# stats = {}
# # Non blocking call in order to get cpu usage. First call will return 0
# stats["cpu_usage_percent"] = CpuPercent.get(interval=None)
# stats["memory_usage_percent"] = psutil.virtual_memory().percent
# stats["disk_usage_percent"] = psutil.disk_usage(get_default_project_directory()).percent
# return {"action": "ping", "event": stats}
@router.get("/projects/{project_id}/files", response_model=List[schemas.ProjectFile])
async def get_compute_project_files(project: Project = Depends(dep_project)) -> List[schemas.ProjectFile]:
@ -196,14 +144,16 @@ async def get_compute_project_files(project: Project = Depends(dep_project)) ->
async def get_compute_node_files(
node_type: str,
node_id: str,
project: Project = Depends(dep_project)
project: Project = Depends(dep_project),
path: str = Query("", description="Subdirectory path within node directory"),
recursive: bool = Query(False, description="Recursively list all files")
) -> List[schemas.NodeFile]:
"""
Return files belonging to a specific node with detailed metadata.
"""
node_path = f"project-files/{node_type}/{node_id}"
return await project.list_node_files(node_path)
return await project.list_node_files(node_path, subpath=path, recursive=recursive)
@router.get("/projects/{project_id}/files/{file_path:path}")
@ -238,20 +188,49 @@ async def write_compute_project_file(
# Raise error if user try to escape
if not is_safe_path(path, project.path):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Path is outside the project directory")
path = os.path.join(project.path, path)
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
try:
with open(path, "wb+") as f:
async for chunk in request.stream():
f.write(chunk)
except (UnicodeEncodeError, OSError) as e:
pass # FIXME
with open(path, "wb+") as f:
async for chunk in request.stream():
f.write(chunk)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
except PermissionError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Permission denied writing to '{path}'")
except OSError as e:
log.error(f"Error writing file '{path}': {e}")
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
@router.delete("/projects/{project_id}/files/{file_path:path}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_compute_project_file(
file_path: str,
project: Project = Depends(dep_project)
) -> None:
file_path = urllib.parse.unquote(file_path)
path = os.path.normpath(file_path)
if not is_safe_path(path, project.path):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Path is outside the project directory")
path = os.path.join(project.path, path)
if not os.path.exists(path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
try:
if os.path.isdir(path):
shutil.rmtree(path)
else:
os.remove(path)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
except PermissionError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Permission denied deleting '{path}'")
except OSError as e:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

View File

@ -24,6 +24,7 @@ import ipaddress
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Request, Response, status, Query, HTTPException
from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse
from fastapi.routing import APIRoute
from typing import List, Callable, Optional
from uuid import UUID
@ -523,17 +524,30 @@ async def delete_disk_image(
@router.get("/{node_id}/files", response_model=List[schemas.NodeFile], dependencies=[Depends(has_privilege("Node.Audit"))])
async def list_node_files(node: Node = Depends(dep_node)) -> List[schemas.NodeFile]:
async def list_node_files(
node: Node = Depends(dep_node),
path: str = Query("", description="Subdirectory path within node directory"),
recursive: bool = Query(False, description="Recursively list all files")
) -> List[schemas.NodeFile]:
"""
List files in a node directory with detailed metadata.
By default lists only the current directory level (non-recursive).
Use recursive=true for a full recursive listing.
Required privilege: Node.Audit
"""
node_type = node.node_type
url = f"/projects/{node.project.id}/nodes/{node_type}/{node.id}/files"
params = {}
if path:
params["path"] = path
if recursive:
params["recursive"] = "true"
res = await node.compute.http_query(
"GET",
f"/projects/{node.project.id}/nodes/{node_type}/{node.id}/files",
"GET", url,
params=params if params else None,
timeout=None
)
return res.json
@ -550,14 +564,32 @@ async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
path = force_unix_path(file_path)
# Raise error if user try to escape
if path[0] == ".":
if path.startswith(".."):
raise ControllerForbiddenError("It is forbidden to get a file outside the project directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
res = await node.compute.http_query("GET", f"/projects/{node.project.id}/files{path}", timeout=None, raw=True)
return Response(res.body, media_type="application/octet-stream", status_code=res.status)
compute_resp = await node.compute.http_query(
"GET", f"/projects/{node.project.id}/files{path}",
timeout=None, stream=True
)
async def streamer():
try:
async for chunk in compute_resp.content.iter_chunked(65536):
yield chunk
except (IOError, OSError, asyncio.TimeoutError) as e:
log.error(f"Error streaming file '{path}' from compute: {e}")
raise
finally:
compute_resp.close()
return StreamingResponse(
streamer(),
media_type="application/octet-stream",
status_code=compute_resp.status,
)
@router.post(
@ -575,15 +607,43 @@ async def post_file(file_path: str, request: Request, node: Node = Depends(dep_n
path = force_unix_path(file_path)
# Raise error if user try to escape
if path[0] == ".":
if path.startswith(".."):
raise ControllerForbiddenError("Cannot write outside the node directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
data = await request.body() # FIXME: are we handling timeout or large files correctly?
await node.compute.http_query("POST", f"/projects/{node.project.id}/files{path}", data=data, timeout=None, raw=True)
# FIXME: response with correct status code (from compute)
# Stream request body directly to compute node
await node.compute.http_query(
"POST", f"/projects/{node.project.id}/files{path}",
data=request.stream(), timeout=None
)
@router.delete(
"/{node_id}/files/{file_path:path}",
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(has_privilege("Node.Modify"))]
)
async def delete_node_file(file_path: str, node: Node = Depends(dep_node)) -> None:
"""
Delete a file from the node directory.
Required privilege: Node.Modify
"""
path = force_unix_path(file_path)
if path.startswith(".."):
raise ControllerForbiddenError("It is forbidden to delete a file outside the project directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
await node.compute.http_query(
"DELETE", f"/projects/{node.project.id}/files{path}",
timeout=None
)
@router.websocket("/{node_id}/console/ws")

View File

@ -56,6 +56,8 @@ from .nodes import (
stop_node_handler, reload_node_handler, suspend_node_handler,
create_node_handler, delete_node_handler, update_node_handler,
get_node_console_info_handler,
list_node_files_handler, get_node_file_handler,
write_node_file_handler, delete_node_file_handler,
)
from .links import (
get_links_handler, get_link_handler, create_link_handler,
@ -588,6 +590,74 @@ async def get_compute_images(
})
# ── Node file tools ────────────────────────────────────────────────────
@mcp.tool()
async def list_node_files(
project_id: Annotated[str, Field(description="UUID of the project")],
node_id: Annotated[str, Field(description="UUID of the node")],
path: Annotated[str, Field(description="Subdirectory path within node directory (optional)")] = "",
recursive: Annotated[bool, Field(description="Recursively list all files (optional, default: false)")] = False,
) -> list[dict[str, Any]]:
"""List files in a node directory with metadata (name, size, type, modified time).
Use this first to check file sizes before reading files with get_node_file.
Large config files should be read in chunks using offset/limit.
"""
return await asyncio.to_thread(_run_handler_sync, list_node_files_handler, {
"project_id": project_id, "node_id": node_id, "path": path, "recursive": recursive,
})
@mcp.tool()
async def get_node_file(
project_id: Annotated[str, Field(description="UUID of the project")],
node_id: Annotated[str, Field(description="UUID of the node")],
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
offset: Annotated[int, Field(description="Line offset to start reading from (optional, default: 0)")] = 0,
limit: Annotated[int, Field(description="Maximum number of lines to return (optional, default: 200)")] = 200,
) -> list[dict[str, Any]]:
"""Read a text file from a node directory line-by-line with offset/limit support.
Best practice:
1. First call list_node_files to see the file size before deciding to read.
2. Start with offset=0, limit=200 to preview the file.
3. If metadata.has_more is true, read more by increasing offset.
Large files (>50KB) are auto-truncated; check the metadata.truncated flag.
For binary files, check the file type via list_node_files first.
"""
return await asyncio.to_thread(_run_handler_sync, get_node_file_handler, {
"project_id": project_id, "node_id": node_id, "file_path": file_path,
"offset": offset, "limit": limit,
})
@mcp.tool()
async def write_node_file(
project_id: Annotated[str, Field(description="UUID of the project")],
node_id: Annotated[str, Field(description="UUID of the node")],
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
content: Annotated[str, Field(description="Content to write to the file")],
) -> list[dict[str, Any]]:
"""Write content to a file in a node directory. Creates the file if it doesn't exist. Overwrites existing content."""
return await asyncio.to_thread(_run_handler_sync, write_node_file_handler, {
"project_id": project_id, "node_id": node_id, "file_path": file_path, "content": content,
})
@mcp.tool()
async def delete_node_file(
project_id: Annotated[str, Field(description="UUID of the project")],
node_id: Annotated[str, Field(description="UUID of the node")],
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
) -> list[dict[str, Any]]:
"""Delete a file from a node directory. Cannot be undone."""
return await asyncio.to_thread(_run_handler_sync, delete_node_file_handler, {
"project_id": project_id, "node_id": node_id, "file_path": file_path,
})
# ── Authwrapped SSE app ──────────────────────────────────────────────
def _make_auth_wrapper(inner_app):

View File

@ -28,6 +28,12 @@ import logging
log = logging.getLogger(__name__)
# ── Constants ──────────────────────────────────────────────────────────────
# Maximum bytes to return from get_node_file (safety net).
# Larger files are truncated with a truncated=True flag.
MAX_NODE_FILE_BYTES = 50 * 1024 # 50 KiB
# ── Helper ─────────────────────────────────────────────────────────────────
@ -166,6 +172,89 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An
return result
# ── Node file handlers ────────────────────────────────────────────────────
def list_node_files_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
project_id = params.get("project_id")
node_id = params.get("node_id")
if not project_id or not node_id:
return {"error": "project_id and node_id are required"}
conn = _get_connector(gns3_ctx)
files = conn.list_node_files(
project_id=project_id,
node_id=node_id,
path=params.get("path", ""),
recursive=params.get("recursive", False),
)
return {"files": files, "count": len(files)}
def get_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
project_id = params.get("project_id")
node_id = params.get("node_id")
file_path = params.get("file_path")
if not project_id or not node_id or not file_path:
return {"error": "project_id, node_id and file_path are required"}
offset = params.get("offset", 0)
limit = params.get("limit", 200)
conn = _get_connector(gns3_ctx)
raw = conn.get_node_file(project_id=project_id, node_id=node_id, file_path=file_path)
total_bytes = len(raw.encode("utf-8"))
truncated = False
if total_bytes > MAX_NODE_FILE_BYTES:
raw = raw[:MAX_NODE_FILE_BYTES]
truncated = True
lines = raw.splitlines(keepends=False)
total_lines = len(lines)
# Apply offset/limit
selected = lines[offset: offset + limit] if offset < total_lines else []
has_more = (offset + limit) < total_lines or truncated
return {
"file_path": file_path,
"content": "\n".join(selected),
"metadata": {
"total_lines": total_lines,
"total_bytes": total_bytes,
"offset": offset,
"limit": limit,
"returned_lines": len(selected),
"returned_bytes": len("\n".join(selected).encode("utf-8")),
"truncated": truncated or has_more,
"has_more": has_more,
},
}
def write_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
project_id = params.get("project_id")
node_id = params.get("node_id")
file_path = params.get("file_path")
content = params.get("content")
if not project_id or not node_id or not file_path or content is None:
return {"error": "project_id, node_id, file_path and content are required"}
conn = _get_connector(gns3_ctx)
conn.write_node_file(project_id=project_id, node_id=node_id, file_path=file_path, content=content)
return {"message": f"File {file_path} written to node {node_id}", "file_path": file_path, "node_id": node_id}
def delete_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
project_id = params.get("project_id")
node_id = params.get("node_id")
file_path = params.get("file_path")
if not project_id or not node_id or not file_path:
return {"error": "project_id, node_id and file_path are required"}
conn = _get_connector(gns3_ctx)
conn.delete_node_file(project_id=project_id, node_id=node_id, file_path=file_path)
return {"message": f"File {file_path} deleted from node {node_id}", "file_path": file_path, "node_id": node_id}
# ── Tool definitions ───────────────────────────────────────────────────────
NODE_TOOLS = [
@ -305,4 +394,71 @@ NODE_TOOLS = [
},
"handler": get_node_console_info_handler,
},
{
"name": "list_node_files",
"description": "List files in a node directory with metadata (name, size, type, modified time). "
"Use recursive=true for a full recursive listing. "
"Check file sizes before reading large files with get_node_file.",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "Project UUID"},
"node_id": {"type": "string", "description": "Node UUID"},
"path": {"type": "string", "description": "Subdirectory path within node directory (optional)"},
"recursive": {"type": "boolean", "description": "Recursively list all files (optional, default: false)"},
},
"required": ["project_id", "node_id"],
},
"handler": list_node_files_handler,
},
{
"name": "get_node_file",
"description": "Read a text file from a node directory. Returns file content line-by-line with offset/limit support. "
"Best practice: start with offset=0, limit=200 to preview, then increase offset to read more. "
"Large files (>50KB) are auto-truncated; check the metadata.truncated flag. "
"For binary files, check file type via list_node_files first.",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "Project UUID"},
"node_id": {"type": "string", "description": "Node UUID"},
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
"offset": {"type": "integer", "description": "Line offset to start reading from (optional, default: 0)"},
"limit": {"type": "integer", "description": "Maximum number of lines to return (optional, default: 200)"},
},
"required": ["project_id", "node_id", "file_path"],
},
"handler": get_node_file_handler,
},
{
"name": "write_node_file",
"description": "Write content to a file in a node directory. Creates the file if it doesn't exist. "
"Overwrites existing content. Useful for updating configuration files on nodes.",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "Project UUID"},
"node_id": {"type": "string", "description": "Node UUID"},
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
"content": {"type": "string", "description": "Content to write to the file"},
},
"required": ["project_id", "node_id", "file_path", "content"],
},
"handler": write_node_file_handler,
},
{
"name": "delete_node_file",
"description": "Delete a file from a node directory. Cannot be undone. "
"Use list_node_files to confirm the file path before deleting.",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "Project UUID"},
"node_id": {"type": "string", "description": "Node UUID"},
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
},
"required": ["project_id", "node_id", "file_path"],
},
"handler": delete_node_file_handler,
},
]

View File

@ -669,6 +669,12 @@ class DockerVM(BaseNode):
await self.manager.query("POST", f"containers/{self._cid}/start")
await asyncio.sleep(0.5) # give the Docker container some time to start
# Fix host-side directory ownership after Docker (re)creates
# volume mount points as root (rootful Docker only).
# This allows the GNS3 process to write files into node directories
# while the container is running. Permissions are recorded and
# restored inside the container by init.sh on next startup.
# await self._fix_permissions()
self._namespace = await self._get_namespace()
await self._start_ubridge(require_privileged_access=True)
@ -772,11 +778,19 @@ class DockerVM(BaseNode):
' && /gns3/bin/busybox chown {uid}:{gid} -R "{path}"'.format(
uid=os.getuid(), gid=os.getgid(), path=volume
),
stderr=asyncio.subprocess.PIPE,
)
except OSError as e:
raise DockerError(f"Could not fix permissions for {volume}: {e}")
await process.wait()
self._permissions_fixed = True
if process.returncode != 0:
stderr = (await process.stderr.read()).decode(errors="replace").strip()
log.error(
"Failed to fix permissions on '%s' for container '%s': %s",
volume, self._name, stderr or f"exit code {process.returncode}"
)
else:
self._permissions_fixed = True
async def _start_vnc_process(self, restart=False):
"""

View File

@ -16,6 +16,7 @@
import os
import shutil
import magic
import asyncio
import hashlib
import datetime
@ -424,65 +425,148 @@ class Project:
return files
async def list_node_files(self, node_path: str):
async def list_node_files(self, node_path: str, subpath: str = "", recursive: bool = False):
"""
List files in a specific node directory with detailed metadata.
:param node_path: Relative path to node directory (e.g., "project-files/qemu/node-id")
:param subpath: Optional subdirectory path. Defaults to root of node directory.
:param recursive: If True, recursively list all files (use with caution on large directories).
:returns: Array of files in the node directory with metadata
"""
node_full_path = os.path.normpath(os.path.join(self.path, node_path))
subpath = subpath.lstrip("/")
# Security check: ensure the path is within the project directory
if not os.path.commonpath([node_full_path, self.path]) == self.path:
if subpath:
target_path = os.path.normpath(os.path.join(node_full_path, subpath))
else:
target_path = node_full_path
# Security check: ensure the path is within the node directory
if not os.path.commonpath([target_path, node_full_path]) == node_full_path:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
detail="Path is outside the project directory")
if not os.path.exists(node_full_path):
detail="Path is outside the node directory")
if not os.path.exists(target_path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="Node directory not found")
detail="Path not found")
if recursive:
return await self._list_node_files_recursive(node_full_path, target_path,
node_full_path if subpath else None)
# Non-recursive: list only the current directory level
files = []
try:
filenames = os.listdir(node_full_path)
except OSError as e:
log.error(f"Error listing node directory: {e}")
scandir_iter = os.scandir(target_path)
except PermissionError:
return files
for filename in filenames:
file_path = os.path.join(node_full_path, filename)
if not os.path.isfile(file_path) or filename.endswith(".ghost"):
continue
except OSError as e:
log.error(f"Error listing node directory '{target_path}': {e}")
return files
for entry in scandir_iter:
name = entry.name
rel_path = name if not subpath else os.path.join(subpath, name)
try:
# Get file stat information
stat_info = await wait_run_in_executor(os.stat, file_path)
# Get file extension
_, extension = os.path.splitext(filename)
extension = extension.lstrip('.')
# Format timestamps as ISO 8601
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError) as e:
log.warning(f"Invalid timestamp for '{filename}': {e}")
created_at = modified_at = ""
file_info = {
"path": filename,
"size": stat_info.st_size,
"created_at": created_at,
"modified_at": modified_at,
"extension": extension
}
files.append(file_info)
except OSError as e:
log.warning(f"Error getting metadata for file '{filename}': {e}")
stat_info = await wait_run_in_executor(lambda e=entry: e.stat())
is_dir = await wait_run_in_executor(lambda e=entry: e.is_dir())
if is_dir:
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
files.append({
"path": rel_path,
"size": stat_info.st_size,
"created_at": created_at,
"modified_at": modified_at,
"file_type": "directory"
})
else:
if name.endswith(".ghost"):
continue
try:
file_type = await wait_run_in_executor(
lambda e=entry: magic.from_file(e.path, mime=False)
)
except Exception as e:
log.warning(f"Error getting file type for '{rel_path}': {e}")
file_type = ""
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
files.append({
"path": rel_path,
"size": stat_info.st_size,
"created_at": created_at,
"modified_at": modified_at,
"file_type": file_type
})
except OSError:
continue
return files
async def _list_node_files_recursive(self, node_full_path, start_path, base_path=None):
"""
Recursively list all files and directories under start_path.
"""
if base_path is None:
base_path = node_full_path
files = []
for dirpath, dirnames, filenames in os.walk(start_path, followlinks=False):
for dirname in dirnames:
dir_full_path = os.path.join(dirpath, dirname)
try:
stat_info = await wait_run_in_executor(os.stat, dir_full_path)
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
files.append({
"path": os.path.relpath(dir_full_path, base_path),
"size": stat_info.st_size,
"created_at": created_at,
"modified_at": modified_at,
"file_type": "directory"
})
except OSError:
continue
for filename in filenames:
if filename.endswith(".ghost"):
continue
file_path = os.path.join(dirpath, filename)
rel_path = os.path.relpath(file_path, base_path)
try:
stat_info = await wait_run_in_executor(os.stat, file_path)
try:
file_type = await wait_run_in_executor(
lambda fp=file_path: magic.from_file(fp, mime=False)
)
except Exception as e:
log.warning(f"Error getting file type for '{rel_path}': {e}")
file_type = ""
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError) as e:
log.warning(f"Invalid timestamp for '{rel_path}': {e}")
created_at = modified_at = ""
files.append({
"path": rel_path,
"size": stat_info.st_size,
"created_at": created_at,
"modified_at": modified_at,
"file_type": file_type
})
except OSError as e:
log.warning(f"Error getting metadata for file '{rel_path}': {e}")
continue
return files
def _hash_file(self, path):

View File

@ -32,7 +32,7 @@ else:
from async_timeout import timeout as asynctimeout
from ..utils import parse_version
from ..utils.asyncio import locking
from ..utils.asyncio import locking, async_iterable_to_stream
from ..controller.controller_error import (
ControllerError,
ControllerBadRequestError,
@ -341,9 +341,11 @@ class Compute:
raise ControllerNotFoundError(f"{image} not found on compute")
return response
async def http_query(self, method, path, data=None, dont_connect=False, **kwargs):
async def http_query(self, method, path, data=None, dont_connect=False, stream=False, params=None, **kwargs):
"""
:param dont_connect: If true do not reconnect if not connected
:param stream: If True, return raw aiohttp response for streaming
:param params: Optional dict of query parameters to append to the URL
"""
if not self._connected and not dont_connect:
@ -352,7 +354,7 @@ class Compute:
await self.connect()
if not self._connected and not dont_connect:
raise ComputeError(f"Cannot connect to compute '{self._name}' with request {method} {path}")
response = await self._run_http_query(method, path, data=data, **kwargs)
response = await self._run_http_query(method, path, data=data, stream=stream, params=params, **kwargs)
return response
async def _try_reconnect(self):
@ -515,7 +517,7 @@ class Compute:
""" Returns URL for specific path at Compute"""
return self._getUrl(path)
async def _run_http_query(self, method, path, data=None, timeout=120, raw=False):
async def _run_http_query(self, method, path, data=None, timeout=120, raw=False, stream=False, params=None):
async with asynctimeout(delay=timeout):
url = self._getUrl(path)
headers = {"content-type": "application/json"}
@ -531,6 +533,11 @@ class Compute:
elif isinstance(data, aiohttp.streams.StreamReader) or isinstance(data, bytes):
chunked = True
headers["content-type"] = "application/octet-stream"
# Stream from an async iterable (e.g. Starlette request.stream())
elif hasattr(data, "__aiter__"):
chunked = True
headers["content-type"] = "application/octet-stream"
data = await async_iterable_to_stream(data)
# If the data is an open file we will iterate on it
elif isinstance(data, io.BufferedIOBase):
chunked = True
@ -540,7 +547,7 @@ class Compute:
try:
log.debug(f"Attempting request to compute: {method} {url} {headers}")
response = await self._session().request(
method, url, headers=headers, data=data, auth=self._auth, chunked=chunked, timeout=timeout
method, url, headers=headers, data=data, auth=self._auth, params=params, chunked=chunked, timeout=timeout
)
except asyncio.TimeoutError:
raise ComputeError(f"Timeout error for {method} call to {url} after {timeout}s")
@ -554,6 +561,18 @@ class Compute:
) as e:
# aiohttp 2.3.1 raises socket.gaierror when cannot find host
raise ComputeError(str(e))
if stream:
if response.status >= 300:
body = await response.read()
msg = body.decode() if body else ""
if response.status == 404:
raise ControllerNotFoundError(f"{method} {path} not found")
elif response.status == 403:
raise ControllerForbiddenError(msg)
raise ControllerError(f"HTTP {response.status}: {msg}")
return response
body = await response.read()
if body and not raw:
body = body.decode()

View File

@ -114,7 +114,7 @@ class NodeFile(BaseModel):
size: int = Field(..., description="File size in bytes")
created_at: str = Field(..., description="File creation time (ISO 8601)")
modified_at: str = Field(..., description="File modification time (ISO 8601)")
extension: str = Field(..., description="File extension")
file_type: str = Field(..., description="File type determined by the file command")
class ProjectCompression(str, Enum):

View File

@ -136,3 +136,44 @@ def locking(f):
return await f(oself, *args, **kwargs)
return wrapper
async def async_iterable_to_stream(async_iter, limit=65536):
"""
Convert an async iterable into an aiohttp StreamReader.
This avoids passing async generators directly to aiohttp's payload
system, which can cause compatibility issues with certain HTTP servers.
:param async_iter: An async iterable that yields bytes
:param limit: Buffer limit for the StreamReader (default 64KB)
:returns: aiohttp.streams.StreamReader
"""
from aiohttp.streams import StreamReader
class _NoopProtocol:
_reading_paused = False
connected = True
def pause_reading(self):
self._reading_paused = True
def resume_reading(self, resume_parser=False):
self._reading_paused = False
reader = StreamReader(_NoopProtocol(), limit=limit)
async def _feed():
try:
async for chunk in async_iter:
reader.feed_data(chunk)
except GeneratorExit:
raise
except Exception:
pass
finally:
reader.feed_eof()
asyncio.ensure_future(_feed())
return reader

View File

@ -35,6 +35,9 @@ urllib3>=2.7.0
fastmcp>=3.4.0
# ==============================================================================
# File type detection
python-magic>=0.4.27
# AI Copilot Optional Dependencies
# ==============================================================================
# AI Copilot features are now optional. Install with:

View File

@ -553,24 +553,33 @@ class TestNodeRoutes:
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.body = b"world"
response.status = status.HTTP_200_OK
compute.http_query = AsyncioMagicMock(return_value=response)
# Mock the streaming response
async def mock_iter_chunked(chunk_size):
yield b"world"
mock_stream = AsyncioMagicMock()
mock_stream.iter_chunked = mock_iter_chunked
mock_stream.close = MagicMock()
mock_response = AsyncioMagicMock()
mock_response.status = status.HTTP_200_OK
mock_response.content = mock_stream
compute.http_query = AsyncioMagicMock(return_value=mock_response)
response = await client.get(app.url_path_for("get_file", project_id=project.id, node_id=node.id, file_path="hello"))
assert response.status_code == status.HTTP_200_OK
assert response.content == b'world'
compute.http_query.assert_called_with(
"GET",
"/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(
project_id=project.id,
node_id=node.id),
timeout=None,
raw=True)
stream=True)
response = await client.get(app.url_path_for(
"get_file",
project_id=project.id,
@ -595,8 +604,15 @@ class TestNodeRoutes:
node_id=node.id,
file_path="hello"), content=b"hello")
assert response.status_code == status.HTTP_201_CREATED
compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
# Verify http_query was called with stream parameter
compute.http_query.assert_called_once()
call_args = compute.http_query.call_args
assert call_args[0][0] == "POST"
assert call_args[0][1] == "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id)
assert call_args[1]["timeout"] is None
# data should be an async generator from request.stream()
assert hasattr(call_args[1]["data"], "__aiter__")
response = await client.get("/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_404_NOT_FOUND

View File

@ -1735,9 +1735,10 @@ async def test_fix_permission(vm):
vm._volumes = ["/etc"]
vm._get_container_state = AsyncioMagicMock(return_value="running")
process = MagicMock()
process.returncode = 0
with asyncio_patch("asyncio.subprocess.create_subprocess_exec", return_value=process) as mock_exec:
await vm._fix_permissions()
mock_exec.assert_called_with('docker', 'exec', 'e90e34656842', '/gns3/bin/busybox', 'sh', '-c', '(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(os.getuid(), os.getgid()))
mock_exec.assert_called_with('docker', 'exec', 'e90e34656842', '/gns3/bin/busybox', 'sh', '-c', '(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(os.getuid(), os.getgid()), stderr=asyncio.subprocess.PIPE)
assert process.wait.called
@ -1747,10 +1748,11 @@ async def test_fix_permission_not_running(vm):
vm._volumes = ["/etc"]
vm._get_container_state = AsyncioMagicMock(return_value="stopped")
process = MagicMock()
process.returncode = 0
with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_start:
with asyncio_patch("asyncio.subprocess.create_subprocess_exec", return_value=process) as mock_exec:
await vm._fix_permissions()
mock_exec.assert_called_with('docker', 'exec', 'e90e34656842', '/gns3/bin/busybox', 'sh', '-c', '(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(os.getuid(), os.getgid()))
mock_exec.assert_called_with('docker', 'exec', 'e90e34656842', '/gns3/bin/busybox', 'sh', '-c', '(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(os.getuid(), os.getgid()), stderr=asyncio.subprocess.PIPE)
assert mock_start.called
assert process.wait.called

View File

@ -87,7 +87,7 @@ async def test_compute_httpQuery(compute):
response.status = 200
await compute.post("/projects", {"a": "b"})
await compute.close()
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", data=b'{"a": "b"}', headers={'content-type': 'application/json'}, auth=None, chunked=None, timeout=120)
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", headers={'content-type': 'application/json'}, data=b'{"a": "b"}', auth=None, params=None, chunked=None, timeout=120)
assert compute._auth is None
@ -102,7 +102,7 @@ async def test_compute_httpQueryAuth(compute):
compute.password = SecretStr("toor")
await compute.post("/projects", {"a": "b"})
await compute.close()
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", data=b'{"a": "b"}', headers={'content-type': 'application/json'}, auth=compute._auth, chunked=None, timeout=120)
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", headers={'content-type': 'application/json'}, data=b'{"a": "b"}', auth=compute._auth, params=None, chunked=None, timeout=120)
assert compute._auth.login == "root"
assert compute._auth.password == "toor"
@ -162,7 +162,7 @@ async def test_compute_httpQueryNotConnectedInvalidVersion(compute):
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
with pytest.raises(ControllerError):
await compute.post("/projects", {"a": "b"})
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, chunked=None, timeout=120)
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=120)
await compute.close()
@ -176,7 +176,7 @@ async def test_compute_httpQueryNotConnectedNonGNS3Server(compute):
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
with pytest.raises(ControllerError):
await compute.post("/projects", {"a": "b"})
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, chunked=None, timeout=120)
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=120)
await compute.close()
@ -190,7 +190,7 @@ async def test_compute_httpQueryNotConnectedNonGNS3Server2(compute):
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
with pytest.raises(ControllerError):
await compute.post("/projects", {"a": "b"})
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, chunked=None, timeout=120)
mock.assert_any_call("GET", "https://example.com:84/v3/compute/capabilities", headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=120)
@pytest.mark.asyncio
@ -228,7 +228,7 @@ async def test_compute_httpQuery_project(compute):
project = Project(name="Test")
mock_notification.assert_called()
await compute.post("/projects", project)
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", data=json.dumps(project.asdict()), headers={'content-type': 'application/json'}, auth=None, chunked=None, timeout=120)
mock.assert_called_with("POST", "https://example.com:84/v3/compute/projects", headers={'content-type': 'application/json'}, data=json.dumps(project.asdict()), auth=None, params=None, chunked=None, timeout=120)
await compute.close()
# FIXME: https://github.com/aio-libs/aiohttp/issues/2525
@ -372,7 +372,7 @@ async def test_forward_get(compute):
response.status = 200
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
await compute.forward("GET", "qemu", "images")
mock.assert_called_with("GET", "https://example.com:84/v3/compute/qemu/images", auth=None, data=None, headers={'content-type': 'application/json'}, chunked=None, timeout=None)
mock.assert_called_with("GET", "https://example.com:84/v3/compute/qemu/images", headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=None)
await compute.close()
@ -395,7 +395,7 @@ async def test_forward_post(compute):
response.status = 200
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
await compute.forward("POST", "qemu", "img", data={"id": 42})
mock.assert_called_with("POST", "https://example.com:84/v3/compute/qemu/img", auth=None, data=b'{"id": 42}', headers={'content-type': 'application/json'}, chunked=None, timeout=None)
mock.assert_called_with("POST", "https://example.com:84/v3/compute/qemu/img", headers={'content-type': 'application/json'}, data=b'{"id": 42}', auth=None, params=None, chunked=None, timeout=None)
await compute.close()
@ -408,7 +408,7 @@ async def test_list_files(project, compute):
response.status = 200
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
assert await compute.list_files(project) == res
mock.assert_any_call("GET", "https://example.com:84/v3/compute/projects/{}/files".format(project.id), auth=None, chunked=None, data=None, headers={'content-type': 'application/json'}, timeout=None)
mock.assert_any_call("GET", "https://example.com:84/v3/compute/projects/{}/files".format(project.id), headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=None)
await compute.close()
@ -430,7 +430,7 @@ async def test_interfaces(compute):
response.status = 200
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
assert await compute.interfaces() == res
mock.assert_any_call("GET", "https://example.com:84/v3/compute/network/interfaces", auth=None, chunked=None, data=None, headers={'content-type': 'application/json'}, timeout=120)
mock.assert_any_call("GET", "https://example.com:84/v3/compute/network/interfaces", headers={'content-type': 'application/json'}, data=None, auth=None, params=None, chunked=None, timeout=120)
await compute.close()