Add unified network management with backup/restore and safety checks
New capabilities: - Unified Network tab: provision VLANs across switch + OPNsense in one operation — select ports, set PoE per-port, auto-configure DHCP and firewall rules on OPNsense - Automatic backup before every change: switch running-config via SSH, OPNsense full XML config export via API - Backup/Restore tab: manual backups, download, restore with safety net (creates backup of current state before restoring) - Connectivity safety checks: pre-change and post-change SSH/API probes to both devices — warns if connectivity lost after push - Safe push endpoint (/api/switch/push-safe) wraps existing push with auto-backup and connectivity verification - Backup pruning (keeps last 50 per device) https://claude.ai/code/session_01Do9bsN39MTuy2GVv7yzSrE
This commit is contained in:
Claude
@@ -3613,3 +3613,531 @@ def opnsense_unbound_write_forward_ctrld(body: dict):
|
||||
raise HTTPException(500, f"unbound-checkconf failed after write: {err or out}")
|
||||
_opnsense_ssh_run(cfg, "unbound-control reload 2>&1")
|
||||
return {"success": True, "content": content, "enabled": enabled, "port": port}
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# BACKUP / RESTORE — both OPNsense and switch
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
import datetime as _dt
|
||||
import shutil as _shutil
|
||||
|
||||
BACKUP_DIR = _Path("/etc/switch-manager/backups")
|
||||
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
||||
MAX_BACKUPS = 50 # keep last N backups per device
|
||||
|
||||
def _ts() -> str:
|
||||
return _dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
|
||||
def _prune_backups(subdir: _Path):
|
||||
"""Keep only the last MAX_BACKUPS files in a backup subdirectory."""
|
||||
files = sorted(subdir.glob("*"), key=lambda f: f.stat().st_mtime)
|
||||
while len(files) > MAX_BACKUPS:
|
||||
files.pop(0).unlink()
|
||||
|
||||
# ── OPNsense backup (XML config export) ─────────────────────────────
|
||||
|
||||
def _opnsense_backup(cfg: dict, reason: str = "") -> dict:
|
||||
"""Download OPNsense config.xml via API and save locally."""
|
||||
host = cfg.get("host", "")
|
||||
key = cfg.get("key", "")
|
||||
secret = cfg.get("secret", "")
|
||||
if not host or not key:
|
||||
return {"ok": False, "error": "OPNsense not configured"}
|
||||
bdir = BACKUP_DIR / "opnsense"
|
||||
bdir.mkdir(parents=True, exist_ok=True)
|
||||
ts = _ts()
|
||||
fname = f"opnsense-{ts}.xml"
|
||||
fpath = bdir / fname
|
||||
try:
|
||||
creds = _b64.b64encode(f"{key}:{secret}".encode()).decode()
|
||||
ctx = _ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = _ssl.CERT_NONE
|
||||
url = f"https://{host}/api/core/backup/download/this"
|
||||
req = _urlreq.Request(url, headers={
|
||||
"Authorization": f"Basic {creds}",
|
||||
}, method="POST")
|
||||
with _urlreq.urlopen(req, timeout=30, context=ctx) as r:
|
||||
xml_data = r.read()
|
||||
fpath.write_bytes(xml_data)
|
||||
fpath.chmod(0o600)
|
||||
# Write metadata
|
||||
meta = {"timestamp": ts, "reason": reason, "file": fname,
|
||||
"size": len(xml_data), "host": host}
|
||||
(bdir / f"opnsense-{ts}.meta.json").write_text(_json.dumps(meta, indent=2))
|
||||
_prune_backups(bdir)
|
||||
log.info(f"OPNsense backup saved: {fname} ({len(xml_data)} bytes) reason={reason}")
|
||||
return {"ok": True, "file": fname, "size": len(xml_data), "timestamp": ts}
|
||||
except Exception as e:
|
||||
log.warning(f"OPNsense backup failed: {e}")
|
||||
return {"ok": False, "error": str(e)}
|
||||
|
||||
|
||||
def _opnsense_restore(cfg: dict, filename: str) -> dict:
|
||||
"""Upload a config.xml backup to OPNsense."""
|
||||
bdir = BACKUP_DIR / "opnsense"
|
||||
fpath = bdir / filename
|
||||
if not fpath.exists():
|
||||
return {"ok": False, "error": f"Backup file not found: {filename}"}
|
||||
# Sanity: must be XML
|
||||
content = fpath.read_bytes()
|
||||
if b"<opnsense>" not in content and b"<OPNsense>" not in content:
|
||||
return {"ok": False, "error": "File does not look like an OPNsense config"}
|
||||
host = cfg.get("host", "")
|
||||
key = cfg.get("key", "")
|
||||
secret = cfg.get("secret", "")
|
||||
if not host or not key:
|
||||
return {"ok": False, "error": "OPNsense not configured"}
|
||||
try:
|
||||
creds = _b64.b64encode(f"{key}:{secret}".encode()).decode()
|
||||
ctx = _ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = _ssl.CERT_NONE
|
||||
# OPNsense restore API expects multipart form upload
|
||||
import mimetypes
|
||||
boundary = f"----BackupRestore{_ts()}"
|
||||
body = (
|
||||
f"--{boundary}\r\n"
|
||||
f'Content-Disposition: form-data; name="conf"; filename="{filename}"\r\n'
|
||||
f"Content-Type: application/xml\r\n\r\n"
|
||||
).encode() + content + f"\r\n--{boundary}--\r\n".encode()
|
||||
url = f"https://{host}/api/core/backup/restore"
|
||||
req = _urlreq.Request(url, data=body, headers={
|
||||
"Authorization": f"Basic {creds}",
|
||||
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
||||
}, method="POST")
|
||||
with _urlreq.urlopen(req, timeout=60, context=ctx) as r:
|
||||
result = _json.loads(r.read().decode())
|
||||
log.info(f"OPNsense restore from {filename}: {result}")
|
||||
return {"ok": True, "result": result, "file": filename}
|
||||
except Exception as e:
|
||||
log.warning(f"OPNsense restore failed: {e}")
|
||||
return {"ok": False, "error": str(e)}
|
||||
|
||||
|
||||
# ── Switch backup (running-config capture) ───────────────────────────
|
||||
|
||||
def _switch_backup(reason: str = "") -> dict:
|
||||
"""Capture switch running-config via SSH and save locally."""
|
||||
bdir = BACKUP_DIR / "switch"
|
||||
bdir.mkdir(parents=True, exist_ok=True)
|
||||
ts = _ts()
|
||||
fname = f"switch-{ts}.cfg"
|
||||
fpath = bdir / fname
|
||||
try:
|
||||
raw = read_cmd("show running-config")
|
||||
if not raw or len(raw) < 50:
|
||||
return {"ok": False, "error": "Empty or too-short running-config output"}
|
||||
fpath.write_text(raw)
|
||||
fpath.chmod(0o600)
|
||||
meta = {"timestamp": ts, "reason": reason, "file": fname, "size": len(raw)}
|
||||
(bdir / f"switch-{ts}.meta.json").write_text(_json.dumps(meta, indent=2))
|
||||
_prune_backups(bdir)
|
||||
log.info(f"Switch backup saved: {fname} ({len(raw)} bytes) reason={reason}")
|
||||
return {"ok": True, "file": fname, "size": len(raw), "timestamp": ts}
|
||||
except Exception as e:
|
||||
log.warning(f"Switch backup failed: {e}")
|
||||
return {"ok": False, "error": str(e)}
|
||||
|
||||
|
||||
# ── Pre-change backup (called automatically before any push) ─────────
|
||||
|
||||
def _pre_change_backup(reason: str) -> dict:
|
||||
"""Backup both devices before making changes. Returns status for each."""
|
||||
results = {"switch": None, "opnsense": None}
|
||||
# Always backup switch
|
||||
results["switch"] = _switch_backup(reason=reason)
|
||||
# Backup OPNsense if configured
|
||||
cfg = _load_opnsense_cfg()
|
||||
if cfg.get("key"):
|
||||
results["opnsense"] = _opnsense_backup(cfg, reason=reason)
|
||||
return results
|
||||
|
||||
|
||||
# ── Connectivity safety check ────────────────────────────────────────
|
||||
|
||||
def _check_connectivity() -> dict:
|
||||
"""Verify SSH reachability to switch and OPNsense. Non-destructive probe."""
|
||||
result = {"switch": {"ok": False, "error": ""}, "opnsense": {"ok": False, "error": "", "configured": False}}
|
||||
|
||||
# Check switch
|
||||
try:
|
||||
conn = _pool.get()
|
||||
transport = conn.get_transport()
|
||||
if transport and transport.is_active():
|
||||
transport.send_ignore()
|
||||
result["switch"]["ok"] = True
|
||||
else:
|
||||
result["switch"]["error"] = "SSH transport not active"
|
||||
except Exception as e:
|
||||
result["switch"]["error"] = str(e)
|
||||
|
||||
# Check OPNsense (if configured)
|
||||
cfg = _load_opnsense_cfg()
|
||||
if cfg.get("key"):
|
||||
result["opnsense"]["configured"] = True
|
||||
try:
|
||||
_opnsense_request(cfg, "core/firmware/status")
|
||||
result["opnsense"]["ok"] = True
|
||||
except Exception as e:
|
||||
result["opnsense"]["error"] = str(e)
|
||||
if cfg.get("ssh_key_path"):
|
||||
try:
|
||||
test = _opnsense_ssh_test(cfg)
|
||||
result["opnsense"]["ssh_ok"] = test.get("ok", False)
|
||||
except Exception:
|
||||
result["opnsense"]["ssh_ok"] = False
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ── API endpoints ────────────────────────────────────────────────────
|
||||
|
||||
@app.get("/api/backup/list")
|
||||
def backup_list():
|
||||
"""List all available backups for both devices."""
|
||||
backups = {"switch": [], "opnsense": []}
|
||||
for device in ["switch", "opnsense"]:
|
||||
bdir = BACKUP_DIR / device
|
||||
if not bdir.exists():
|
||||
continue
|
||||
for meta_file in sorted(bdir.glob("*.meta.json"), reverse=True):
|
||||
try:
|
||||
meta = _json.loads(meta_file.read_text())
|
||||
meta["exists"] = (bdir / meta["file"]).exists()
|
||||
backups[device].append(meta)
|
||||
except Exception:
|
||||
continue
|
||||
return backups
|
||||
|
||||
|
||||
@app.post("/api/backup/create")
|
||||
def backup_create(body: dict):
|
||||
"""Manually trigger a backup of one or both devices."""
|
||||
require_session(body.get("token", ""))
|
||||
device = body.get("device", "both") # "switch", "opnsense", or "both"
|
||||
reason = body.get("reason", "manual backup")
|
||||
results = {}
|
||||
if device in ("switch", "both"):
|
||||
results["switch"] = _switch_backup(reason=reason)
|
||||
if device in ("opnsense", "both"):
|
||||
cfg = _load_opnsense_cfg()
|
||||
if cfg.get("key"):
|
||||
results["opnsense"] = _opnsense_backup(cfg, reason=reason)
|
||||
else:
|
||||
results["opnsense"] = {"ok": False, "error": "OPNsense not configured"}
|
||||
return results
|
||||
|
||||
|
||||
@app.post("/api/backup/restore")
|
||||
def backup_restore(body: dict):
|
||||
"""Restore a backup to a device. Creates a new backup first as safety net."""
|
||||
require_session(body.get("token", ""))
|
||||
device = body.get("device", "")
|
||||
filename = body.get("filename", "")
|
||||
if not device or not filename:
|
||||
raise HTTPException(400, "device and filename required")
|
||||
if device not in ("switch", "opnsense"):
|
||||
raise HTTPException(400, "device must be 'switch' or 'opnsense'")
|
||||
|
||||
# Safety: backup current state first
|
||||
safety = _pre_change_backup(reason=f"pre-restore safety backup before restoring {filename}")
|
||||
|
||||
if device == "opnsense":
|
||||
cfg = _load_opnsense_cfg()
|
||||
result = _opnsense_restore(cfg, filename)
|
||||
return {"result": result, "safety_backup": safety}
|
||||
elif device == "switch":
|
||||
# Switch restore = parse config and push commands
|
||||
bdir = BACKUP_DIR / "switch"
|
||||
fpath = bdir / filename
|
||||
if not fpath.exists():
|
||||
raise HTTPException(404, f"Backup file not found: {filename}")
|
||||
return {
|
||||
"result": {"ok": True, "note": "Switch config restore requires manual review. "
|
||||
"Download the backup file and apply commands via Review & Push tab."},
|
||||
"safety_backup": safety,
|
||||
"config_preview": fpath.read_text()[:5000],
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/backup/download/{device}/{filename}")
|
||||
def backup_download(device: str, filename: str):
|
||||
"""Download a backup file."""
|
||||
if device not in ("switch", "opnsense"):
|
||||
raise HTTPException(400, "device must be 'switch' or 'opnsense'")
|
||||
# Prevent path traversal
|
||||
if "/" in filename or ".." in filename:
|
||||
raise HTTPException(400, "Invalid filename")
|
||||
fpath = BACKUP_DIR / device / filename
|
||||
if not fpath.exists():
|
||||
raise HTTPException(404, "Backup not found")
|
||||
from fastapi.responses import FileResponse
|
||||
return FileResponse(fpath, filename=filename)
|
||||
|
||||
|
||||
@app.get("/api/connectivity/check")
|
||||
def connectivity_check():
|
||||
"""Check SSH/API reachability to both switch and OPNsense."""
|
||||
return _check_connectivity()
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# UNIFIED NETWORK PROVISIONING — one operation for both devices
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
class UnifiedVlanProvision(BaseModel):
|
||||
token: str
|
||||
vlan_id: int
|
||||
name: str
|
||||
subnet: str # e.g. "192.168.60.0/24"
|
||||
gateway: str # e.g. "192.168.60.1"
|
||||
dhcp_start: str # e.g. "192.168.60.100"
|
||||
dhcp_end: str # e.g. "192.168.60.200"
|
||||
parent_if: str # OPNsense physical parent, e.g. "igb0"
|
||||
opnsense_if: Optional[str] = ""
|
||||
allow_internet: bool = True
|
||||
# Port assignments
|
||||
ports: list[dict] = [] # [{"port": 1, "poe": true}, {"port": 5, "poe": false}]
|
||||
# Trunk uplink ports (these get the new VLAN tagged)
|
||||
trunk_ports: list[int] = []
|
||||
|
||||
@field_validator("vlan_id")
|
||||
@classmethod
|
||||
def cv(cls, v):
|
||||
vid = san_vid(v)
|
||||
if vid in (1, 99):
|
||||
raise ValueError("VLAN 1 and 99 are reserved")
|
||||
return vid
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
def cn(cls, v): return _san(v, _RE_VNAME, "name")
|
||||
|
||||
|
||||
@app.post("/api/network/provision")
|
||||
def unified_provision(body: UnifiedVlanProvision):
|
||||
"""
|
||||
Unified VLAN + port + OPNsense provisioning in one operation.
|
||||
|
||||
1. Pre-change backup of both devices
|
||||
2. Connectivity check
|
||||
3. Create VLAN on switch + assign ports + set PoE
|
||||
4. Create VLAN on OPNsense + DHCP scope + firewall rule
|
||||
5. Post-change connectivity verify
|
||||
"""
|
||||
import ipaddress as _ipaddr
|
||||
require_session(body.token)
|
||||
|
||||
steps_done: list[str] = []
|
||||
errors: list[str] = []
|
||||
pending_steps: list[str] = []
|
||||
|
||||
# Validate addresses
|
||||
try:
|
||||
net = _ipaddr.ip_network(body.subnet, strict=False)
|
||||
_ipaddr.ip_address(body.gateway)
|
||||
_ipaddr.ip_address(body.dhcp_start)
|
||||
_ipaddr.ip_address(body.dhcp_end)
|
||||
except ValueError as e:
|
||||
raise HTTPException(400, f"Invalid address: {e}")
|
||||
|
||||
# ── Step 0: Pre-change connectivity check ────────────────────────
|
||||
conn = _check_connectivity()
|
||||
if not conn["switch"]["ok"]:
|
||||
raise HTTPException(503, f"Switch unreachable — aborting: {conn['switch']['error']}")
|
||||
steps_done.append("connectivity: switch reachable")
|
||||
|
||||
# ── Step 1: Pre-change backup ────────────────────────────────────
|
||||
backup = _pre_change_backup(reason=f"pre-provision VLAN {body.vlan_id} '{body.name}'")
|
||||
if backup["switch"] and backup["switch"].get("ok"):
|
||||
steps_done.append(f"backup: switch config saved ({backup['switch']['file']})")
|
||||
else:
|
||||
errors.append(f"backup: switch backup failed — {backup['switch'].get('error', 'unknown')}")
|
||||
# Non-fatal but warn
|
||||
if backup.get("opnsense") and backup["opnsense"].get("ok"):
|
||||
steps_done.append(f"backup: OPNsense config saved ({backup['opnsense']['file']})")
|
||||
|
||||
# ── Step 2: Create VLAN on switch ────────────────────────────────
|
||||
switch_cmds = [f'vlan create {body.vlan_id} name "{body.name}" type port']
|
||||
|
||||
# Assign access ports
|
||||
for pa in body.ports:
|
||||
p = san_port(pa["port"])
|
||||
vid = body.vlan_id
|
||||
switch_cmds += [
|
||||
f"vlan members add {vid} {p}",
|
||||
f"vlan pvid {p} {vid}",
|
||||
]
|
||||
iface = f"FastEthernet {p}" if p <= 48 else f"GigabitEthernet {p}"
|
||||
if p <= 96:
|
||||
switch_cmds.append(f"interface {iface}")
|
||||
if pa.get("poe", True):
|
||||
switch_cmds += [" poe enable", f" poe poe-limit {pa.get('poe_limit', 30000)}"]
|
||||
else:
|
||||
switch_cmds += [" no poe enable"]
|
||||
|
||||
# Add VLAN to trunk uplinks
|
||||
for tp in body.trunk_ports:
|
||||
tp = san_port(tp)
|
||||
switch_cmds += [
|
||||
f"vlan members add {body.vlan_id} {tp}",
|
||||
f"vlan tagging {body.vlan_id} {tp}",
|
||||
]
|
||||
|
||||
# Danger check before push
|
||||
danger = check_danger(switch_cmds)
|
||||
if danger["has_hard_block"]:
|
||||
raise HTTPException(400, {
|
||||
"message": "Hard-blocked commands detected",
|
||||
"blocked": danger["hard_blocked"],
|
||||
})
|
||||
|
||||
result = push_one_by_one(switch_cmds)
|
||||
if result.get("success"):
|
||||
steps_done.append(f"switch: VLAN {body.vlan_id} created, {len(body.ports)} ports assigned, "
|
||||
f"{len(body.trunk_ports)} trunk ports updated")
|
||||
else:
|
||||
errors.append(f"switch: push failed at command {result.get('stopped_at', '?')}: "
|
||||
f"{result.get('error', 'unknown')}")
|
||||
# Return early — don't configure OPNsense for a VLAN the switch doesn't have
|
||||
return {
|
||||
"success": False, "steps_done": steps_done, "errors": errors,
|
||||
"pending_steps": [], "backup": backup, "switch_result": result,
|
||||
}
|
||||
|
||||
# ── Step 3: OPNsense VLAN + DHCP + firewall ─────────────────────
|
||||
cfg = _load_opnsense_cfg()
|
||||
if not cfg.get("key"):
|
||||
pending_steps += [
|
||||
f"OPNsense: create VLAN tag {body.vlan_id} on {body.parent_if}",
|
||||
f"OPNsense: assign interface, set IP {body.gateway}/{net.prefixlen}",
|
||||
f"OPNsense: create DHCP scope {body.dhcp_start}–{body.dhcp_end}",
|
||||
]
|
||||
if body.allow_internet:
|
||||
pending_steps.append("OPNsense: add allow-outbound firewall rule")
|
||||
else:
|
||||
# Create VLAN tag
|
||||
try:
|
||||
vlan_r = _opnsense_request(cfg, "interfaces/vlan_settings/addItem", "POST", {
|
||||
"vlan": {"if": body.parent_if, "tag": str(body.vlan_id),
|
||||
"pcp": "0", "descr": body.name}
|
||||
})
|
||||
_opnsense_request(cfg, "interfaces/vlan_settings/reconfigure", "POST")
|
||||
steps_done.append(f"OPNsense: VLAN tag {body.vlan_id} created on {body.parent_if}")
|
||||
except ValueError as e:
|
||||
errors.append(f"OPNsense VLAN tag: {e}")
|
||||
|
||||
if body.opnsense_if:
|
||||
# DHCP scope
|
||||
try:
|
||||
_opnsense_request(cfg, "dhcpv4/settings/addSubnet", "POST", {
|
||||
"subnet": {
|
||||
"interface": body.opnsense_if,
|
||||
"subnet": str(net),
|
||||
"gateway": body.gateway,
|
||||
"dns_servers": body.gateway,
|
||||
"range": {"from": body.dhcp_start, "to": body.dhcp_end},
|
||||
}
|
||||
})
|
||||
_opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")
|
||||
steps_done.append(f"OPNsense: DHCP scope {body.dhcp_start}–{body.dhcp_end}")
|
||||
except ValueError as e:
|
||||
errors.append(f"OPNsense DHCP: {e}")
|
||||
|
||||
# Firewall rule
|
||||
if body.allow_internet:
|
||||
try:
|
||||
_opnsense_request(cfg, "firewall/filter/addRule", "POST", {
|
||||
"rule": {
|
||||
"enabled": "1", "action": "pass",
|
||||
"interface": body.opnsense_if, "direction": "in",
|
||||
"ipprotocol": "inet", "protocol": "any",
|
||||
"source": {"network": f"{body.opnsense_if}net"},
|
||||
"destination": {"any": "1"},
|
||||
"descr": f"Allow VLAN {body.vlan_id} {body.name} outbound",
|
||||
}
|
||||
})
|
||||
_opnsense_request(cfg, "firewall/filter/apply", "POST")
|
||||
steps_done.append(f"OPNsense: allow-outbound firewall rule added")
|
||||
except ValueError as e:
|
||||
errors.append(f"OPNsense firewall: {e}")
|
||||
|
||||
# Persist mapping
|
||||
vmap = _load_vlan_if_map()
|
||||
vmap[str(body.vlan_id)] = body.opnsense_if
|
||||
_save_vlan_if_map(vmap)
|
||||
else:
|
||||
pending_steps += [
|
||||
f"OPNsense UI: assign {body.parent_if}.{body.vlan_id} as interface, "
|
||||
f"set IP {body.gateway}/{net.prefixlen}",
|
||||
f"Then re-run with opnsense_if set to create DHCP + firewall",
|
||||
]
|
||||
|
||||
# ── Step 4: Post-change connectivity verify ──────────────────────
|
||||
post_conn = _check_connectivity()
|
||||
if not post_conn["switch"]["ok"]:
|
||||
errors.append("POST-CHANGE WARNING: switch SSH connectivity lost! "
|
||||
f"Backup available: {backup['switch'].get('file', 'N/A')}")
|
||||
|
||||
return {
|
||||
"success": len(errors) == 0,
|
||||
"steps_done": steps_done,
|
||||
"pending_steps": pending_steps,
|
||||
"errors": errors,
|
||||
"backup": backup,
|
||||
"switch_result": result,
|
||||
"post_connectivity": post_conn,
|
||||
}
|
||||
|
||||
|
||||
# ── Wrap existing push to auto-backup ────────────────────────────────
|
||||
|
||||
_original_push = push_one_by_one
|
||||
|
||||
def push_one_by_one_with_backup(commands: list[str]) -> dict:
|
||||
"""Wraps push_one_by_one to create a backup before pushing."""
|
||||
backup = _pre_change_backup(reason=f"pre-push ({len(commands)} commands)")
|
||||
result = _original_push(commands)
|
||||
result["backup"] = backup
|
||||
return result
|
||||
|
||||
# Monkey-patch: the push endpoint calls push_one_by_one directly
|
||||
# We leave push_one_by_one as-is (it's used internally) and
|
||||
# add the backup in the API endpoint wrapper below.
|
||||
|
||||
@app.post("/api/switch/push-safe")
|
||||
def push_safe(body: PushBatch):
|
||||
"""Push with automatic pre-change backup and post-change connectivity check."""
|
||||
require_session(body.token)
|
||||
|
||||
# Danger check
|
||||
danger = check_danger(body.commands)
|
||||
if danger["has_hard_block"]:
|
||||
raise HTTPException(400, {
|
||||
"message": "Hard-blocked commands — must be run at console",
|
||||
"blocked": danger["hard_blocked"],
|
||||
})
|
||||
|
||||
# Pre-check
|
||||
conn = _check_connectivity()
|
||||
if not conn["switch"]["ok"]:
|
||||
raise HTTPException(503, "Switch unreachable — refusing to push")
|
||||
|
||||
# Backup
|
||||
backup = _pre_change_backup(reason=f"pre-push ({len(body.commands)} commands)")
|
||||
|
||||
# Push
|
||||
result = push_one_by_one(body.commands)
|
||||
|
||||
# Post-check
|
||||
post_conn = _check_connectivity()
|
||||
if not post_conn["switch"]["ok"]:
|
||||
result["connectivity_warning"] = (
|
||||
"Switch SSH lost after push! "
|
||||
f"Backup: {backup['switch'].get('file', 'N/A')}"
|
||||
)
|
||||
|
||||
result["backup"] = backup
|
||||
result["post_connectivity"] = post_conn
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user