C:\TripServer\control_server\tripserver_control_server.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()