1028

files/pppoe_to_dhcp.py

@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+"""
+Convert a device config from PPPoE uplink to DHCP on VLAN 4000.
+
+Edits ONLY these paths:
+  .network.zones.wan.mode                         -> "dhcp"
+  .network.zones.wan.alias                        -> []          (ensure present)
+  .network.zones.wan.stp                          -> false       (ensure present)
+  .network.zones.wan.dns                          -> ["41.222.55.1"]
+  .ethernet.ports.eth0.network.vlan_access.enabled-> true
+  .ethernet.ports.eth0.network.vlan_access.id     -> 4000
+
+Everything else is preserved as-is (no key sorting, order preserved).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import shutil
+import sys
+from collections import OrderedDict
+from typing import Any, MutableMapping, Sequence
+
+
+def load_json_preserve_order(path: str) -> Any:
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f, object_pairs_hook=OrderedDict)
+
+
+def detect_indent(raw_text: str) -> int:
+    """
+    Best-effort indent detection. Defaults to 2 if unclear.
+    """
+    # Look for the first line that begins with spaces then a quote (a key).
+    m = re.search(r"\n( +)\"", raw_text)
+    if not m:
+        return 2
+    spaces = len(m.group(1))
+    # Common indents are 2 or 4; accept any positive count.
+    return spaces if spaces > 0 else 2
+
+
+def get_mapping(root: Any, path: Sequence[str]) -> MutableMapping[str, Any]:
+    """
+    Walks down dict-like objects; raises KeyError/TypeError if structure is missing.
+    Returns the mapping at the end of the path.
+    """
+    cur = root
+    for key in path:
+        if not isinstance(cur, MutableMapping):
+            raise TypeError(f"Expected object at {'.'.join(path)}, got {type(cur).__name__}")
+        if key not in cur:
+            raise KeyError(f"Missing key '{key}' at {'.'.join(path)}")
+        cur = cur[key]
+    if not isinstance(cur, MutableMapping):
+        raise TypeError(f"Expected object at {'.'.join(path)}, got {type(cur).__name__}")
+    return cur
+
+
+def ensure_path(root: Any, path: Sequence[str]) -> MutableMapping[str, Any]:
+    """
+    Ensures nested dicts exist; creates missing dicts as OrderedDict.
+    Returns the mapping at the end of the path.
+    """
+    cur = root
+    for key in path:
+        if not isinstance(cur, MutableMapping):
+            raise TypeError(f"Expected object while creating {'.'.join(path)}, got {type(cur).__name__}")
+        if key not in cur or cur[key] is None:
+            cur[key] = OrderedDict()
+        cur = cur[key]
+    if not isinstance(cur, MutableMapping):
+        raise TypeError(f"Expected object at {'.'.join(path)}, got {type(cur).__name__}")
+    return cur
+
+
+def set_value(root: Any, path: Sequence[str], value: Any, create: bool = False) -> tuple[bool, Any, Any]:
+    """
+    Set value at path. If create=False, path must exist. If create=True, missing
+    objects along the way are created.
+    Returns (changed, old_value, new_value).
+    """
+    if len(path) < 1:
+        raise ValueError("Path must have at least one key")
+    parent_path = path[:-1]
+    leaf = path[-1]
+
+    parent = ensure_path(root, parent_path) if create else get_mapping(root, parent_path)
+
+    old = parent.get(leaf, None)
+    if old == value:
+        return (False, old, value)
+    parent[leaf] = value
+    return (True, old, value)
+
+
+def validate_post(root: Any) -> None:
+    """
+    Minimal structural and value validation for the changed fields.
+    Raises exceptions on mismatch.
+    """
+    # Check mode
+    wan = get_mapping(root, ["network", "zones", "wan"])
+    if wan.get("mode") != "dhcp":
+        raise ValueError("Post-check failed: .network.zones.wan.mode != 'dhcp'")
+
+    # Check VLAN access
+    vlan_access = get_mapping(root, ["ethernet", "ports", "eth0", "network", "vlan_access"])
+    if vlan_access.get("enabled") is not True:
+        raise ValueError("Post-check failed: vlan_access.enabled is not true")
+    if vlan_access.get("id") != 4000:
+        raise ValueError("Post-check failed: vlan_access.id != 4000")
+
+    # Check schema fields
+    if wan.get("alias") != []:
+        raise ValueError("Post-check failed: .network.zones.wan.alias != []")
+    if wan.get("stp") is not False:
+        raise ValueError("Post-check failed: .network.zones.wan.stp != false")
+
+    # DNS
+    if wan.get("dns") != ["41.222.55.1"]:
+        raise ValueError("Post-check failed: .network.zones.wan.dns != ['41.222.55.1']")
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Convert config JSON from PPPoE to DHCP on VLAN 4000 (minimal edits).")
+    ap.add_argument("input", help="Input JSON file (e.g., config.json)")
+    ap.add_argument("-o", "--output", help="Output file. If omitted and --in-place not set, prints to stdout.")
+    ap.add_argument("--in-place", action="store_true", help="Modify input file in place (creates .bak backup).")
+    ap.add_argument("--backup-suffix", default=".bak", help="Backup suffix for --in-place (default: .bak)")
+    args = ap.parse_args()
+
+    in_path = args.input
+
+    raw = open(in_path, "r", encoding="utf-8").read()
+    indent = detect_indent(raw)
+
+    data = load_json_preserve_order(in_path)
+
+    changes: list[str] = []
+
+    def apply(path: Sequence[str], value: Any, create: bool = False) -> None:
+        changed, old, new = set_value(data, path, value, create=create)
+        if changed:
+            changes.append(f"{'.' + '.'.join(path)}: {old!r} -> {new!r}")
+
+    # Required edits
+    apply(["network", "zones", "wan", "mode"], "dhcp", create=False)
+
+    # "Only in new" fields must exist exactly like new
+    apply(["network", "zones", "wan", "alias"], [], create=True)
+    apply(["network", "zones", "wan", "stp"], False, create=True)
+
+    # DNS pinned as requested
+    apply(["network", "zones", "wan", "dns"], ["41.222.55.1"], create=True)
+
+    # VLAN 4000 on eth0
+    apply(["ethernet", "ports", "eth0", "network", "vlan_access", "enabled"], True, create=True)
+    apply(["ethernet", "ports", "eth0", "network", "vlan_access", "id"], 4000, create=True)
+
+    # Validate
+    validate_post(data)
+
+    # Serialize
+    out_text = json.dumps(data, indent=indent, ensure_ascii=False) + "\n"
+
+    # Write
+    if args.in_place:
+        backup_path = in_path + args.backup_suffix
+        if not os.path.exists(backup_path):
+            shutil.copy2(in_path, backup_path)
+        else:
+            # Avoid overwriting an existing backup silently
+            raise FileExistsError(f"Backup already exists: {backup_path}")
+        with open(in_path, "w", encoding="utf-8") as f:
+            f.write(out_text)
+    elif args.output:
+        with open(args.output, "w", encoding="utf-8") as f:
+            f.write(out_text)
+    else:
+        sys.stdout.write(out_text)
+
+    # Report to stderr for CLI usage
+    sys.stderr.write("Applied changes:\n")
+    if changes:
+        for line in changes:
+            sys.stderr.write(f"  - {line}\n")
+    else:
+        sys.stderr.write("  (no changes needed; already in desired state)\n")
+
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Exception as e:
+        sys.stderr.write(f"ERROR: {e}\n")
+        raise SystemExit(2)
+