reudy-net/panel
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add volume management features including file upload, download, and clear actions

Browse source 
Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
Jakub Dorfman 2026-04-28 23:30:22 +02:00
parent e3e910c9cd
commit 9b0c3aa190
3 changed files with 379 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

164
panel-api.py
View file

@ -311,6 +311,168 @@ class Handler(BaseHTTPRequestHandler):
self._json(200, {"ok": True, "app": parse_env_blob(result["stdout"])})
return
# /apps/<name>/volume/files
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
name = parts[1]
app, err = read_app_info(name)
if err is not None:
self._json(404, err)
return
subpath = query.get("path", [""])[0].strip("/")
data_dir = os.path.join(app["APP_VOLUME_DIR"], "data")
target_dir = os.path.abspath(os.path.join(data_dir, subpath))
# Ensure traversal didn't escape data_dir
if not target_dir.startswith(os.path.abspath(data_dir)):
self._json(403, {"ok": False, "error": "path traversal denied"})
return
if not os.path.exists(target_dir):
self._json(404, {"ok": False, "error": "directory not found"})
return
if not os.path.isdir(target_dir):
self._json(400, {"ok": False, "error": "target is not a directory"})
return
files = []
for item in os.listdir(target_dir):
if item == "." or item == "..":
continue
item_path = os.path.join(target_dir, item)
try:
stat = os.stat(item_path)
files.append({
"name": item,
"is_dir": os.path.isdir(item_path),
"size": stat.st_size,
"mtime": stat.st_mtime
})
except OSError:
continue
files.sort(key=lambda x: (not x["is_dir"], x["name"].lower()))
self._json(200, {"ok": True, "path": subpath, "files": files})
return
# /apps/<name>/volume/download
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "download":
name = parts[1]
app, err = read_app_info(name)
if err is not None:
self._json(404, err)
return
subpath = query.get("path", [""])[0].strip("/")
if not subpath:
self._json(400, {"ok": False, "error": "path parameter required"})
return
data_dir = os.path.join(app["APP_VOLUME_DIR"], "data")
target_file = os.path.abspath(os.path.join(data_dir, subpath))
if not target_file.startswith(os.path.abspath(data_dir)):
self._json(403, {"ok": False, "error": "path traversal denied"})
return
if not os.path.isfile(target_file):
self._json(404, {"ok": False, "error": "file not found"})
return
self.send_response(200)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Disposition", f'attachment; filename="{os.path.basename(target_file)}"')
size = os.path.getsize(target_file)
self.send_header("Content-Length", str(size))
self.end_headers()
with open(target_file, "rb") as fh:
while True:
chunk = fh.read(65536)
if not chunk:
break
self.wfile.write(chunk)
return
self._json(404, {"ok": False, "error": "not found"})
# ── PUT ──
def do_PUT(self):
path, parts, query = self._parse_path()
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
name = parts[1]
app, err = read_app_info(name)
if err is not None:
self._json(404, err)
return
subpath = query.get("path", [""])[0].strip("/")
if not subpath:
self._json(400, {"ok": False, "error": "path parameter required"})
return
data_dir = os.path.join(app["APP_VOLUME_DIR"], "data")
target_file = os.path.abspath(os.path.join(data_dir, subpath))
if not target_file.startswith(os.path.abspath(data_dir)):
self._json(403, {"ok": False, "error": "path traversal denied"})
return
try:
os.makedirs(os.path.dirname(target_file), exist_ok=True)
length = int(self.headers.get("Content-Length", "0"))
with open(target_file, "wb") as fh:
bytes_read = 0
while bytes_read < length:
chunk = self.rfile.read(min(65536, length - bytes_read))
if not chunk:
break
fh.write(chunk)
bytes_read += len(chunk)
self._json(200, {"ok": True, "path": subpath})
except Exception as exc:
self._json(500, {"ok": False, "error": str(exc)})
return
self._json(404, {"ok": False, "error": "not found"})
# ── DELETE ──
def do_DELETE(self):
path, parts, query = self._parse_path()
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
name = parts[1]
app, err = read_app_info(name)
if err is not None:
self._json(404, err)
return
subpath = query.get("path", [""])[0].strip("/")
if not subpath:
self._json(400, {"ok": False, "error": "path parameter required"})
return
data_dir = os.path.join(app["APP_VOLUME_DIR"], "data")
target_file = os.path.abspath(os.path.join(data_dir, subpath))
if not target_file.startswith(os.path.abspath(data_dir)):
self._json(403, {"ok": False, "error": "path traversal denied"})
return
if not os.path.exists(target_file):
self._json(404, {"ok": False, "error": "file or directory not found"})
return
try:
if os.path.isdir(target_file):
import shutil
shutil.rmtree(target_file)
else:
os.remove(target_file)
self._json(200, {"ok": True, "deleted": subpath})
except Exception as exc:
self._json(500, {"ok": False, "error": str(exc)})
return
self._json(404, {"ok": False, "error": "not found"})
# ── POST ──
@ -401,7 +563,7 @@ class Handler(BaseHTTPRequestHandler):
return
# Simple panelctl pass-through actions
if action in {"deploy", "stop", "restart", "render-route"}:
if action in {"deploy", "stop", "restart", "render-route", "volume-clear"}:
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return