# -*- coding: utf-8 -*- # GOST 文件分发源 v4 最终版: 列表 + 查看 + 编辑(保存) + 上传 + 分页 + 文件夹 + serverip # 单文件, 阿里云(relly同理可部署) import os, urllib.parse, html, json, time, re from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import os as _os PUB_DIR = _os.environ.get("MIRROR_PUB_DIR", "/tmp/pub") AUTH_USER = _os.environ.get("MIRROR_USER", "admin") AUTH_PASS = _os.environ.get("MIRROR_PASS", "oyh159159oiw.@") UPSTREAM = _os.environ.get("MIRROR_UPSTREAM", "") DESCRIPTIONS = { "gost-deploy.sh": ("🚀 GOST 三件套一键部署/管理", "安装 · 菜单管理 · 启停重启 · 升级 · 卸载(内嵌完整面板,自包含)"), "npm.sh": ("🟦 Nginx Proxy Manager 一键部署", "在线拉取 / 离线 docker load 双模式 · 启停 · 升级 · 备份 · 卸载"), "npm.tar": ("📦 NPM 离线镜像包", "docker load -i npm.tar 导入后配合 npm.sh 离线安装"), "panel.html": ("🖥️ GOST 面板页面 (v2)", "多API切换 · 启停开关 · 探活 · 限速 · yaml查看 · 操作日志 · 主题切换"), "gost-panel-server.py": ("⚙️ GOST 面板后端服务", "同源代理 · 状态监控 · 端口冲突检测 · 探活 · 操作日志接口"), "gost-save-server.py": ("💾 GOST 规则保存服务", "/save 持久化 · /yaml /rawyaml · 60秒自动同步守护"), "frp单版本": ("🔀 FRP 单版本脚本集", "FRP 服务端/客户端部署脚本"), "nps单版本": ("🔀 NPS 单版本脚本集", "NPS 服务端/客户端部署脚本"), } def auto_desc(fp, base): try: with open(fp, "rb") as f: head = f.read(4096).decode("utf-8", errors="replace") except Exception: head = "" lines = head.split("\n")[:20] title = "" subs = [] import re as _re for ln in lines: s = ln.strip().lstrip("#").strip() if not s or s.startswith("!"): continue # 跳过纯分隔线 if _re.fullmatch(r"[=\-*_]+", s): continue for key in ("描述:", "说明:", "简介:", "功能:"): if s.startswith(key): title = s[len(key):].strip() break if title: continue if s.startswith("用法") or s.startswith("Usage") or s.startswith("子命令"): if ":" in s or len(s) > 8: subs.append(s) # 带内容的说明行 elif s.startswith("bash ") or s.startswith("./"): subs.append(s) # 命令示例 elif len(s) >= 4 and not title and not subs and len(s) <= 60 and not _re.search(r"[=\-]{3,}", s): title = s title = title.strip("=").strip("-").strip(" ").strip("*").strip() if not title: title = "📄 " + base sub_text = " · ".join(subs[:2]) if subs else "脚本文件" return (title[:60], sub_text[:120]) def check_auth(headers): a = headers.get("Authorization", "") if a.startswith("Basic "): import base64 try: u, p = base64.b64decode(a[6:]).decode().split(":", 1) return u == AUTH_USER and p == AUTH_PASS except Exception: return False return False # 登录session存储(文件持久化, 重启不丢失) import secrets as _secrets _SESS_FILE = "/tmp/pub_real/.sessions" try: with open(_SESS_FILE) as _f: SESSIONS = set(l.strip() for l in _f if l.strip()) except Exception: SESSIONS = set() def _save_sessions(): try: with open(_SESS_FILE, "w") as f: f.write('\n'.join(SESSIONS)) except Exception: pass def _new_token(): return _secrets.token_hex(32) def _get_cookie(self): """从请求头解析cookie, 返回session token或None""" ck = self.headers.get("Cookie", "") for part in ck.split(";"): part = part.strip() if part.startswith("gsess="): return part[6:] return None def _is_logged_in(self): """检查请求是否已登录(session cookie 或 Basic Auth)""" tok = _get_cookie(self) if tok and tok in SESSIONS: return True return check_auth(self.headers) def safe_path(rel): fpath = os.path.realpath(os.path.join(PUB_DIR, rel)) for root, dirs, files in os.walk(PUB_DIR): for f in files: try: rp = os.path.realpath(os.path.join(root, f)) if fpath == rp: return fpath except Exception: pass for d in dirs: try: rd = os.path.realpath(os.path.join(root, d)) if fpath == rd: return fpath except Exception: pass return None def get_real_files(): result = [] for root, dirs, files in os.walk(PUB_DIR): rel_root = os.path.relpath(root, PUB_DIR) for d in dirs: rel = os.path.join(rel_root, d) if rel_root != "." else d if rel not in [r[1] for r in result]: result.append(("dir", rel, 0)) for f in files: if f.startswith(".") or f.endswith(".bak") or f == "desc.json" : continue # 隐藏文件/备份不显示 rel = os.path.join(rel_root, f) if rel_root != "." else f fp = os.path.join(root, f) if os.path.isfile(fp) or os.path.islink(fp): try: sz = os.path.getsize(os.path.realpath(fp)) except Exception: sz = 0 result.append(("file", rel, sz)) return sorted(result, key=lambda x: (x[0] == "file", x[1])) def fsize_h(n): f = float(n); i = 0 for u in ["B","KB","MB","GB"]: if f < 1024 or i == 3: break f /= 1024; i += 1 return f"{f:.1f}{u}" if i else f"{int(n)}B" def detect_ip(): import urllib.request as u for src in ("http://members.3322.org/dyndns/getip", "http://ip.3322.net", "https://api.ipify.org"): try: v = u.urlopen(src, timeout=4).read().decode().strip() if v: return v except Exception: continue try: import socket as s sk = s.socket(s.AF_INET, s.SOCK_DGRAM); sk.connect(("8.8.8.8", 80)) ip = sk.getsockname()[0]; sk.close(); return ip except Exception: return "127.0.0.1" CSS = ''':root{--bg1:#0f0c29;--bg2:#302b63;--bg3:#24243e;--pink:#ff8ac9;--purple:#b39dff;--cyan:#7af0ff;--bg:rgba(15,12,41,.75);--panel:rgba(24,18,52,.5);--panel2:rgba(30,24,66,.55);--line:#3d3868;--text:#e8e4ff;--muted:#9a94c9;--accent:#7af0ff;--green:#6ee7b7;--red:#fda4af;--amber:#fbbf24} *{box-sizing:border-box} body{font-family:"Segoe UI","PingFang SC","Microsoft YaHei",sans-serif;background:linear-gradient(rgba(15,12,41,.72),rgba(15,12,41,.72)),url("https://t.alcy.cc/moez") center/cover no-repeat fixed;margin:0;padding:40px 24px;color:var(--text);min-height:100vh;position:relative;overflow-x:hidden} .stars,.stars2,.stars3{position:fixed;inset:0;pointer-events:none;z-index:0} .stars{background-image:radial-gradient(1.5px 1.5px at 20% 30%,#fff 50%,transparent 51%),radial-gradient(1px 1px at 40% 70%,#ffe9f7 50%,transparent 51%),radial-gradient(2px 2px at 60% 20%,#d0e9ff 50%,transparent 51%),radial-gradient(1px 1px at 80% 50%,#fff 50%,transparent 51%),radial-gradient(1.5px 1.5px at 10% 80%,#fff 50%,transparent 51%),radial-gradient(1px 1px at 90% 10%,#ffe9f7 50%,transparent 51%),radial-gradient(1.5px 1.5px at 35% 45%,#fff 50%,transparent 51%),radial-gradient(1px 1px at 70% 85%,#d0e9ff 50%,transparent 51%);animation:twinkle 4s ease-in-out infinite alternate} .stars2{background-image:radial-gradient(1px 1px at 15% 15%,#fff 50%,transparent 51%),radial-gradient(1.5px 1.5px at 45% 55%,#e0c3ff 50%,transparent 51%),radial-gradient(1px 1px at 75% 35%,#fff 50%,transparent 51%),radial-gradient(2px 2px at 25% 90%,#fff 50%,transparent 51%),radial-gradient(1px 1px at 85% 75%,#ffe9f7 50%,transparent 51%);animation:twinkle 6s ease-in-out infinite alternate-reverse;opacity:.7} .stars3{background-image:radial-gradient(1px 1px at 50% 10%,#fff 50%,transparent 51%),radial-gradient(1.5px 1.5px at 30% 60%,#7af0ff55 50%,transparent 51%),radial-gradient(1px 1px at 65% 45%,#fff 50%,transparent 51%),radial-gradient(1px 1px at 5% 40%,#ffc9e055 50%,transparent 51%);animation:twinkle 8s ease-in-out infinite alternate;opacity:.5} @keyframes twinkle{0%{opacity:.85}100%{opacity:.45}} .wrap,.card,.top,.uploadform,.cmd,.vpage,.vpage *{position:relative;z-index:1} .wrap{max-width:900px;margin:0 auto} header{text-align:center;margin-bottom:30px} header h1{font-size:24px;margin:0 0 8px} header .sub{color:var(--muted);font-size:13px} header .sub code{background:var(--panel2);border:1px solid var(--line);padding:2px 10px;border-radius:99px;color:var(--accent)} .card{background:var(--panel);border:1px solid var(--line);border-radius:16px;overflow:hidden;box-shadow:0 8px 32px rgba(0,0,0,.35);backdrop-filter:blur(20px) saturate(1.35)} .fitem{display:flex;align-items:center;gap:16px;padding:16px 22px;border-bottom:1px solid var(--line);transition:background .15s} .fitem:last-child{border-bottom:none} .fitem:hover{background:#ffffff08} .ficon{font-size:26px;width:36px;text-align:center;flex-shrink:0} .finfo{flex:1;min-width:0} .fname{font-size:15px;font-weight:600} .fname a{color:var(--accent);text-decoration:none} .fname a:hover{text-decoration:underline} .fsize{color:var(--muted);font-size:12px;margin-left:8px} .fdesc{color:var(--muted);font-size:12.5px;margin-top:3px;line-height:1.5} .fbtn{flex-shrink:0;display:flex;gap:8px} .fbtn a{color:#fff;text-decoration:none;padding:7px 14px;border-radius:9px;font-size:13px;font-weight:600;display:inline-block} .fbtn a:hover{filter:brightness(1.2);box-shadow:0 0 16px rgba(179,157,255,.4)} .fbtn a.edit{background:linear-gradient(135deg,#10b981,#059669)} .fbtn a.view{background:linear-gradient(135deg,#38bdf8,#0284c7)} .fbtn a.dl{background:linear-gradient(135deg,#0ea5e9,#2563eb)} .fbtn a.copy{background:linear-gradient(135deg,#8b5cf6,#6d28d9)} .fbtn a.bash{background:linear-gradient(135deg,#059669,#047857)} .fbtn a.dir{background:linear-gradient(135deg,#6366f1,#4338ca)} .cmd{background:var(--panel2);border:1px solid var(--line);border-radius:10px;padding:14px 16px;margin-top:26px;font-size:12.5px;line-height:2;color:var(--muted)} .cmd code{background:var(--bg);border:1px solid var(--line);border-radius:6px;padding:3px 10px;color:#7dd3fc;font-size:12px;user-select:all} footer{text-align:center;color:#64748b;font-size:12px;margin-top:26px} .badge{display:inline-block;background:#10b98122;border:1px solid #10b98155;color:var(--green);border-radius:99px;padding:2px 12px;font-size:12px;margin-left:8px} input#search{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:9px;padding:8px 14px;font-size:13px;width:100%;margin-bottom:16px} .uploadform input[type=file]{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:8px;padding:8px 12px;font-size:13px} .uploadform button{background:linear-gradient(135deg,#10b981,#059669);color:#fff;border:none;border-radius:8px;padding:8px 18px;font-size:13px;font-weight:600;cursor:pointer} .uploadform{padding:0 22px 18px} .uploadform b{font-size:14px;color:var(--text)} #pager{display:flex;gap:8px;justify-content:center;padding:14px;align-items:center;font-size:13px;color:var(--muted)} #pager button{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:7px;padding:5px 12px;cursor:pointer} .msg{display:none;padding:8px 16px;border-radius:8px;font-size:13px;position:fixed;top:16px;right:16px;z-index:300} .msg.ok{background:#10b98122;border:1px solid #10b98155;color:#34d399;display:block} .msg.bad{background:#f8717122;border:1px solid #f8717155;color:#f87171;display:block} .vpage textarea{width:100%;box-sizing:border-box;min-height:66vh;background:rgba(15,12,41,.55);border:1px solid var(--line);border-radius:14px;padding:18px;font-size:13px;line-height:1.8;color:#d5f0ff;font-family:Consolas,"JetBrains Mono",monospace;white-space:pre;overflow:auto;tab-size:4;outline:none;resize:vertical;backdrop-filter:blur(18px) saturate(1.3);box-shadow:0 8px 32px rgba(0,0,0,.45)} .vpage textarea:focus{border-color:var(--purple);box-shadow:0 0 0 3px rgba(179,157,255,.15)} .vpage .top{margin-bottom:14px;display:flex;align-items:center;gap:12px;flex-wrap:wrap} .vpage .top b{font-size:15px} .vpage .top a,.vpage .top button{color:#38bdf8;text-decoration:none;cursor:pointer;background:none;border:1px solid #334155;font-size:13px;padding:6px 14px;border-radius:8px} .vpage .top button.save{background:linear-gradient(135deg,#10b981,#059669);color:#fff;border:none;font-weight:600} .vpage .top button.save:hover{filter:brightness(1.15)} /* 语法高亮 */ .kw{color:#7af0ff;font-weight:600} .str{color:#ff8ac9} .cmt{color:#6b6699;font-style:italic} .num{color:#fbbf77} .fn{color:#ffd479} .op{color:#b39dff} .shellcmd{color:#6ee7b7} ''' def icon_for(base): if "gost-deploy" in base: return "🚀" if base == "npm.sh": return "🟦" if base == "panel.html": return "🖥" if "panel-server" in base: return "⚙" if "save-server" in base: return "💾" return "📄" def build_list_page(): items = [] for typ, rel, sz in get_real_files(): fn = html.escape(rel) base = os.path.basename(rel) ic = icon_for(base) if typ == "dir": desc, sub = DESCRIPTIONS.get(rel, ("📁 文件夹", "子目录")) items.append('
' '
' + ic + '
' '
' '
' + desc + '
' + sub + '
' '
打开
') continue if rel in DESCRIPTIONS: desc, sub = DESCRIPTIONS[rel] else: desc, sub = auto_desc(safe_path(rel) or (PUB_DIR + "/" + rel), base) try: import json as _J2 with open("/tmp/pub_real/desc.json") as _f: _cd = _J2.load(_f).get(rel) if _cd and _cd[0]: desc = _cd[0] if _cd[1]: sub = _cd[1] except Exception: pass qrel = urllib.parse.quote(rel) szs = fsize_h(sz) items.append('
' '
' + ic + '
' '
' + fn + '' + szs + '
' '
' + desc + ' ✏️
' + sub + '
' '
' '📝 编辑 ' '👁 查看 ' '⬇ 下载 ' '✏️ 重命名 ' '🗑 删除' '
') CSS_TOP = CSS + """ input#search{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:9px;padding:8px 14px;font-size:13px;width:100%;margin-bottom:16px} .uploadform input[type=file]{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:8px;padding:8px 12px;font-size:13px} .uploadform button{background:linear-gradient(135deg,#10b981,#059669);color:#fff;border:none;border-radius:8px;padding:8px 18px;font-size:13px;font-weight:600;cursor:pointer} .uploadform{padding:0 22px 18px} .uploadform b{font-size:14px;color:var(--text)} #pager{display:flex;gap:8px;justify-content:center;padding:14px;align-items:center;font-size:13px;color:var(--muted)} #pager button{background:var(--panel2);border:1px solid var(--line);color:var(--text);border-radius:7px;padding:5px 12px;cursor:pointer} .fbtn a.rn{background:linear-gradient(135deg,#f59e0b,#d97706)} .fbtn a.del{background:linear-gradient(135deg,#f87171,#dc2626)}""" page = "\n\n\n" page += "\n\n" page += "GOST 分发源\n\n\n\n
\n" page += "
\n" page += "
\n" page += "

\U0001F4E6 GOST / NPM 分发源 \u25cf 在线

\n" page += "
所有文件实时同步最新版 · 使用: bash <(curl -Ls http://当前地址/文件名)
\n" page += "
\n" page += " \n" page += "
\n" page += "\n".join(items) page += "\n
\n" page += """
📤 上传文件到分发源
上传后自动出现在列表 · 单文件最大 2GB
快速使用:
GOST 三件套: bash <(curl -Ls http://当前地址/gost-deploy.sh)
NPM 面板:     bash <(curl -Ls http://当前地址/npm.sh)
已装机器升级: bash gost-deploy.sh update (源默认指向本服务)
""" return page.encode("utf-8") LOGIN_PAGE = """ 登录 - GOST 分发源

🔐 GOST 分发源

登录后可使用编辑 · 重命名 · 删除
用户名或密码错误
← 返回列表
""" def _hl(code_text): """轻量语法高亮: python/bash/js 通用""" import re as _re esc = html.escape(code_text) # 1. 注释整行先提取为占位符(避免内部被其他规则二次处理) comments = [] def _stash_cmt(m): comments.append('' + m.group(2) + '') return m.group(1) + "\x00CMT" + str(len(comments)-1) + "\x00" esc = _re.sub(r'(^|\n)(#[^\n]*)', _stash_cmt, esc) # 2. 关键字 kws = ("import|from|def|class|return|if|elif|else|for|while|try|except|finally|with|as|in|not|and|or|pass|break|continue|lambda|yield|print|True|False|None|function|var|let|const|new|this|typeof") esc = _re.sub(r'\b(' + kws + r')\b', '\1', esc) # 3. 数字 esc = _re.sub(r'\b(\d+\.?\d*)\b', '\1', esc) # 4. 还原注释占位 esc = _re.sub(r"\x00CMT(\d+)\x00", lambda m: comments[int(m.group(1))], esc) return esc class H(BaseHTTPRequestHandler): def do_GET(self): path = urllib.parse.unquote(self.path) if path.startswith("/filedesc/"): fn = path[len("/filedesc/"):] DJ = "/tmp/pub_real/desc.json" import json as _J if self.command == "GET": try: with open(DJ) as f: data = _J.load(f).get(fn, ["",""]) except Exception: data = ["",""] body = _J.dumps({"title": data[0], "sub": data[1]}).encode() self.send_response(200) self.send_header("Content-Type","application/json") self.send_header("Content-Length",str(len(body))) self.end_headers(); self.wfile.write(body); return if self.command == "POST": n = int(self.headers.get("Content-Length","0") or 0) try: d = _J.loads(self.rfile.read(n)) try: with open(DJ) as f: all_d = _J.load(f) except Exception: all_d = {} all_d[fn] = [str(d.get("title",""))[:80], str(d.get("sub",""))[:200]] with open(DJ,"w") as f: f.write(_J.dumps(all_d, ensure_ascii=False, indent=1)) r = b'{"ok":true}' self.send_response(200) except Exception as e: r = str(e).encode() self.send_response(400) self.send_header("Content-Type","application/json") self.send_header("Content-Length",str(len(r))) self.end_headers(); self.wfile.write(r); return self.send_response(405); self.end_headers(); return if path == "/serverip": ip = detect_ip() self.send_response(200) self.send_header("Content-Type","text/plain") self.send_header("Access-Control-Allow-Origin","*") self.end_headers(); self.wfile.write(ip.encode()); return # 编辑页 if path == "/login": # 已登录则跳首页 tok = _get_cookie(self) if tok and tok in SESSIONS: self.send_response(302); self.send_header("Location","/"); self.end_headers(); return page = LOGIN_PAGE body = page.encode("utf-8") self.send_response(200) self.send_header("Content-Type","text/html; charset=utf-8") self.send_header("Cache-Control","no-cache") self.send_header("Content-Length",str(len(body))) self.end_headers(); self.wfile.write(body); return if path.startswith("/edit/"): fn = path[len("/edit/"):] fpath = safe_path(fn) if not fpath or not os.path.isfile(fpath): self.send_response(404); self.end_headers(); return if not _is_logged_in(self): self.send_response(302) self.send_header("Location","/login") self.end_headers(); return with open(fpath, "rb") as f: content = f.read() try: text = content.decode("utf-8") except Exception: text = content.decode("utf-8", errors="replace") esc = html.escape(text, quote=False) qfn = urllib.parse.quote(fn) page = (f''' 编辑 {html.escape(fn)}
📝 编辑: {html.escape(fn)} 👁 只读查看 ⬇ 下载 ← 返回列表
''').encode("utf-8") self.send_response(200) self.send_header("Content-Type","text/html; charset=utf-8") self.send_header("Content-Length",str(len(page))) self.end_headers(); self.wfile.write(page); return # 只读查看页 if path.startswith("/view/"): fn = path[len("/view/"):] fpath = safe_path(fn) if not fpath or not os.path.isfile(fpath): self.send_response(404); self.end_headers(); return with open(fpath, "rb") as f: content = f.read() try: text = content.decode("utf-8") except Exception: text = content.decode("utf-8", errors="replace") esc = html.escape(text) qfn = urllib.parse.quote(fn) page = (f''' View {html.escape(fn)}
📄 {html.escape(fn)} 📝 编辑 ⬇ 下载 ← 返回列表
{_hl(text)}
''').encode("utf-8") self.send_response(200) self.send_header("Content-Type","text/html; charset=utf-8") self.send_header("Content-Length",str(len(page))) self.end_headers(); self.wfile.write(page); return # 目录页 if path == "/" or path == "/index.html": body = build_list_page() self.send_response(200) self.send_header("Content-Type","text/html; charset=utf-8") self.send_header("Content-Length",str(len(body))) self.end_headers(); self.wfile.write(body); return # 下载 rel = path.lstrip("/") if rel.startswith("dl/"): rel = rel[3:] fpath = safe_path(rel) if not fpath: self.send_response(404) self.send_header("Content-Type","text/plain; charset=utf-8") self.end_headers(); self.wfile.write("404".encode()); return if os.path.isdir(fpath): self.send_response(302); self.send_header("Location","/"); self.end_headers(); return if os.path.isfile(fpath): with open(fpath,"rb") as f: body=f.read() ctype = "text/html; charset=utf-8" if fpath.endswith(".html") else "text/plain; charset=utf-8" self.send_response(200) self.send_header("Content-Type",ctype) self.send_header("Content-Length",str(len(body))) self.end_headers(); self.wfile.write(body); return self.send_response(404); self.end_headers(); self.wfile.write("404".encode()) def do_POST(self): path = urllib.parse.unquote(self.path) if path == "/login": n = int(self.headers.get("Content-Length","0") or 0) body = self.rfile.read(n).decode("utf-8", errors="replace") import urllib.parse as UP params = UP.parse_qs(body) u = params.get("username",[""])[0] pw = params.get("password",[""])[0] if u == AUTH_USER and pw == AUTH_PASS: tok = _new_token() SESSIONS.add(tok) _save_sessions() body2 = b"OK" self.send_response(200) self.send_header("Set-Cookie","gsess=" + tok + "; Path=/; HttpOnly; SameSite=Lax; Max-Age=86400") self.send_header("Content-Type","text/plain; charset=utf-8") self.send_header("Content-Length",str(len(body2))) self.end_headers(); self.wfile.write(body2) else: body2 = b"BAD" self.send_response(401) self.send_header("Content-Type","text/plain; charset=utf-8") self.send_header("Content-Length",str(len(body2))) self.end_headers(); self.wfile.write(body2) return if path.startswith("/filedesc/"): fn = path[len("/filedesc/"):] if not _is_logged_in(self): self.send_response(401); self.end_headers(); return DJ = "/tmp/pub_real/desc.json" import json as _J n = int(self.headers.get("Content-Length","0") or 0) try: d = _J.loads(self.rfile.read(n)) try: with open(DJ) as f: all_d = _J.load(f) except Exception: all_d = {} all_d[fn] = [str(d.get("title",""))[:80], str(d.get("sub",""))[:200]] with open(DJ,"w") as f: f.write(_J.dumps(all_d, ensure_ascii=False, indent=1)) r = b'{"ok":true}' self.send_response(200) except Exception as e: r = str(e).encode() self.send_response(400) self.send_header("Content-Type","application/json") self.send_header("Content-Length",str(len(r))) self.end_headers(); self.wfile.write(r); return if path.startswith("/saveedit/"): fn = path[len("/saveedit/"):] fpath = safe_path(fn) if not fpath or not os.path.isfile(fpath): self.send_response(404); self.end_headers(); return if not _is_logged_in(self): self.send_response(401); self.end_headers(); return n = int(self.headers.get("Content-Length","0") or 0) new_content = self.rfile.read(n) with open(fpath, "wb") as f: f.write(new_content) self.send_response(200) self.send_header("Content-Type","text/plain; charset=utf-8") self.end_headers() self.wfile.write(f"OK, saved {fn} ({len(new_content)}B)".encode()); return if path == "/rename": if not _is_logged_in(self): self.send_response(401); self.end_headers(); return n = int(self.headers.get("Content-Length","0") or 0) try: import json as J data = J.loads(self.rfile.read(n)) old_rel = data.get("old","") new_rel = os.path.basename(data.get("new","").strip()) old_link = os.path.join(PUB_DIR, old_rel.strip("/")) old_fp = safe_path(old_rel) if not old_fp and not os.path.islink(old_link): self.send_response(400); self.end_headers(); return new_link = os.path.join(PUB_DIR, new_rel) if os.path.exists(new_link) or os.path.islink(new_link): self.send_response(409); self.end_headers(); self.wfile.write(b"exists"); return if os.path.islink(old_link): # 软链接: 只重命名链接本身 os.rename(old_link, new_link) else: os.rename(old_fp, new_link) self.send_response(200); self.end_headers(); self.wfile.write(b"OK") except Exception as e: self.send_response(500); self.end_headers(); self.wfile.write(str(e).encode()) return if path == "/delete": if not _is_logged_in(self): self.send_response(401); self.end_headers(); return n = int(self.headers.get("Content-Length","0") or 0) try: import json as J import shutil data = J.loads(self.rfile.read(n)) rel = data.get("path","").strip("/") link_fp = os.path.join(PUB_DIR, rel) fp = safe_path(rel) if not fp and not os.path.islink(link_fp): self.send_response(404); self.end_headers(); return if os.path.islink(link_fp): # 软链接: 只删链接本身, 不动源文件 os.unlink(link_fp) elif os.path.isdir(fp): shutil.rmtree(fp) else: os.remove(fp) self.send_response(200); self.end_headers(); self.wfile.write(b"OK") except Exception as e: self.send_response(500); self.end_headers(); self.wfile.write(str(e).encode()) return if path == "/upload": ct = self.headers.get("Content-Type","") n = int(self.headers.get("Content-Length","0") or 0) if n > 2*1024*1024*1024: self.send_response(413); self.end_headers(); return body = self.rfile.read(n) m = re.search(r'boundary="?([^";\s]+)"?', ct) if not m: self.send_response(400); self.end_headers(); self.wfile.write(b"no boundary"); return boundary = ("--"+m.group(1)).encode() parts = body.split(boundary) saved = [] for part in parts: if b"filename=" not in part: continue fnm = re.search(r'filename="([^"]+)"', part.decode("utf-8",errors="replace")) if not fnm: continue fname = os.path.basename(fnm.group(1)) fname = re.sub(r'[^\w.\-()]','_',fname) if not fname: continue didx = part.find(b"\r\n\r\n") if didx < 0: continue data = part[didx+4:] if data.endswith(b"\r\n"): data = data[:-2] dest = os.path.join(PUB_DIR, fname) with open(dest,"wb") as f: f.write(data) saved.append((fname,len(data))) if saved: names = ", ".join(f"{html.escape(s[0])}({s[1]//1024}KB)" for s in saved) page = f'''

✅ 上传成功

{names}

← 返回列表 (2秒后自动跳转)

''' else: page = '''

❌ 上传失败

返回

''' body = page.encode("utf-8") self.send_response(200) self.send_header("Content-Type","text/html; charset=utf-8") self.send_header("Content-Length",str(len(body))) self.end_headers(); self.wfile.write(body); return self.send_response(404); self.end_headers() def log_message(self,*a): pass ThreadingHTTPServer(("0.0.0.0",18999),H).serve_forever()