share/share.py

#!/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 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(3600)
        now = time.time()
        try:
            for f in upload_dir.iterdir():
                if f.is_file() and now - f.stat().st_mtime > max_age:
                    f.unlink()
                    print(f"  - expired: {f.name}")
        except Exception:
            pass


# ── SSL ──


def get_local_ip():
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception:
        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

    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", "rsa:2048",
        "-keyout", str(key), "-out", str(cert),
        "-days", "3650", "-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

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 open(path, 'rb') as f:
            return 'text' if b'\x00' not in f.read(8192) 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(5)
            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, ct):
        self.send_response(200)
        self.send_header("Content-Type", ct)
        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())

    # ── 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)()
        self.send_error(404)

    do_GET = do_POST = do_DELETE = _dispatch

    def _route_index(self):
        self._respond(self.server.html, "text/html; charset=utf-8")

    def _route_files(self):
        files = []
        for p in sorted(self.server.upload_dir.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True):
            if p.is_file():
                files.append({"name": p.name, "size": p.stat().st_size, "type": _file_type(p)})
        self._json(files)

    def _route_download(self):
        name = urllib.parse.unquote(self.path[4:])
        safe = self.server.upload_dir / Path(name).name
        if not safe.is_file():
            self.send_error(404)
            return
        ct = mimetypes.guess_type(safe.name)[0] or "application/octet-stream"
        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.end_headers()
        with open(safe, "rb") as f:
            while chunk := f.read(65536):
                self.wfile.write(chunk)

    def _route_upload(self):
        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))
        body = self.rfile.read(length)
        count = self._save_parts(ct, body)
        self._json({"ok": True, "count": count})

    def _route_delete(self):
        name = urllib.parse.unquote(self.path[5:])
        safe = self.server.upload_dir / Path(name).name
        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 ──


DEFAULTS = {
    "port": "3001",
    "dir": str(Path.home() / "Downloads" / "shared"),
    "ttl": "7",
    "san": "",
    "lang": "ru",
    "refresh": "5",
    "per_page": "10,25,50,100",
}


def load_config() -> dict:
    """Load config: defaults ← config file ← CLI args."""
    cfg_path = BASE_DIR / "share.conf"
    cp = configparser.ConfigParser()
    cp.read_dict({"share": DEFAULTS})
    cp.read(cfg_path)
    conf = dict(cp["share"])

    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("--ttl", type=int)
    p.add_argument("--san", nargs="*")
    p.add_argument("--lang", type=str)
    p.add_argument("--refresh", type=int)
    args = p.parse_args()

    if args.port is not None:
        conf["port"] = str(args.port)
    if args.dir is not None:
        conf["dir"] = args.dir
    if args.ttl is not None:
        conf["ttl"] = str(args.ttl)
    if args.san is not None:
        conf["san"] = " ".join(args.san)
    if args.lang is not None:
        conf["lang"] = args.lang
    if args.refresh is not None:
        conf["refresh"] = str(args.refresh)

    return conf


def main():
    conf = load_config()

    port = int(conf["port"])
    upload_dir = Path(conf["dir"]).expanduser().resolve()
    max_age = int(conf["ttl"]) * 86400
    san_list = conf["san"].split() if conf["san"] else None

    upload_dir.mkdir(parents=True, exist_ok=True)

    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(",")],
    })
    cfg_tag = f"<script>const CFG={cfg_json}</script>".encode()
    html = (BASE_DIR / "static" / "index.html").read_bytes()
    html = html.replace(b"<!--CFG-->", cfg_tag)

    cert_dir = BASE_DIR / ".certs"
    cert, key = ensure_cert(cert_dir, extra_ips=san_list)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(cert, key)

    server = ShareServer(("0.0.0.0", port), Handler, upload_dir=upload_dir, html=html, ssl_ctx=ctx)

    print(f"\n  Share running")
    print(f"  Local:   https://localhost:{port}")
    print(f"  Network: https://{ip}:{port}")
    print(f"  Files:   {upload_dir}\n")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n  Stopped.")
        server.server_close()


if __name__ == "__main__":
    main()