home/share
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden server and fix client-side edge cases

Browse Source 
Python: per-file error handling in cleanup, single stat() in download,
S_ISREG check, Content-Disposition escaping, Content-Length on all
responses, route prefix passed to handlers (no magic numbers), validate
empty upload, ensure_cert handles partial cert pair, narrower exception
catches.

JS: visibility-aware polling, concurrent upload guard, fetch r.ok checks,
missing .catch() on delete/loadFiles, REFRESH_SEC→REFRESH_MS rename,
explicit parseInt radix, transition: all→specific properties.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
eberkheev
2026-03-10 20:47:47 +04:00
co-authored by Claude Opus 4.6
parent ebe5c6ec90
commit 082a80a6c5
2 changed files with 66 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
share.py
+43 -24
View File
@@ -9,6 +9,7 @@ import json
import mimetypes
import socket
import ssl
import stat
import subprocess
import threading
import time
@@ -26,12 +27,17 @@ def cleanup_loop(upload_dir: Path, max_age: int):
time.sleep(3600)
now = time.time()
try:
for f in upload_dir.iterdir():
entries = list(upload_dir.iterdir())
except OSError as e:
print(f" cleanup error: {e}")
continue
for f in entries:
try:
if f.is_file() and now - f.stat().st_mtime > max_age:
f.unlink()
print(f" - expired: {f.name}")
except Exception:
pass
except OSError:
pass
# ── SSL ──
@@ -42,7 +48,7 @@ def get_local_ip():
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
except Exception:
except OSError:
return "127.0.0.1"
@@ -53,6 +59,8 @@ def ensure_cert(cert_dir: Path, extra_ips: list[str] | None = None) -> tuple[Pat
key = cert_dir / "key.pem"
if cert.is_file() and key.is_file():
return cert, key
cert.unlink(missing_ok=True)
key.unlink(missing_ok=True)
ips = {"127.0.0.1", get_local_ip()}
if extra_ips:
@@ -79,7 +87,7 @@ def _file_type(path: Path) -> str:
if ct.startswith('text/'):
return 'text'
try:
with open(path, 'rb') as f:
with path.open('rb') as f:
return 'text' if b'\x00' not in f.read(8192) else 'other'
except OSError:
return 'other'
@@ -123,13 +131,14 @@ class Handler(http.server.BaseHTTPRequestHandler):
def _json(self, data):
self._respond(json.dumps(data).encode(), "application/json")
def _respond(self, data, ct):
def _respond(self, data: bytes, ct: str):
self.send_response(200)
self.send_header("Content-Type", ct)
self.send_header("Content-Length", str(len(data)))
self.send_header("Connection", "close")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(data if isinstance(data, bytes) else data.encode())
self.wfile.write(data)
# ── Routes ──
@@ -147,15 +156,15 @@ class Handler(http.server.BaseHTTPRequestHandler):
def _dispatch(self):
for prefix, method in self.ROUTES.get(self.command, []):
if self.path == prefix or (len(prefix) > 1 and prefix.endswith("/") and self.path.startswith(prefix)):
return getattr(self, method)()
return getattr(self, method)(prefix)
self.send_error(404)
do_GET = do_POST = do_DELETE = _dispatch
def _route_index(self):
def _route_index(self, _prefix):
self._respond(self.server.html, "text/html; charset=utf-8")
def _route_files(self):
def _route_files(self, _prefix):
entries = []
for p in self.server.upload_dir.iterdir():
if p.is_file():
@@ -163,37 +172,47 @@ class Handler(http.server.BaseHTTPRequestHandler):
entries.sort(key=lambda e: e[1].st_mtime, reverse=True)
self._json([{"name": p.name, "size": st.st_size, "type": _file_type(p)} for p, st in entries])
def _safe_path(self, prefix_len: int) -> Path:
name = urllib.parse.unquote(self.path[prefix_len:])
def _safe_path(self, prefix: str) -> Path:
name = urllib.parse.unquote(self.path[len(prefix):])
return self.server.upload_dir / Path(name).name
def _route_download(self):
safe = self._safe_path(4)
if not safe.is_file():
def _route_download(self, prefix):
safe = self._safe_path(prefix)
try:
st = safe.stat()
except FileNotFoundError:
self.send_error(404)
return
if not stat.S_ISREG(st.st_mode):
self.send_error(404)
return
ct = mimetypes.guess_type(safe.name)[0] or "application/octet-stream"
fname = safe.name.replace("\\", "\\\\").replace('"', '\\"')
self.send_response(200)
self.send_header("Content-Type", ct)
self.send_header("Content-Disposition", f'attachment; filename="{safe.name}"')
self.send_header("Content-Length", str(safe.stat().st_size))
self.send_header("Content-Disposition", f'attachment; filename="{fname}"')
self.send_header("Content-Length", str(st.st_size))
self.send_header("Connection", "close")
self.end_headers()
with open(safe, "rb") as f:
with safe.open("rb") as f:
while chunk := f.read(65536):
self.wfile.write(chunk)
def _route_upload(self):
def _route_upload(self, _prefix):
ct = self.headers.get("Content-Type", "")
if "multipart/form-data" not in ct:
self.send_error(400)
return
length = int(self.headers.get("Content-Length", 0))
if not length:
self.send_error(400)
return
body = self.rfile.read(length)
count = self._save_parts(ct, body)
self._json({"ok": True, "count": count})
def _route_delete(self):
safe = self._safe_path(5)
def _route_delete(self, prefix):
safe = self._safe_path(prefix)
if safe.is_file():
safe.unlink()
self._json({"ok": True})
@@ -254,10 +273,10 @@ def load_config() -> dict:
p = argparse.ArgumentParser(description="Share — local file sharing server")
p.add_argument("-p", "--port", type=int)
p.add_argument("-d", "--dir", type=str)
p.add_argument("-d", "--dir")
p.add_argument("--ttl", type=int)
p.add_argument("--san", nargs="*")
p.add_argument("--lang", type=str)
p.add_argument("--lang")
p.add_argument("--refresh", type=int)
args = p.parse_args()
@@ -297,7 +316,7 @@ def main():
server = ShareServer(("0.0.0.0", port), Handler, upload_dir=upload_dir, html=html, ssl_ctx=ctx)
print(f"\n Share running")
print("\n Share running")
print(f" Local: https://localhost:{port}")
print(f" Network: https://{ip}:{port}")
print(f" Files: {upload_dir}\n")