ansible-worker/files/pppoe_to_dhcp.py

#!/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)