brandon
5b76457ad8
- Added main script (main.py) for syncing IPs between NSX managers. - Implemented logging setup and YAML configuration loading. - Created NSXClient for interacting with NSX Policy API. - Developed SyncEngine for comparing and syncing IPs. - Added interactive group selection feature. - Implemented data export functionality for JSON and CSV formats. - Created requirements.txt for dependencies. - Added __init__.py for package versioning. - Included error handling and validation for configuration.
273 lines
8.5 KiB
Python
273 lines
8.5 KiB
Python
"""
|
|
Export source/target group IP data to JSON or CSV for auditing and testing.
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional
|
|
|
|
from .nsx_client import NSXClient
|
|
from .sync_engine import GroupDiff
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _timestamp() -> str:
|
|
return datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
|
|
def _ensure_dir(path: str) -> None:
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Data collection helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def collect_group_data(
|
|
client: NSXClient,
|
|
group_names: List[str],
|
|
domain: str = "default",
|
|
include_effective_ips: bool = True,
|
|
) -> List[Dict]:
|
|
"""
|
|
Build a list of dicts, one per group, with metadata and IP lists.
|
|
Effective IP fetching is optional (slow for large environments).
|
|
"""
|
|
records = []
|
|
for name in group_names:
|
|
record: Dict = {
|
|
"group_name": name,
|
|
"domain": domain,
|
|
"nsx_manager": client.label,
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"static_ips": [],
|
|
"effective_ips": [],
|
|
"group_found": True,
|
|
"error": None,
|
|
}
|
|
try:
|
|
group = client.get_group(name, domain)
|
|
record["group_id"] = group.get("id", "")
|
|
record["display_name"] = group.get("display_name", "")
|
|
record["description"] = group.get("description", "")
|
|
|
|
for expr in group.get("expression", []):
|
|
if expr.get("resource_type") == "IPAddressExpression":
|
|
record["static_ips"].extend(expr.get("ip_addresses", []))
|
|
record["static_ips"] = sorted(set(record["static_ips"]))
|
|
|
|
if include_effective_ips:
|
|
record["effective_ips"] = client.get_group_effective_ips(name, domain)
|
|
|
|
except Exception as exc:
|
|
msg = str(exc)
|
|
if "not found" in msg.lower():
|
|
record["group_found"] = False
|
|
record["error"] = msg
|
|
|
|
records.append(record)
|
|
return records
|
|
|
|
|
|
def collect_diff_data(diffs: List[GroupDiff]) -> List[Dict]:
|
|
"""Serialize a list of GroupDiff objects to plain dicts for export."""
|
|
return [
|
|
{
|
|
"group_name": d.group_name,
|
|
"source_ip_count": len(d.source_ips),
|
|
"target_static_ip_count": len(d.target_static_ips),
|
|
"ips_to_add_count": len(d.ips_to_add),
|
|
"ips_to_remove_count": len(d.ips_to_remove),
|
|
"source_found": d.source_found,
|
|
"target_found": d.target_found,
|
|
"source_ips": d.source_ips,
|
|
"target_static_ips": d.target_static_ips,
|
|
"ips_to_add": d.ips_to_add,
|
|
"ips_to_remove": d.ips_to_remove,
|
|
"error": d.error,
|
|
}
|
|
for d in diffs
|
|
]
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# JSON export
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def export_json(
|
|
data: List[Dict],
|
|
output_path: Optional[str] = None,
|
|
label: str = "export",
|
|
) -> str:
|
|
if not output_path:
|
|
output_path = os.path.join("exports", f"{label}_{_timestamp()}.json")
|
|
_ensure_dir(output_path)
|
|
with open(output_path, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, indent=2, default=str)
|
|
logger.info("JSON export written to: %s", output_path)
|
|
return output_path
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# CSV export
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _flatten_for_csv(record: Dict) -> List[Dict]:
|
|
"""
|
|
Expand a group record so each IP occupies its own row.
|
|
If there are no IPs we still emit one row.
|
|
"""
|
|
base = {
|
|
"group_name": record.get("group_name", ""),
|
|
"group_id": record.get("group_id", ""),
|
|
"display_name": record.get("display_name", ""),
|
|
"nsx_manager": record.get("nsx_manager", ""),
|
|
"domain": record.get("domain", ""),
|
|
"timestamp": record.get("timestamp", ""),
|
|
"group_found": record.get("group_found", ""),
|
|
"error": record.get("error", ""),
|
|
}
|
|
all_ips = sorted(
|
|
set(record.get("effective_ips", []) + record.get("static_ips", []))
|
|
)
|
|
if not all_ips:
|
|
return [{**base, "ip_address": "", "ip_type": ""}]
|
|
|
|
rows = []
|
|
static_set = set(record.get("static_ips", []))
|
|
effective_set = set(record.get("effective_ips", []))
|
|
for ip in all_ips:
|
|
types = []
|
|
if ip in static_set:
|
|
types.append("static")
|
|
if ip in effective_set:
|
|
types.append("effective")
|
|
rows.append({**base, "ip_address": ip, "ip_type": "/".join(types)})
|
|
return rows
|
|
|
|
|
|
def _flatten_diff_for_csv(record: Dict) -> List[Dict]:
|
|
"""Expand a diff record so each IP occupies its own row."""
|
|
base = {
|
|
"group_name": record.get("group_name", ""),
|
|
"source_found": record.get("source_found", ""),
|
|
"target_found": record.get("target_found", ""),
|
|
"error": record.get("error", ""),
|
|
}
|
|
all_ips = sorted(
|
|
set(
|
|
record.get("source_ips", [])
|
|
+ record.get("target_static_ips", [])
|
|
+ record.get("ips_to_add", [])
|
|
+ record.get("ips_to_remove", [])
|
|
)
|
|
)
|
|
if not all_ips:
|
|
return [{**base, "ip_address": "", "action": "no_change"}]
|
|
|
|
src = set(record.get("source_ips", []))
|
|
tgt = set(record.get("target_static_ips", []))
|
|
add = set(record.get("ips_to_add", []))
|
|
rem = set(record.get("ips_to_remove", []))
|
|
|
|
rows = []
|
|
for ip in all_ips:
|
|
if ip in add:
|
|
action = "add"
|
|
elif ip in rem:
|
|
action = "remove"
|
|
else:
|
|
action = "in_sync"
|
|
rows.append(
|
|
{
|
|
**base,
|
|
"ip_address": ip,
|
|
"in_source": ip in src,
|
|
"in_target": ip in tgt,
|
|
"action": action,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def export_csv(
|
|
data: List[Dict],
|
|
output_path: Optional[str] = None,
|
|
label: str = "export",
|
|
is_diff: bool = False,
|
|
) -> str:
|
|
if not output_path:
|
|
output_path = os.path.join("exports", f"{label}_{_timestamp()}.csv")
|
|
_ensure_dir(output_path)
|
|
|
|
rows: List[Dict] = []
|
|
for record in data:
|
|
if is_diff:
|
|
rows.extend(_flatten_diff_for_csv(record))
|
|
else:
|
|
rows.extend(_flatten_for_csv(record))
|
|
|
|
if not rows:
|
|
logger.warning("No data to write to CSV.")
|
|
return output_path
|
|
|
|
with open(output_path, "w", newline="", encoding="utf-8") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
logger.info("CSV export written to: %s", output_path)
|
|
return output_path
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Convenience: export both source and target snapshots + diff
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def export_full_report(
|
|
source_client: NSXClient,
|
|
target_client: NSXClient,
|
|
group_names: List[str],
|
|
diffs: List[GroupDiff],
|
|
output_dir: str = "exports",
|
|
fmt: str = "both",
|
|
source_domain: str = "default",
|
|
target_domain: str = "default",
|
|
) -> Dict[str, str]:
|
|
"""
|
|
Write source snapshot, target snapshot, and diff report.
|
|
fmt: 'json' | 'csv' | 'both'
|
|
Returns a dict of {report_name: file_path}.
|
|
"""
|
|
ts = _timestamp()
|
|
written: Dict[str, str] = {}
|
|
|
|
src_data = collect_group_data(source_client, group_names, source_domain)
|
|
tgt_data = collect_group_data(target_client, group_names, target_domain)
|
|
diff_data = collect_diff_data(diffs)
|
|
|
|
def _path(name: str, ext: str) -> str:
|
|
return os.path.join(output_dir, f"{name}_{ts}.{ext}")
|
|
|
|
if fmt in ("json", "both"):
|
|
written["source_json"] = export_json(src_data, _path("source", "json"), "source")
|
|
written["target_json"] = export_json(tgt_data, _path("target", "json"), "target")
|
|
written["diff_json"] = export_json(diff_data, _path("diff", "json"), "diff")
|
|
|
|
if fmt in ("csv", "both"):
|
|
written["source_csv"] = export_csv(src_data, _path("source", "csv"), "source")
|
|
written["target_csv"] = export_csv(tgt_data, _path("target", "csv"), "target")
|
|
written["diff_csv"] = export_csv(
|
|
diff_data, _path("diff", "csv"), "diff", is_diff=True
|
|
)
|
|
|
|
return written
|