⬅ חזרה לתיקייה
✏️ עריכת control_before_recovery_20260718_120555.py
C:\TripServer\backups\recovery_v102\control_before_recovery_20260718_120555.py
# -*- coding: utf-8 -*- import os, sys, time, json, shutil, subprocess, socket, threading, traceback, hashlib, re from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse from urllib.request import urlopen BASE_DIR = 'C:\\TripServer' ADMIN_PORT = 8000 CONTROL_PORT = 8010 ADMIN_SCRIPT = os.path.join(BASE_DIR, "admin_server.py") CONTROL_DIR = os.path.join(BASE_DIR, "control_server") PENDING_ADMIN = os.path.join(CONTROL_DIR, "pending_admin_server.py") REQUEST_JSON = os.path.join(CONTROL_DIR, "request.json") STATUS_JSON = os.path.join(CONTROL_DIR, "status.json") FAILSAFE_STATUS_JSON = os.path.join(BASE_DIR, "admin_failsafe_status.json") LOG_PATH = os.path.join(CONTROL_DIR, "tripserver_control_server.log") BACKUP_DIR = os.path.join(BASE_DIR, "backups", "admin_server") PYTHON_EXE = sys.executable if os.path.basename(PYTHON_EXE).lower() == "pythonw.exe": candidate = os.path.join(os.path.dirname(PYTHON_EXE), "python.exe") if os.path.exists(candidate): PYTHON_EXE = candidate os.makedirs(CONTROL_DIR, exist_ok=True) os.makedirs(BACKUP_DIR, exist_ok=True) ACTION_LOCK = threading.RLock() LAST_START_ATTEMPT = 0.0 START_COOLDOWN_SECONDS = 12.0 def now(): return time.strftime("%Y-%m-%d %H:%M:%S") def hidden_flags(extra=0): flags = int(extra or 0) if os.name == "nt": flags |= getattr(subprocess, "CREATE_NO_WINDOW", 0) return flags def log(msg): try: with open(LOG_PATH, "a", encoding="utf-8") as fh: fh.write(now() + " | control-v92 | " + str(msg) + "\n") except Exception: pass def atomic_json(path, data): tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2, default=str) fh.flush() try: os.fsync(fh.fileno()) except Exception: pass os.replace(tmp, path) def read_json(path): try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) return data if isinstance(data, dict) else {} except Exception: return {} def write_status(**kw): previous = read_json(STATUS_JSON) data = { "control_version": "TripServerControl/92", "updated_at": now(), "control_port": CONTROL_PORT, "admin_port": ADMIN_PORT, "python": PYTHON_EXE, } for key in ("update_id", "expected_sha", "expected_version", "expected_build_id", "backup"): if key in previous: data[key] = previous[key] data.update(kw) try: atomic_json(STATUS_JSON, data) except Exception as exc: log("status write failed " + repr(exc)) return data def sha256(path): h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def source_identity(path): result = {"path": os.path.abspath(path), "exists": os.path.isfile(path), "sha256": "", "version": "", "build_id": ""} if not result["exists"]: return result try: result["sha256"] = sha256(path) text = open(path, "r", encoding="utf-8", errors="replace").read() m = re.search(r'^\s*ADMIN_PANEL_VERSION\s*=\s*["\']([^"\']+)["\']', text, re.MULTILINE) b = re.search(r'^\s*ADMIN_BUILD_ID\s*=\s*["\']([^"\']+)["\']', text, re.MULTILINE) if m: result["version"] = m.group(1).strip() if b: result["build_id"] = b.group(1).strip() except Exception as exc: result["error"] = repr(exc) return result def compile_ok(path): try: src = open(path, "r", encoding="utf-8", errors="replace").read() compile(src, path, "exec") return True, "OK" except Exception as exc: return False, repr(exc) def port_open(port): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(0.7) try: return sock.connect_ex(("127.0.0.1", int(port))) == 0 finally: try: sock.close() except Exception: pass def endpoint_port(endpoint): endpoint = str(endpoint or "").strip() if not endpoint: return None try: if endpoint.startswith("["): m = re.search(r"\]:(\d+)$", endpoint) return int(m.group(1)) if m else None return int(endpoint.rsplit(":", 1)[1]) except Exception: return None def pids_on_port(port): pids = [] if os.name != "nt": return pids wanted_port = int(port) try: cp = subprocess.run(["netstat", "-ano", "-p", "tcp"], capture_output=True, text=True, timeout=10, creationflags=hidden_flags()) for line in (cp.stdout or "").splitlines(): parts = line.split() if len(parts) < 5: continue if parts[0].upper() != "TCP" or "listen" not in parts[-2].lower(): continue if endpoint_port(parts[1]) != wanted_port: continue try: pid = int(parts[-1]) except Exception: continue if pid > 0 and pid not in pids: pids.append(pid) except Exception as exc: log("pids_on_port failed " + repr(exc)) return pids def kill_admin(): pids = pids_on_port(ADMIN_PORT) log("kill_admin pids=" + repr(pids)) for pid in pids: if pid == os.getpid(): continue try: cp = subprocess.run(["taskkill", "/PID", str(pid), "/F"], capture_output=True, text=True, timeout=15, creationflags=hidden_flags()) log("taskkill pid=%s rc=%s out=%s" % (pid, cp.returncode, (cp.stdout or cp.stderr or "").strip()[:600])) except Exception as exc: log("taskkill failed pid=%s %r" % (pid, exc)) deadline = time.time() + 12 while time.time() < deadline: if not port_open(ADMIN_PORT): return True time.sleep(0.3) return not port_open(ADMIN_PORT) def start_admin(): global LAST_START_ATTEMPT LAST_START_ATTEMPT = time.time() flags = 0 if os.name == "nt": flags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(subprocess, "DETACHED_PROCESS", 0) out = open(os.path.join(BASE_DIR, "admin_stdout.log"), "a", encoding="utf-8", errors="replace") try: proc = subprocess.Popen([PYTHON_EXE, ADMIN_SCRIPT], cwd=BASE_DIR, stdout=out, stderr=out, stdin=subprocess.DEVNULL, creationflags=hidden_flags(flags)) finally: out.close() log("started admin pid=" + str(proc.pid)) return proc.pid def fetch_admin_runtime(timeout=2.5): try: with urlopen("http://127.0.0.1:%s/ping.json" % ADMIN_PORT, timeout=timeout) as resp: if resp.status != 200: return {} data = json.loads(resp.read().decode("utf-8", errors="replace")) return data if isinstance(data, dict) else {} except Exception: return {} def wait_for_runtime(expected_sha, expected_version="", expected_build_id="", timeout=35): deadline = time.time() + timeout last = {} while time.time() < deadline: payload = fetch_admin_runtime() if payload: last = payload runtime = payload.get("runtime") if isinstance(payload.get("runtime"), dict) else payload loaded_sha = str(runtime.get("loaded_sha256") or "").lower() disk_sha = str(runtime.get("disk_sha256") or "").lower() version = str(runtime.get("version") or payload.get("version") or "") build_id = str(runtime.get("build_id") or payload.get("build_id") or "") path_ok = bool(runtime.get("running_from_expected_file")) loaded_disk_ok = bool(runtime.get("loaded_matches_disk")) ok = loaded_sha == expected_sha.lower() and disk_sha == expected_sha.lower() and path_ok and loaded_disk_ok if expected_version: ok = ok and version == expected_version if expected_build_id: ok = ok and build_id == expected_build_id if ok: return True, payload time.sleep(0.7) return False, last def write_failsafe_status(state, msg, **extra): try: data = {"state": state, "message": msg, "updated_at": now(), "source": "control_server_v92"} data.update(extra) atomic_json(FAILSAFE_STATUS_JSON, data) except Exception as exc: log("failsafe status write failed " + repr(exc)) def no_pending_shutdown_msg(txt): low = (txt or "").lower() return "1116" in low or "no shutdown was in progress" in low def cancel_pending_shutdown(reason="verified_update"): if os.name != "nt": return False, "not windows" try: cp = subprocess.run(["shutdown", "/a"], capture_output=True, text=True, timeout=10, creationflags=hidden_flags()) msg = (cp.stdout or cp.stderr or "").strip() if cp.returncode == 0 or no_pending_shutdown_msg(msg): write_failsafe_status("auto_cancelled" if cp.returncode == 0 else "none", "Fail Safe בוטל רק לאחר אימות Runtime מלא.", windows_output=msg, reason=reason) return True, msg write_failsafe_status("cancel_failed", "אימות Runtime עבר, אך ביטול Fail Safe נכשל.", windows_output=msg, return_code=cp.returncode) return False, msg except Exception as exc: write_failsafe_status("cancel_failed", "שגיאה בביטול Fail Safe: " + repr(exc)) return False, repr(exc) def restore_backup(backup, reason): if not backup or not os.path.isfile(backup): return False, "backup missing" try: shutil.copy2(backup, ADMIN_SCRIPT) actual = sha256(ADMIN_SCRIPT) log("rollback restored reason=%s sha=%s backup=%s" % (reason, actual, backup)) return True, actual except Exception as exc: log("rollback failed " + repr(exc)) return False, repr(exc) def apply_update(reason="manual", request_data=None): req = request_data if isinstance(request_data, dict) else {} update_id = str(req.get("update_id") or "") expected_sha = str(req.get("expected_sha") or "").lower() expected_version = str(req.get("expected_version") or "") expected_build_id = str(req.get("expected_build_id") or "") write_status(last_action="apply_update", state="validating", ok=False, verification_ok=None, reason=reason, update_id=update_id, expected_sha=expected_sha, expected_version=expected_version, expected_build_id=expected_build_id) if not os.path.isfile(PENDING_ADMIN): msg = "pending admin not found" write_status(last_action="apply_update", state="validation_failed", ok=False, verification_ok=False, error=msg) return False, msg ok, info = compile_ok(PENDING_ADMIN) if not ok: write_status(last_action="apply_update", state="validation_failed", ok=False, verification_ok=False, error=info) return False, info candidate = source_identity(PENDING_ADMIN) new_sha = candidate.get("sha256", "").lower() if expected_sha and new_sha != expected_sha: msg = "pending SHA mismatch expected=%s actual=%s" % (expected_sha, new_sha) write_status(last_action="apply_update", state="validation_failed", ok=False, verification_ok=False, error=msg, candidate=candidate) return False, msg expected_sha = expected_sha or new_sha expected_version = expected_version or str(candidate.get("version") or "") expected_build_id = expected_build_id or str(candidate.get("build_id") or "") stamp = time.strftime("%Y%m%d_%H%M%S") backup = os.path.join(BACKUP_DIR, "admin_server_backup_" + stamp + ".py") try: if os.path.isfile(ADMIN_SCRIPT): shutil.copy2(ADMIN_SCRIPT, backup) except Exception as exc: msg = "backup failed: " + repr(exc) write_status(last_action="apply_update", state="backup_failed", ok=False, verification_ok=False, error=msg, backup=backup) return False, msg if not kill_admin(): msg = "admin port did not close" write_status(last_action="apply_update", state="stop_failed", ok=False, verification_ok=False, error=msg, backup=backup) return False, msg try: os.replace(PENDING_ADMIN, ADMIN_SCRIPT) except Exception as exc: msg = "replace failed: " + repr(exc) write_status(last_action="apply_update", state="replace_failed", ok=False, verification_ok=False, error=msg, backup=backup) return False, msg disk_sha = "" try: disk_sha = sha256(ADMIN_SCRIPT).lower() except Exception as exc: log("disk sha failed " + repr(exc)) if disk_sha != expected_sha: rollback_ok, rollback_info = restore_backup(backup, "disk_sha_mismatch") if rollback_ok: try: start_admin() except Exception as exc: rollback_info += " | restart=" + repr(exc) msg = "disk SHA mismatch after replace expected=%s actual=%s" % (expected_sha, disk_sha) write_status(last_action="apply_update", state="disk_verify_failed", ok=False, verification_ok=False, error=msg, expected_sha=expected_sha, disk_sha=disk_sha, backup=backup, rollback_performed=rollback_ok, rollback_result=rollback_info) return False, msg write_status(last_action="apply_update", state="replaced", ok=False, verification_ok=None, expected_sha=expected_sha, expected_version=expected_version, expected_build_id=expected_build_id, disk_sha=disk_sha, backup=backup, candidate=candidate) try: pid = start_admin() except Exception as exc: msg = "start failed: " + repr(exc) write_status(last_action="apply_update", state="installed_unverified", ok=False, verification_ok=False, error=msg, expected_sha=expected_sha, disk_sha=disk_sha, backup=backup) return False, msg verified, runtime_payload = wait_for_runtime(expected_sha, expected_version, expected_build_id, timeout=35) cancel_ok = False cancel_msg = "" if verified: cancel_ok, cancel_msg = cancel_pending_shutdown("runtime_identity_verified") state = "verified" if verified else "installed_unverified" message = "verified update" if verified else "new file is on disk, but loaded runtime identity did not match" write_status(last_action="apply_update", state=state, ok=verified, verification_ok=verified, message=message, update_id=update_id, expected_sha=expected_sha, expected_version=expected_version, expected_build_id=expected_build_id, disk_sha=disk_sha, backup=backup, started_pid=pid, runtime=runtime_payload, failsafe_cancel_ok=cancel_ok, failsafe_cancel_msg=cancel_msg) return verified, message def restart_admin(reason="manual"): expected_sha = sha256(ADMIN_SCRIPT) if os.path.isfile(ADMIN_SCRIPT) else "" if not kill_admin(): write_status(last_action="restart_admin", state="stop_failed", ok=False, verification_ok=False, error="admin port did not close") return False, "admin port did not close" pid = start_admin() verified, runtime_payload = wait_for_runtime(expected_sha, timeout=30) write_status(last_action="restart_admin", state="verified" if verified else "installed_unverified", ok=verified, verification_ok=verified, expected_sha=expected_sha, started_pid=pid, runtime=runtime_payload) return verified, "restart verified" if verified else "restart was not verified" def run_locked(func, *args, **kwargs): with ACTION_LOCK: return func(*args, **kwargs) def process_request_file(): if not os.path.isfile(REQUEST_JSON): return try: req = read_json(REQUEST_JSON) except Exception as exc: log("bad request " + repr(exc)); return try: not_before = float(req.get("not_before", 0) or 0) except Exception: not_before = 0 if time.time() < not_before: return try: os.remove(REQUEST_JSON) except Exception: pass action = req.get("action") if action == "apply_update": run_locked(apply_update, req.get("reason", "request"), req) elif action == "restart_admin": run_locked(restart_admin, req.get("reason", "request")) else: write_status(last_action="unknown_request", state="request_failed", ok=False, verification_ok=False, error="unknown action " + repr(action)) def poll_loop(): while True: try: process_request_file() if not port_open(ADMIN_PORT) and time.time() - LAST_START_ATTEMPT >= START_COOLDOWN_SECONDS: with ACTION_LOCK: if not port_open(ADMIN_PORT) and time.time() - LAST_START_ATTEMPT >= START_COOLDOWN_SECONDS: log("admin port closed outside update; starting canonical admin") start_admin() except Exception as exc: log("poll error " + repr(exc) + "\n" + traceback.format_exc()) time.sleep(2) class Handler(BaseHTTPRequestHandler): server_version = "TripServerControl/92" def send_body(self, code, body, ctype="application/json; charset=utf-8"): if isinstance(body, str): body = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def do_GET(self): path = urlparse(self.path).path if path == "/ping.json": payload = fetch_admin_runtime() self.send_body(200, json.dumps({"ok": True, "control_version": "TripServerControl/92", "admin_port_open": port_open(ADMIN_PORT), "admin_runtime": payload}, ensure_ascii=False)) return if path == "/status.json": data = read_json(STATUS_JSON) data.update({"control_version": "TripServerControl/92", "admin_port_open": port_open(ADMIN_PORT), "pids": pids_on_port(ADMIN_PORT), "active_disk_identity": source_identity(ADMIN_SCRIPT), "admin_runtime": fetch_admin_runtime()}) self.send_body(200, json.dumps(data, ensure_ascii=False, indent=2, default=str)) return if path == "/log": try: txt = open(LOG_PATH, "r", encoding="utf-8", errors="replace").read()[-30000:] except Exception as exc: txt = repr(exc) self.send_body(200, "<pre style='white-space:pre-wrap;direction:ltr'>" + txt.replace("<", "<") + "</pre>", "text/html; charset=utf-8") return html = """<html><head><meta charset='utf-8'><title>TripServer Control v92</title></head><body style='font-family:Arial;background:#0b1220;color:#eef;direction:rtl;padding:30px'><h1>TripServer Control v92</h1><p>עדכון נחשב מוצלח רק לאחר אימות Runtime מלא.</p><p><a style='color:#93c5fd' href='/status.json'>Status JSON</a> · <a style='color:#93c5fd' href='/log'>Log</a></p></body></html>""" self.send_body(200, html, "text/html; charset=utf-8") def do_POST(self): path = urlparse(self.path).path if path == "/restart": ok, msg = run_locked(restart_admin, "http") elif path == "/apply-update": ok, msg = run_locked(apply_update, "http", {}) else: self.send_body(404, "not found", "text/plain; charset=utf-8") return self.send_body(200 if ok else 500, json.dumps({"ok": ok, "message": msg}, ensure_ascii=False)) def log_message(self, fmt, *args): log("http " + (fmt % args)) if __name__ == "__main__": log("starting control server v92 python=" + sys.executable) write_status(last_action="control_start", state="ready", ok=True, verification_ok=None) threading.Thread(target=poll_loop, daemon=True).start() server = ThreadingHTTPServer(("0.0.0.0", CONTROL_PORT), Handler) log("listening on port " + str(CONTROL_PORT)) server.serve_forever()
שמור קובץ