diff --git a/nodeprov.py b/nodeprov.py
index 9b4fb31..4c3f877 100644
--- a/nodeprov.py
+++ b/nodeprov.py
@@ -250,6 +250,10 @@ def _mgmt_connect(address: str, ssh_port: int = 22) -> paramiko.SSHClient:
     return client
 
 
+class RemoteConfigError(Exception):
+    pass
+
+
 def _remote_edit_clients(node: dict, mutate_fn):
     client = _mgmt_connect(node["address"])
     try:
@@ -265,14 +269,27 @@ def _remote_edit_clients(node: dict, mutate_fn):
             if new_clients is not None:
                 ib["settings"]["clients"] = new_clients
                 changed = True
-        if changed:
-            data = json.dumps(cfg, indent=2).encode()
-            with sftp.open("/usr/local/etc/xray/config.json", "wb") as f:
-                f.write(data)
+        if not changed:
             sftp.close()
-            client.exec_command("systemctl restart xray")[1].channel.recv_exit_status()
-        else:
+            return
+        data = json.dumps(cfg, indent=2).encode()
+        tmp_path = "/usr/local/etc/xray/config.json.validate.tmp"
+        with sftp.open(tmp_path, "wb") as f:
+            f.write(data)
+        _, stdout, stderr = client.exec_command(f"/usr/local/bin/xray run -test -format=json -config {tmp_path}", timeout=15)
+        test_exit = stdout.channel.recv_exit_status()
+        test_out = (stdout.read().decode(errors="replace") + stderr.read().decode(errors="replace")).strip()
+        if test_exit != 0:
+            client.exec_command(f"rm -f {tmp_path}")
             sftp.close()
+            raise RemoteConfigError(f"config test failed on {node['address']}: {test_out}")
+        client.exec_command(f"mv {tmp_path} /usr/local/etc/xray/config.json")[1].channel.recv_exit_status()
+        sftp.close()
+        _, stdout, stderr = client.exec_command("systemctl restart xray", timeout=20)
+        restart_exit = stdout.channel.recv_exit_status()
+        if restart_exit != 0:
+            err = stderr.read().decode(errors="replace").strip()
+            raise RemoteConfigError(f"xray restart failed on {node['address']}: {err}")
     finally:
         client.close()
 
diff --git a/xray_manager.py b/xray_manager.py
index 4c1d04e..1c66010 100644
--- a/xray_manager.py
+++ b/xray_manager.py
@@ -26,7 +26,70 @@ def _load():
         return json.load(f)
 
 
+class ConfigValidationError(Exception):
+    pass
+
+
+def _readable_by(path, uid, gid) -> bool:
+    try:
+        st = os.stat(path)
+    except OSError:
+        return False
+    mode = st.st_mode
+    if st.st_uid == uid and mode & 0o400:
+        return True
+    if st.st_gid == gid and mode & 0o040:
+        return True
+    if mode & 0o004:
+        return True
+    return False
+
+
+def check_cert_permissions(cfg, uid=65534, gid=65534) -> list:
+    problems = []
+    for ib in cfg.get("inbounds", []):
+        stream = ib.get("streamSettings") or {}
+        tls = stream.get("tlsSettings")
+        if not tls:
+            continue
+        for cert in tls.get("certificates") or []:
+            for key in ("certificateFile", "keyFile"):
+                path = cert.get(key)
+                if path and not _readable_by(path, uid, gid):
+                    problems.append(f"{ib.get('tag', '?')}: {key}={path} not readable by the xray service user")
+    return problems
+
+
+def validate_config(cfg, xray_bin="/usr/local/bin/xray") -> tuple:
+    tmp = XRAY_CONFIG_PATH + ".validate.tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(cfg, f, indent=2)
+    try:
+        result = subprocess.run(
+            [xray_bin, "run", "-test", "-format=json", "-config", tmp],
+            capture_output=True, text=True, timeout=15,
+        )
+        ok = result.returncode == 0
+        detail = (result.stdout + result.stderr).strip()
+    except Exception as e:
+        return False, str(e)
+    finally:
+        try:
+            os.remove(tmp)
+        except OSError:
+            pass
+    if not ok:
+        return False, detail
+    perm_problems = check_cert_permissions(cfg)
+    if perm_problems:
+        return False, "cert permission problem(s): " + "; ".join(perm_problems)
+    return True, detail
+
+
 def _save(cfg):
+    ok, detail = validate_config(cfg)
+    if not ok:
+        raise ConfigValidationError(detail)
     tmp = XRAY_CONFIG_PATH + ".tmp"
     with open(tmp, "w", encoding="utf-8") as f:
         json.dump(cfg, f, indent=2)