from __future__ import annotations import json import mimetypes from datetime import datetime, timezone from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse from services.license_manager import ( APP_VERSION, build_license_status, create_inquiry_message, create_notice_post, delete_admin_user, get_latest_user_message_bundle, get_machine_fingerprint, init_license_db, list_inquiry_messages, list_notice_posts, load_admin_users, load_app_config, perform_email_login, perform_license_check, save_app_config, update_license_status, upsert_admin_message, upsert_admin_user, ) ROOT_DIR = Path(__file__).resolve().parent HOST = "127.0.0.1" PORT = 8100 SERVER_URL = f"http://{HOST}:{PORT}" ALLOWED_ORIGINS = { "http://127.0.0.1:8000", "http://localhost:8000", "http://127.0.0.1:8100", "http://localhost:8100", "https://lewroey.com", "https://www.lewroey.com", "https://api.lewroey.com", "https://admin.lewroey.com", } class LicenseServerHandler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT_DIR), **kwargs) def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/admin/users": self._handle_admin_user_list() return if parsed.path == "/api/license/status": self._handle_license_status_get() return if parsed.path == "/api/user/messages": self._handle_user_messages_get(parsed) return if parsed.path == "/api/admin/messages": self._handle_admin_messages_get(parsed) return if parsed.path == "/api/admin/notices": self._handle_admin_notices_get(parsed) return if parsed.path == "/api/admin/inquiries": self._handle_admin_inquiries_get(parsed) return super().do_GET() def do_OPTIONS(self) -> None: self.send_response(HTTPStatus.NO_CONTENT) self.end_headers() def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/account/email-login": self._handle_email_account_login() return if parsed.path == "/api/license/check": self._handle_license_check() return if parsed.path == "/api/admin/users/upsert": self._handle_admin_user_upsert() return if parsed.path == "/api/admin/users/delete": self._handle_admin_user_delete() return if parsed.path == "/api/admin/messages/upsert": self._handle_admin_message_upsert() return if parsed.path == "/api/admin/notices": self._handle_admin_notice_create() return if parsed.path == "/api/user/inquiries": self._handle_user_inquiry_create() return if parsed.path == "/api/admin/inquiries/reply": self._handle_admin_inquiry_reply_create() return self.send_error(HTTPStatus.NOT_FOUND, "Not Found") def _handle_admin_user_list(self) -> None: users = load_admin_users() self._send_json( HTTPStatus.OK, { "success": True, "users": users, "count": len(users), }, ) def _handle_email_account_login(self) -> None: payload = self._read_json_body() if payload is None: return try: config, saved_status = perform_email_login(str(payload.get("email") or "")) except ValueError as exc: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": str(exc)}) return except RuntimeError as exc: self._send_json(HTTPStatus.CONFLICT, {"success": False, "message": str(exc)}) return except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"계정 연결 저장 실패: {exc}"}) return self._send_json( HTTPStatus.OK, { "success": True, "message": "이메일 계정과 현재 PC를 연결했습니다.", "settings": { "accountEmail": str(config.get("accountEmail") or ""), "deviceBound": bool(config.get("boundFingerprint")), "machineId": get_machine_fingerprint(), "appVersion": APP_VERSION, }, "licenseStatus": saved_status, }, ) def _handle_license_status_get(self) -> None: config = load_app_config() self._send_json(HTTPStatus.OK, {"success": True, "licenseStatus": build_license_status(config)}) def _handle_user_messages_get(self, parsed) -> None: params = parse_qs(parsed.query or "") email = str((params.get("email") or [""])[0] or "").strip().lower() messages = get_latest_user_message_bundle(email) self._send_json( HTTPStatus.OK, { "success": True, "email": email, "notice": messages.get("notice"), "reply": messages.get("reply"), }, ) def _handle_admin_messages_get(self, parsed) -> None: params = parse_qs(parsed.query or "") email = str((params.get("email") or [""])[0] or "").strip().lower() messages = get_latest_user_message_bundle(email) self._send_json( HTTPStatus.OK, { "success": True, "targetEmail": email, "notice": messages.get("notice"), "reply": messages.get("reply"), }, ) def _handle_admin_notices_get(self, parsed) -> None: params = parse_qs(parsed.query or "") limit_raw = str((params.get("limit") or ["20"])[0] or "20").strip() try: limit = int(limit_raw) except ValueError: limit = 20 notices = list_notice_posts(limit=limit) self._send_json( HTTPStatus.OK, { "success": True, "notices": notices, "count": len(notices), }, ) def _handle_admin_inquiries_get(self, parsed) -> None: params = parse_qs(parsed.query or "") email = str((params.get("email") or [""])[0] or "").strip().lower() limit_raw = str((params.get("limit") or ["100"])[0] or "100").strip() try: limit = int(limit_raw) except ValueError: limit = 100 inquiries = list_inquiry_messages(email, limit=limit) self._send_json( HTTPStatus.OK, { "success": True, "email": email, "inquiries": inquiries, "count": len(inquiries), }, ) def _handle_license_check(self) -> None: config = load_app_config() status = perform_license_check(config) try: saved_status = update_license_status(config, status) except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"라이선스 상태 저장 실패: {exc}"}) return response_status = HTTPStatus.OK if saved_status.get("status") in {"active", "pending", "unconfigured"} else HTTPStatus.BAD_REQUEST self._send_json( response_status, { "success": response_status == HTTPStatus.OK, "message": str(saved_status.get("message") or ""), "licenseStatus": saved_status, }, ) def _handle_admin_user_upsert(self) -> None: payload = self._read_json_body() if payload is None: return email = str(payload.get("email") or "").strip().lower() if not email or "@" not in email: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": "올바른 이메일 주소가 필요합니다."}) return expires_at = str(payload.get("expiresAt") or "").strip() if expires_at: try: datetime.fromisoformat(expires_at.replace("Z", "+00:00")) except ValueError: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": "expiresAt 형식이 올바르지 않습니다."}) return try: user = upsert_admin_user( email, { "status": "active", "planName": str(payload.get("planName") or "월구독형").strip() or "월구독형", "expiresAt": expires_at or None, "notes": str(payload.get("notes") or "").strip(), "deviceFingerprint": "" if payload.get("clearDevice") else None, "deviceBoundAt": "" if payload.get("clearDevice") else None, }, ) except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"운영자 사용자 저장 실패: {exc}"}) return config = load_app_config() if str(config.get("accountEmail") or "").strip().lower() == email: config["accountStatus"] = str(user.get("status") or "active") config["expiresAt"] = str(user.get("expiresAt") or "") config["planName"] = str(user.get("planName") or "월구독형") if payload.get("clearDevice"): config["boundFingerprint"] = "" saved_status = config.get("licenseStatus") if isinstance(config.get("licenseStatus"), dict) else {} if saved_status: saved_status["expiresAt"] = str(config.get("expiresAt") or "") saved_status["licensee"] = str(config.get("accountEmail") or "") saved_status["planName"] = str(config.get("planName") or "월구독형") saved_status["accountStatus"] = str(config.get("accountStatus") or "active") saved_status["checkedAt"] = datetime.now(timezone.utc).isoformat() saved_status["message"] = "운영자 설정이 반영되었습니다." if payload.get("clearDevice"): saved_status["deviceBound"] = False config["licenseStatus"] = saved_status save_app_config(config) self._send_json(HTTPStatus.OK, {"success": True, "message": "운영자 사용자 정보를 저장했습니다.", "user": user}) def _handle_admin_user_delete(self) -> None: payload = self._read_json_body() if payload is None: return email = str(payload.get("email") or "").strip().lower() if not email or "@" not in email: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": "올바른 이메일 주소가 필요합니다."}) return try: deleted = delete_admin_user(email) except (OSError, ValueError) as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"운영자 사용자 삭제 실패: {exc}"}) return if not deleted: self._send_json(HTTPStatus.NOT_FOUND, {"success": False, "message": "삭제할 사용자를 찾지 못했습니다."}) return config = load_app_config() if str(config.get("accountEmail") or "").strip().lower() == email: config["accountEmail"] = "" config["accountStatus"] = "pending" config["expiresAt"] = "" config["planName"] = "" config["boundFingerprint"] = "" config["boundAt"] = "" config["authToken"] = "" config["licenseStatus"] = { "status": "unconfigured", "message": "계정이 삭제되어 이메일 연결이 필요합니다.", "checkedAt": datetime.now(timezone.utc).isoformat(), "expiresAt": "", "licensee": "", "planName": "", "serverReachable": True, "deviceBound": False, "accountStatus": "pending", } save_app_config(config) self._send_json(HTTPStatus.OK, {"success": True, "message": "사용자 계정을 삭제했습니다.", "email": email}) def _handle_admin_message_upsert(self) -> None: payload = self._read_json_body() if payload is None: return message_type = str(payload.get("messageType") or "").strip().lower() title = str(payload.get("title") or "").strip() body = str(payload.get("body") or "").strip() target_email = str(payload.get("targetEmail") or "").strip().lower() is_active = bool(payload.get("isActive", True)) try: saved = upsert_admin_message( message_type, title, body, target_email=target_email, is_active=is_active, ) except ValueError as exc: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": str(exc)}) return except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"메시지 저장 실패: {exc}"}) return self._send_json( HTTPStatus.OK, { "success": True, "message": "운영자 메시지를 저장했습니다.", "saved": saved, }, ) def _handle_admin_notice_create(self) -> None: payload = self._read_json_body() if payload is None: return title = str(payload.get("title") or "").strip() body = str(payload.get("body") or "").strip() try: saved = create_notice_post(title, body) except ValueError as exc: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": str(exc)}) return except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"공지 저장 실패: {exc}"}) return self._send_json( HTTPStatus.OK, { "success": True, "message": "공지사항을 저장했습니다.", "notice": saved, }, ) def _handle_user_inquiry_create(self) -> None: payload = self._read_json_body() if payload is None: return email = str(payload.get("email") or "").strip().lower() title = str(payload.get("title") or "").strip() body = str(payload.get("body") or "").strip() try: saved = create_inquiry_message(email, "user", title, body) except ValueError as exc: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": str(exc)}) return except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"문의 저장 실패: {exc}"}) return self._send_json( HTTPStatus.OK, { "success": True, "message": "운영자 문의를 전달했습니다.", "inquiry": saved, }, ) def _handle_admin_inquiry_reply_create(self) -> None: payload = self._read_json_body() if payload is None: return email = str(payload.get("email") or "").strip().lower() title = str(payload.get("title") or "").strip() body = str(payload.get("body") or "").strip() try: saved = create_inquiry_message(email, "admin", title, body) except ValueError as exc: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": str(exc)}) return except OSError as exc: self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"success": False, "message": f"운영자 답변 저장 실패: {exc}"}) return self._send_json( HTTPStatus.OK, { "success": True, "message": "운영자 답변을 저장했습니다.", "reply": saved, }, ) def _read_json_body(self) -> dict | None: content_length = int(self.headers.get("Content-Length", "0")) raw_body = self.rfile.read(content_length) try: return json.loads(raw_body.decode("utf-8")) except json.JSONDecodeError: self._send_json(HTTPStatus.BAD_REQUEST, {"success": False, "message": "잘못된 JSON 요청입니다."}) return None def _send_json(self, status: HTTPStatus, payload: dict) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def end_headers(self) -> None: origin = self.headers.get("Origin", "") self.send_header("Cache-Control", "no-store") if origin in ALLOWED_ORIGINS: self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Vary", "Origin") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") super().end_headers() def guess_type(self, path: str) -> str: guessed = mimetypes.guess_type(path)[0] return guessed or "application/octet-stream" def main() -> None: init_license_db() server = ThreadingHTTPServer((HOST, PORT), LicenseServerHandler) print(f"License server running at {SERVER_URL}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()