CORVO-AI's picture
Update file-manager/app.py
ff178eb verified
Raw
History Blame
9.45 kB
import os, shutil, html, urllib.parse, subprocess
from flask import Flask, request, redirect, send_file, Response
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from werkzeug.serving import run_simple
app = Flask(__name__)
app.config["APPLICATION_ROOT"] = "/fm"
SECRET_HINTS = ("TOKEN", "KEY", "SECRET", "PASSWORD", "AUTH", "COOKIE")
def is_sensitive(name):
up = name.upper()
return any(h in up for h in SECRET_HINTS)
def mask_secrets(text):
# Guide rule 10: never surface real credential values in output.
out = []
for line in text.splitlines():
if any(h in line.upper() for h in SECRET_HINTS) and ("=" in line or ":" in line):
out.append("*** line masked (possible secret) ***")
else:
out.append(line)
return "\n".join(out)
PAGE = """<!doctype html><html><head><title>Box</title>
<style>
body{{font-family:monospace;margin:20px;background:#0b0b0b;color:#eee}}
a{{color:#29BEFD;text-decoration:none}} a:hover{{text-decoration:underline}}
table{{border-collapse:collapse;width:100%}} td,th{{padding:4px 8px;border-bottom:1px solid #222;text-align:left}}
.bar{{background:#1a1a1a;padding:10px;margin-bottom:10px;border-radius:6px}}
.tabs a{{margin-right:14px;font-weight:bold}}
textarea{{width:100%;background:#000;color:#47E6C1;border:1px solid #333;padding:8px}}
pre{{background:#000;color:#78D64B;padding:12px;border-radius:6px;overflow:auto;max-height:60vh}}
input[type=text],select{{background:#000;color:#eee;border:1px solid #333;padding:6px}}
button,input[type=submit]{{background:#F46821;color:#fff;border:0;padding:6px 14px;border-radius:4px;cursor:pointer}}
</style></head><body>
<div class="tabs bar"><a href="/fm/">📁 Files</a><a href="/fm/term">🖥️ Terminal</a>
<span style="color:#777">test mode · storage is ephemeral</span></div>
{body}</body></html>"""
# ---------------- TERMINAL ----------------
@app.route("/term", methods=["GET", "POST"])
def term():
cmd = ""
cwd = request.form.get("cwd", "/") if request.method == "POST" else "/"
output = ""
if request.method == "POST":
cmd = request.form.get("cmd", "")
# allow "cd" to persist working directory between commands
if cmd.strip().startswith("cd "):
target = cmd.strip()[3:].strip()
new = os.path.abspath(os.path.join(cwd, target)) if not target.startswith("/") else target
if os.path.isdir(new):
cwd, output = new, f"(cwd -> {new})"
else:
output = f"cd: no such directory: {new}"
elif cmd.strip():
try:
r = subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True,
text=True, timeout=60)
output = (r.stdout or "") + (r.stderr or "")
if r.returncode != 0 and not output:
output = f"[exit code {r.returncode}]"
except subprocess.TimeoutExpired:
output = "[timed out after 60s]"
except Exception as e:
output = f"[error: {e}]"
output = mask_secrets(output)
body = f"""
<div class="bar"><b>cwd:</b> {html.escape(cwd)}</div>
<form method="post">
<input type="hidden" name="cwd" value="{html.escape(cwd)}">
<input type="text" name="cmd" placeholder="type a command e.g. ls -la, df -h, nproc"
style="width:80%" autofocus value="{html.escape(cmd)}">
<input type="submit" value="Run">
</form>
<div class="bar" style="margin-top:10px">Quick:
<a href="#" onclick="q('ls -la')">ls -la</a> ·
<a href="#" onclick="q('df -h')">df -h</a> ·
<a href="#" onclick="q('free -h')">free -h</a> ·
<a href="#" onclick="q('nproc')">nproc</a> ·
<a href="#" onclick="q('cat /etc/os-release')">os-release</a>
</div>
<pre>{html.escape(output)}</pre>
<script>
function q(c){{document.querySelector('[name=cmd]').value=c;
document.querySelector('form').submit();}}
</script>"""
return Response(PAGE.format(body=body))
# ---------------- FILE MANAGER ----------------
@app.route("/")
def browse():
path = os.path.abspath(request.args.get("path", "/"))
if not os.path.exists(path):
return Response(PAGE.format(body=f"<p>Not found: {html.escape(path)}</p>"), status=404)
if os.path.isfile(path):
return redirect(f"/fm/view?path={urllib.parse.quote(path)}")
parent = os.path.dirname(path.rstrip("/")) or "/"
rows = f'<tr><td><a href="/fm/?path={urllib.parse.quote(parent)}">.. (up)</a></td><td></td><td></td></tr>'
try:
entries = sorted(os.listdir(path))
except PermissionError:
return Response(PAGE.format(body=f"<p>Permission denied: {html.escape(path)}</p>"), status=403)
for name in entries:
full = os.path.join(path, name); q = urllib.parse.quote(full)
try:
size = os.path.getsize(full) if os.path.isfile(full) else ""
except OSError:
size = "?"
if os.path.isdir(full):
link = f'<a href="/fm/?path={q}">📁 {html.escape(name)}/</a>'
actions = f'<a href="/fm/delete?path={q}">delete</a>'
else:
link = f'<a href="/fm/view?path={q}">📄 {html.escape(name)}</a>'
actions = (f'<a href="/fm/download?path={q}">download</a> | '
f'<a href="/fm/edit?path={q}">edit</a> | '
f'<a href="/fm/delete?path={q}">delete</a>')
rows += f"<tr><td>{link}</td><td>{size}</td><td>{actions}</td></tr>"
body = f"""
<div class="bar"><b>Path:</b> {html.escape(path)}</div>
<div class="bar">
<form action="/fm/mkdir" method="post" style="display:inline">
<input type="hidden" name="path" value="{html.escape(path)}">
<input name="name" placeholder="new folder"><input type="submit" value="Create folder">
</form>
<form action="/fm/upload" method="post" enctype="multipart/form-data" style="display:inline">
<input type="hidden" name="path" value="{html.escape(path)}">
<input type="file" name="file"><input type="submit" value="Upload">
</form>
</div>
<table><tr><th>Name</th><th>Size</th><th>Actions</th></tr>{rows}</table>"""
return Response(PAGE.format(body=body))
@app.route("/view")
def view():
path = os.path.abspath(request.args.get("path", ""))
if not os.path.isfile(path):
return redirect("/fm/")
if is_sensitive(os.path.basename(path)):
content = "*** masked (sensitive filename) ***"
else:
try:
with open(path, "r", errors="replace") as f:
content = mask_secrets(f.read(200000))
except Exception as e:
content = f"[cannot read as text: {e}]"
q = urllib.parse.quote(path)
body = (f'<div class="bar"><b>{html.escape(path)}</b> | '
f'<a href="/fm/edit?path={q}">edit</a> | '
f'<a href="/fm/download?path={q}">download</a> | '
f'<a href="/fm/?path={urllib.parse.quote(os.path.dirname(path))}">back</a></div>'
f"<pre>{html.escape(content)}</pre>")
return Response(PAGE.format(body=body))
@app.route("/edit")
def edit():
path = os.path.abspath(request.args.get("path", ""))
try:
with open(path, "r", errors="replace") as f:
content = f.read()
except Exception as e:
content = f"[cannot open: {e}]"
body = f"""
<div class="bar"><b>Editing:</b> {html.escape(path)}</div>
<form action="/fm/save" method="post">
<input type="hidden" name="path" value="{html.escape(path)}">
<textarea name="content" style="height:60vh">{html.escape(content)}</textarea>
<br><input type="submit" value="Save (live)">
<a href="/fm/?path={urllib.parse.quote(os.path.dirname(path))}">cancel</a>
</form>"""
return Response(PAGE.format(body=body))
@app.route("/save", methods=["POST"])
def save():
path = os.path.abspath(request.form["path"])
with open(path, "w") as f:
f.write(request.form["content"])
return redirect(f"/fm/view?path={urllib.parse.quote(path)}")
@app.route("/download")
def download():
path = os.path.abspath(request.args.get("path", ""))
if is_sensitive(os.path.basename(path)):
return Response("masked (sensitive filename)", status=403)
return send_file(path, as_attachment=True)
@app.route("/mkdir", methods=["POST"])
def mkdir():
base = os.path.abspath(request.form["path"])
os.makedirs(os.path.join(base, request.form["name"]), exist_ok=True)
return redirect(f"/fm/?path={urllib.parse.quote(base)}")
@app.route("/upload", methods=["POST"])
def upload():
base = os.path.abspath(request.form["path"])
f = request.files.get("file")
if f and f.filename:
f.save(os.path.join(base, f.filename))
return redirect(f"/fm/?path={urllib.parse.quote(base)}")
@app.route("/delete")
def delete():
path = os.path.abspath(request.args.get("path", ""))
parent = os.path.dirname(path)
try:
shutil.rmtree(path) if os.path.isdir(path) else os.remove(path)
except Exception as e:
return Response(PAGE.format(body=f"<p>Delete failed: {e}</p>"), status=500)
return redirect(f"/fm/?path={urllib.parse.quote(parent)}")
if __name__ == "__main__":
wrapped = DispatcherMiddleware(Flask("empty"), {"/fm": app})
run_simple("0.0.0.0", 9001, wrapped)