diff --git a/cert.py b/cert.py
new file mode 100644
index 0000000..8359788
--- /dev/null
+++ b/cert.py
@@ -0,0 +1,42 @@
+"""Self-signed certificate generation."""
+
+import socket
+import subprocess
+from pathlib import Path
+
+from config import CERT_DAYS, CERT_KEY_BITS
+
+
+def get_local_ip() -> str:
+ try:
+ with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+ s.connect(("8.8.8.8", 80))
+ return s.getsockname()[0]
+ except OSError:
+ return "127.0.0.1"
+
+
+def ensure_cert(cert_dir: Path, extra_ips: list[str] | None = None) -> tuple[Path, Path]:
+ """Generate self-signed cert if missing. Returns (certfile, keyfile)."""
+ cert_dir.mkdir(parents=True, exist_ok=True)
+ cert = cert_dir / "cert.pem"
+ 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:
+ ips.update(extra_ips)
+ san = ",".join([f"IP:{ip}" for ip in sorted(ips)] + ["DNS:localhost"])
+
+ subprocess.run([
+ "openssl", "req", "-x509", "-newkey", f"rsa:{CERT_KEY_BITS}",
+ "-keyout", str(key), "-out", str(cert),
+ "-days", str(CERT_DAYS), "-nodes",
+ "-subj", "/CN=Share",
+ "-addext", f"subjectAltName={san}",
+ ], check=True, capture_output=True)
+ print(f" Generated self-signed cert in {cert_dir}")
+ return cert, key
diff --git a/cleanup.py b/cleanup.py
new file mode 100644
index 0000000..c60c932
--- /dev/null
+++ b/cleanup.py
@@ -0,0 +1,26 @@
+"""Periodic cleanup of expired files."""
+
+import time
+from pathlib import Path
+
+from config import CLEANUP_INTERVAL
+
+
+def cleanup_loop(upload_dir: Path, max_age: int):
+ if max_age <= 0:
+ return
+ while True:
+ time.sleep(CLEANUP_INTERVAL)
+ now = time.time()
+ try:
+ 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 OSError:
+ pass
diff --git a/config.py b/config.py
new file mode 100644
index 0000000..6a636f8
--- /dev/null
+++ b/config.py
@@ -0,0 +1,62 @@
+"""Configuration: share.config loading, constants, CLI args."""
+
+import argparse
+import configparser
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent
+
+_cp = configparser.ConfigParser()
+_cp.read_dict({
+ "server": {
+ "port": "3001", "dir": str(Path.home() / "Downloads" / "shared"),
+ "ttl": "7", "san": "", "lang": "ru", "refresh": "5",
+ "per_page": "10,25,50,100",
+ },
+ "internal": {
+ "secs_per_day": "86400", "cleanup_interval": "3600",
+ "ssl_handshake_timeout": "5", "sniff_size": "8192",
+ "chunk_size": "65536", "cert_days": "3650", "cert_key_bits": "2048",
+ "toast_ms": "2500", "progress_hide_ms": "1200",
+ },
+})
+_cp.read(BASE_DIR / "share.config")
+
+SECS_PER_DAY = _cp.getint("internal", "secs_per_day")
+CLEANUP_INTERVAL = _cp.getint("internal", "cleanup_interval")
+SSL_HANDSHAKE_TIMEOUT = _cp.getint("internal", "ssl_handshake_timeout")
+SNIFF_SIZE = _cp.getint("internal", "sniff_size")
+CHUNK_SIZE = _cp.getint("internal", "chunk_size")
+CERT_DAYS = _cp.getint("internal", "cert_days")
+CERT_KEY_BITS = _cp.getint("internal", "cert_key_bits")
+
+
+def load_config() -> dict:
+ """Load config: share.config defaults <- CLI args."""
+ conf = dict(_cp["server"])
+
+ p = argparse.ArgumentParser(description="Share — local file sharing server")
+ p.add_argument("-p", "--port", type=int)
+ p.add_argument("-d", "--dir")
+ p.add_argument("--ttl", type=int)
+ p.add_argument("--san", nargs="*")
+ p.add_argument("--lang")
+ p.add_argument("--refresh", type=int)
+ args = p.parse_args()
+
+ for key, val in vars(args).items():
+ if val is not None:
+ conf[key] = " ".join(val) if isinstance(val, list) else str(val)
+
+ return conf
+
+
+def client_cfg(conf: dict) -> dict:
+ """Build config dict to inject into HTML as CFG."""
+ return {
+ "lang": conf["lang"],
+ "refresh": int(conf["refresh"]),
+ "perPage": [int(x) for x in conf["per_page"].split(",")],
+ "toastMs": _cp.getint("internal", "toast_ms"),
+ "progressHideMs": _cp.getint("internal", "progress_hide_ms"),
+ }
diff --git a/server.py b/server.py
new file mode 100644
index 0000000..23a0fdc
--- /dev/null
+++ b/server.py
@@ -0,0 +1,179 @@
+"""HTTP server and request handler."""
+
+import email.parser
+import http.server
+import json
+import mimetypes
+import ssl
+import stat
+import urllib.parse
+from datetime import datetime
+from pathlib import Path
+
+from config import CHUNK_SIZE, SNIFF_SIZE, SSL_HANDSHAKE_TIMEOUT
+
+
+def file_type(path: Path) -> str:
+ ct = mimetypes.guess_type(path.name)[0] or ''
+ if ct.startswith('image/'):
+ return 'image'
+ if ct.startswith('text/'):
+ return 'text'
+ try:
+ with path.open('rb') as f:
+ return 'text' if b'\x00' not in f.read(SNIFF_SIZE) else 'other'
+ except OSError:
+ return 'other'
+
+
+class ShareServer(http.server.ThreadingHTTPServer):
+ def __init__(self, addr, handler, *, upload_dir: Path, html: bytes, ssl_ctx: ssl.SSLContext | None = None):
+ super().__init__(addr, handler)
+ self.upload_dir = upload_dir
+ self.html = html
+ self.ssl_ctx = ssl_ctx
+
+ def get_request(self):
+ client, addr = self.socket.accept()
+ if self.ssl_ctx:
+ client.settimeout(SSL_HANDSHAKE_TIMEOUT)
+ try:
+ client = self.ssl_ctx.wrap_socket(client, server_side=True)
+ except (ssl.SSLError, OSError):
+ client.close()
+ raise
+ client.settimeout(None)
+ return client, addr
+
+
+class Handler(http.server.BaseHTTPRequestHandler):
+ server: ShareServer
+
+ def finish(self):
+ try:
+ super().finish()
+ except (ssl.SSLError, BrokenPipeError, ConnectionResetError, OSError):
+ pass
+
+ def log_message(self, fmt, *args):
+ print(f" [{datetime.now():%H:%M:%S}] {args[0]}")
+
+ def _json(self, data):
+ self._respond(json.dumps(data).encode(), "application/json")
+
+ 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)
+
+ # ── Routes ──
+
+ ROUTES = {
+ "GET": [
+ ("/", "_route_index"),
+ ("/index.html","_route_index"),
+ ("/files", "_route_files"),
+ ("/dl/", "_route_download"),
+ ],
+ "POST": [("/upload", "_route_upload")],
+ "DELETE": [("/del/", "_route_delete")],
+ }
+
+ 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)(prefix)
+ self.send_error(404)
+
+ do_GET = do_POST = do_DELETE = _dispatch
+
+ def _route_index(self, _prefix):
+ self._respond(self.server.html, "text/html; charset=utf-8")
+
+ def _route_files(self, _prefix):
+ entries = []
+ for p in self.server.upload_dir.iterdir():
+ if p.is_file():
+ entries.append((p, p.stat()))
+ 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: str) -> Path:
+ name = urllib.parse.unquote(self.path[len(prefix):])
+ return self.server.upload_dir / Path(name).name
+
+ 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="{fname}"')
+ self.send_header("Content-Length", str(st.st_size))
+ self.send_header("Connection", "close")
+ self.end_headers()
+ with safe.open("rb") as f:
+ while chunk := f.read(CHUNK_SIZE):
+ self.wfile.write(chunk)
+
+ 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, prefix):
+ safe = self._safe_path(prefix)
+ if safe.is_file():
+ safe.unlink()
+ self._json({"ok": True})
+
+ # ── Helpers ──
+
+ def _save_parts(self, content_type: str, body: bytes) -> int:
+ header = f"Content-Type: {content_type}\r\n\r\n".encode()
+ msg = email.parser.BytesParser().parsebytes(header + body)
+ count = 0
+ for part in msg.walk():
+ fname = part.get_filename()
+ if not fname:
+ continue
+ fname = Path(fname).name or f"paste_{datetime.now():%H%M%S}"
+ data = part.get_payload(decode=True)
+ if not data:
+ continue
+ dest = self._unique_path(fname)
+ dest.write_bytes(data)
+ count += 1
+ print(f" + {dest.name} ({len(data)} bytes)")
+ return count
+
+ def _unique_path(self, filename: str) -> Path:
+ dest = self.server.upload_dir / filename
+ if not dest.exists():
+ return dest
+ stem, suffix = dest.stem, dest.suffix
+ i = 1
+ while dest.exists():
+ dest = self.server.upload_dir / f"{stem}_{i}{suffix}"
+ i += 1
+ return dest
diff --git a/share.py b/share.py
index 6e6cb97..a784d01 100755
--- a/share.py
+++ b/share.py
@@ -1,301 +1,15 @@
#!/usr/bin/env python3
"""Share — local file sharing server (HTTPS)."""
-import argparse
-import configparser
-import email.parser
-import http.server
import json
-import mimetypes
-import socket
import ssl
-import stat
-import subprocess
import threading
-import time
-import urllib.parse
-from datetime import datetime
from pathlib import Path
-# ── Cleanup ──
-
-
-def cleanup_loop(upload_dir: Path, max_age: int):
- if max_age <= 0:
- return
- while True:
- time.sleep(CLEANUP_INTERVAL)
- now = time.time()
- try:
- 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 OSError:
- pass
-
-
-# ── SSL ──
-
-
-def get_local_ip():
- try:
- with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
- s.connect(("8.8.8.8", 80))
- return s.getsockname()[0]
- except OSError:
- return "127.0.0.1"
-
-
-def ensure_cert(cert_dir: Path, extra_ips: list[str] | None = None) -> tuple[Path, Path]:
- """Generate self-signed cert if missing. Returns (certfile, keyfile)."""
- cert_dir.mkdir(parents=True, exist_ok=True)
- cert = cert_dir / "cert.pem"
- 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:
- ips.update(extra_ips)
- san = ",".join([f"IP:{ip}" for ip in sorted(ips)] + ["DNS:localhost"])
-
- subprocess.run([
- "openssl", "req", "-x509", "-newkey", f"rsa:{CERT_KEY_BITS}",
- "-keyout", str(key), "-out", str(cert),
- "-days", str(CERT_DAYS), "-nodes",
- "-subj", "/CN=Share",
- "-addext", f"subjectAltName={san}",
- ], check=True, capture_output=True)
- print(f" Generated self-signed cert in {cert_dir}")
- return cert, key
-
-
-BASE_DIR = Path(__file__).resolve().parent
-
-# ── Config ──
-
-_cp = configparser.ConfigParser()
-_cp.read_dict({
- "server": {
- "port": "3001", "dir": str(Path.home() / "Downloads" / "shared"),
- "ttl": "7", "san": "", "lang": "ru", "refresh": "5",
- "per_page": "10,25,50,100",
- },
- "internal": {
- "secs_per_day": "86400", "cleanup_interval": "3600",
- "ssl_handshake_timeout": "5", "sniff_size": "8192",
- "chunk_size": "65536", "cert_days": "3650", "cert_key_bits": "2048",
- "toast_ms": "2500", "progress_hide_ms": "1200",
- },
-})
-_cp.read(BASE_DIR / "share.config")
-
-SECS_PER_DAY = _cp.getint("internal", "secs_per_day")
-CLEANUP_INTERVAL = _cp.getint("internal", "cleanup_interval")
-SSL_HANDSHAKE_TIMEOUT = _cp.getint("internal", "ssl_handshake_timeout")
-SNIFF_SIZE = _cp.getint("internal", "sniff_size")
-CHUNK_SIZE = _cp.getint("internal", "chunk_size")
-CERT_DAYS = _cp.getint("internal", "cert_days")
-CERT_KEY_BITS = _cp.getint("internal", "cert_key_bits")
-
-def _file_type(path: Path) -> str:
- ct = mimetypes.guess_type(path.name)[0] or ''
- if ct.startswith('image/'):
- return 'image'
- if ct.startswith('text/'):
- return 'text'
- try:
- with path.open('rb') as f:
- return 'text' if b'\x00' not in f.read(SNIFF_SIZE) else 'other'
- except OSError:
- return 'other'
-
-
-# ── HTTP Handler ──
-
-
-class ShareServer(http.server.ThreadingHTTPServer):
- def __init__(self, addr, handler, *, upload_dir: Path, html: bytes, ssl_ctx: ssl.SSLContext | None = None):
- super().__init__(addr, handler)
- self.upload_dir = upload_dir
- self.html = html
- self.ssl_ctx = ssl_ctx
-
- def get_request(self):
- client, addr = self.socket.accept()
- if self.ssl_ctx:
- client.settimeout(SSL_HANDSHAKE_TIMEOUT)
- try:
- client = self.ssl_ctx.wrap_socket(client, server_side=True)
- except (ssl.SSLError, OSError):
- client.close()
- raise
- client.settimeout(None)
- return client, addr
-
-
-class Handler(http.server.BaseHTTPRequestHandler):
- server: ShareServer
-
- def finish(self):
- try:
- super().finish()
- except (ssl.SSLError, BrokenPipeError, ConnectionResetError, OSError):
- pass
-
- def log_message(self, fmt, *args):
- print(f" [{datetime.now():%H:%M:%S}] {args[0]}")
-
- def _json(self, data):
- self._respond(json.dumps(data).encode(), "application/json")
-
- 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)
-
- # ── Routes ──
-
- ROUTES = {
- "GET": [
- ("/", "_route_index"),
- ("/index.html","_route_index"),
- ("/files", "_route_files"),
- ("/dl/", "_route_download"),
- ],
- "POST": [("/upload", "_route_upload")],
- "DELETE": [("/del/", "_route_delete")],
- }
-
- 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)(prefix)
- self.send_error(404)
-
- do_GET = do_POST = do_DELETE = _dispatch
-
- def _route_index(self, _prefix):
- self._respond(self.server.html, "text/html; charset=utf-8")
-
- def _route_files(self, _prefix):
- entries = []
- for p in self.server.upload_dir.iterdir():
- if p.is_file():
- entries.append((p, p.stat()))
- 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: str) -> Path:
- name = urllib.parse.unquote(self.path[len(prefix):])
- return self.server.upload_dir / Path(name).name
-
- 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="{fname}"')
- self.send_header("Content-Length", str(st.st_size))
- self.send_header("Connection", "close")
- self.end_headers()
- with safe.open("rb") as f:
- while chunk := f.read(CHUNK_SIZE):
- self.wfile.write(chunk)
-
- 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, prefix):
- safe = self._safe_path(prefix)
- if safe.is_file():
- safe.unlink()
- self._json({"ok": True})
-
- # ── Helpers ──
-
- def _save_parts(self, content_type: str, body: bytes) -> int:
- header = f"Content-Type: {content_type}\r\n\r\n".encode()
- msg = email.parser.BytesParser().parsebytes(header + body)
- count = 0
- for part in msg.walk():
- fname = part.get_filename()
- if not fname:
- continue
- fname = Path(fname).name or f"paste_{datetime.now():%H%M%S}"
- data = part.get_payload(decode=True)
- if not data:
- continue
- dest = self._unique_path(fname)
- dest.write_bytes(data)
- count += 1
- print(f" + {dest.name} ({len(data)} bytes)")
- return count
-
- def _unique_path(self, filename: str) -> Path:
- dest = self.server.upload_dir / filename
- if not dest.exists():
- return dest
- stem, suffix = dest.stem, dest.suffix
- i = 1
- while dest.exists():
- dest = self.server.upload_dir / f"{stem}_{i}{suffix}"
- i += 1
- return dest
-
-
-# ── Main ──
-
-
-def load_config() -> dict:
- """Load config: share.config defaults ← CLI args."""
- conf = dict(_cp["server"])
-
- p = argparse.ArgumentParser(description="Share — local file sharing server")
- p.add_argument("-p", "--port", type=int)
- p.add_argument("-d", "--dir")
- p.add_argument("--ttl", type=int)
- p.add_argument("--san", nargs="*")
- p.add_argument("--lang")
- p.add_argument("--refresh", type=int)
- args = p.parse_args()
-
- for key, val in vars(args).items():
- if val is not None:
- conf[key] = " ".join(val) if isinstance(val, list) else str(val)
-
- return conf
+from cert import ensure_cert, get_local_ip
+from cleanup import cleanup_loop
+from config import BASE_DIR, SECS_PER_DAY, client_cfg, load_config
+from server import Handler, ShareServer
def main():
@@ -311,14 +25,7 @@ def main():
threading.Thread(target=cleanup_loop, args=(upload_dir, max_age), daemon=True).start()
ip = get_local_ip()
- cfg_json = json.dumps({
- "lang": conf["lang"],
- "refresh": int(conf["refresh"]),
- "perPage": [int(x) for x in conf["per_page"].split(",")],
- "toastMs": _cp.getint("internal", "toast_ms"),
- "progressHideMs": _cp.getint("internal", "progress_hide_ms"),
- })
- cfg_tag = f"".encode()
+ cfg_tag = f"".encode()
html = (BASE_DIR / "static" / "index.html").read_bytes()
html = html.replace(b"", cfg_tag)