# -*- coding: utf-8 -*- """yjm_file_transfer.py viewer.html <-> host 파일전송 HTTP 보조 모듈. 현재 단계(v2.5): - viewer 브라우저에서 선택한 파일을 host PC의 ./recv_files 폴더로 업로드한다. - 단일 파일은 압축하지 않고 원본 그대로 저장한다. - 저장 중에는 .part로 쓰고, 완료 후 원래 파일명으로 rename한다. - token / host 파일전송 허용 체크가 모두 맞아야 저장한다. - 진행 상황은 host 오버레이와 viewer toast로 알린다. 주의: - 아직 FTP가 아니다. - 아직 host -> viewer 임의 파일 다운로드 선택 UI는 넣지 않았다. - 폴더/여러 파일은 viewer에서 여러 POST로 순차 업로드하면 된다. """ from __future__ import annotations import json import re import time from http.server import SimpleHTTPRequestHandler from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Callable, Optional from urllib.parse import parse_qs, unquote, urlparse import secrets StatusCallback = Callable[[str], None] WebNotifyCallback = Callable[[dict], None] EnabledCallback = Callable[[], bool] def _json_bytes(obj: dict, status: str = "ok") -> bytes: data = dict(obj) data.setdefault("status", status) return json.dumps(data, ensure_ascii=False).encode("utf-8") def _safe_rel_path(raw_name: str) -> Path: """브라우저가 보낸 파일명/상대경로를 recv_files 내부 상대경로로만 정리한다.""" name = unquote(raw_name or "upload.bin").replace("\\", "/") # Windows absolute drive / UNC / 상위 이동 제거 parts: list[str] = [] for part in PurePosixPath(name).parts: part = part.strip().strip(". ") if not part or part in ("/", ".."): continue if re.match(r"^[A-Za-z]:$", part): continue # Windows 파일명 금지 문자 최소 치환 part = re.sub(r'[<>:"|?*\x00-\x1f]', "_", part) if part: parts.append(part[:120]) if not parts: parts = ["upload.bin"] # 너무 깊은 경로 방지 parts = parts[-6:] return Path(*parts) def _unique_target_path(base_dir: Path, rel_path: Path) -> Path: target = (base_dir / rel_path).resolve() base_resolved = base_dir.resolve() if base_resolved not in target.parents and target != base_resolved: target = base_resolved / "upload.bin" target.parent.mkdir(parents=True, exist_ok=True) if not target.exists() and not target.with_suffix(target.suffix + ".part").exists(): return target stem = target.stem or "upload" suffix = target.suffix parent = target.parent for idx in range(1, 10000): cand = parent / f"{stem}_{idx:03d}{suffix}" if not cand.exists() and not cand.with_suffix(cand.suffix + ".part").exists(): return cand return parent / f"{stem}_{int(time.time())}{suffix}" def make_file_transfer_handler( *, web_dir: str | Path, recv_dir: str | Path, token: str = "", is_enabled_callback: Optional[EnabledCallback] = None, status_callback: Optional[StatusCallback] = None, web_notify_callback: Optional[WebNotifyCallback] = None, logger=None, ): """ThreadingHTTPServer에 넣을 Handler class를 생성한다.""" web_dir = Path(web_dir).resolve() recv_dir = Path(recv_dir).resolve() recv_dir.mkdir(parents=True, exist_ok=True) is_enabled_callback = is_enabled_callback or (lambda: False) def host_status(text: str) -> None: if status_callback: try: status_callback(text) except Exception: pass def web_notify(payload: dict) -> None: if web_notify_callback: try: web_notify_callback(payload) except Exception: pass class FileTransferHandler(SimpleHTTPRequestHandler): server_version = "YjmWin2RtcHTTP/0.2" def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(web_dir), **kwargs) def _send_json(self, code: int, obj: dict) -> None: body = _json_bytes(obj, "ok" if 200 <= code < 300 else "error") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, X-File-Name, X-Relative-Path, X-File-Size") self.end_headers() self.wfile.write(body) def do_OPTIONS(self): # noqa: N802 - http.server API self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, X-File-Name, X-Relative-Path, X-File-Size") self.send_header("Access-Control-Max-Age", "600") self.end_headers() def _check_token(self) -> bool: if not token: return True try: parsed = urlparse(self.path) got = parse_qs(parsed.query).get("token", [""])[0] return secrets.compare_digest(str(got), str(token)) except Exception: return False def _reject(self, code: int, message: str) -> None: self._send_json(code, {"message": message}) def do_POST(self): # noqa: N802 - http.server API parsed = urlparse(self.path) if parsed.path != "/upload": return self._reject(404, "unknown POST endpoint") if not self._check_token(): return self._reject(403, "invalid token") try: if not bool(is_enabled_callback()): return self._reject(403, "host file transfer disabled") except Exception: return self._reject(403, "host file transfer disabled") raw_name = self.headers.get("X-File-Name") or self.headers.get("X-Relative-Path") or "upload.bin" display_name = unquote(raw_name).replace("\\", "/").split("/")[-1] or "upload.bin" rel_path = _safe_rel_path(raw_name) target = _unique_target_path(recv_dir, rel_path) part = target.with_suffix(target.suffix + ".part") try: total = int(self.headers.get("Content-Length", "0")) except Exception: total = 0 if total <= 0: return self._reject(400, "empty upload") received = 0 last_notice = 0.0 web_notify({"type": "notify", "kind": "file", "state": "start", "filename": display_name, "percent": 0}) host_status(f"원격: 파일 전송중\n{display_name} 0%") try: with part.open("wb") as f: remaining = total while remaining > 0: chunk = self.rfile.read(min(1024 * 256, remaining)) if not chunk: break f.write(chunk) received += len(chunk) remaining -= len(chunk) now = time.time() if now - last_notice >= 0.45 or received == total: last_notice = now percent = int(received * 100 / max(1, total)) web_notify({ "type": "notify", "kind": "file", "state": "progress", "filename": display_name, "percent": percent, }) host_status(f"원격: 파일 전송중\n{display_name} {percent}%") if received != total: try: part.unlink(missing_ok=True) except Exception: pass web_notify({"type": "notify", "kind": "file", "state": "error", "filename": display_name, "message": "incomplete"}) host_status(f"원격: 파일 수신 실패\n{display_name}") return self._reject(400, f"incomplete upload: {received}/{total}") part.replace(target) web_notify({"type": "notify", "kind": "file", "state": "done", "filename": display_name, "percent": 100}) host_status(f"원격: 파일 수신 완료\n{target.name}") if logger: logger.info("file received: %s (%d bytes)", target, received) return self._send_json(200, {"message": "uploaded", "filename": target.name, "size": received, "path": str(target)}) except Exception as exc: try: part.unlink(missing_ok=True) except Exception: pass web_notify({"type": "notify", "kind": "file", "state": "error", "filename": display_name, "message": str(exc)}) host_status(f"원격: 파일 수신 오류\n{display_name}") if logger: logger.error("upload error: %s", exc) return self._reject(500, str(exc)) def do_GET(self): # noqa: N802 - http.server API parsed = urlparse(self.path) if parsed.path == "/transfer-status": ok_token = self._check_token() enabled = False try: enabled = bool(is_enabled_callback()) except Exception: enabled = False return self._send_json(200, {"token_ok": ok_token, "enabled": enabled}) return super().do_GET() def log_message(self, fmt, *args): if logger: try: logger.debug("HTTP: " + fmt, *args) except Exception: pass return FileTransferHandler