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.
262 lines
9.2 KiB
Python
262 lines
9.2 KiB
Python
"""
|
||
Core sync logic: compare source effective IPs to target static IPs and
|
||
produce a change plan, then optionally apply it.
|
||
"""
|
||
|
||
import logging
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import dataclass, field
|
||
from typing import Dict, List, Optional
|
||
|
||
from .nsx_client import NSXClient
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class GroupDiff:
|
||
group_name: str
|
||
source_ips: List[str] = field(default_factory=list)
|
||
target_static_ips: List[str] = field(default_factory=list)
|
||
ips_to_add: List[str] = field(default_factory=list)
|
||
ips_to_remove: List[str] = field(default_factory=list)
|
||
source_found: bool = True
|
||
target_found: bool = True
|
||
error: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class SyncResult:
|
||
group_name: str
|
||
status: str # updated | no_change | whatif | error | skipped
|
||
ips_added: List[str] = field(default_factory=list)
|
||
ips_removed: List[str] = field(default_factory=list)
|
||
error: Optional[str] = None
|
||
|
||
|
||
class SyncEngine:
|
||
"""
|
||
Orchestrates a full source→target sync for a list of Security Group names.
|
||
|
||
Strategy
|
||
--------
|
||
1. Fetch effective IPs from each group on the source manager.
|
||
2. Fetch the current static IP entries from matching groups on the target.
|
||
3. Diff: source effective IPs are the desired state for static entries on target.
|
||
4. Apply: PATCH target groups to add/remove IPs (skipped in whatif mode).
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
source: NSXClient,
|
||
target: NSXClient,
|
||
source_domain: str = "default",
|
||
target_domain: str = "default",
|
||
max_workers: int = 4,
|
||
):
|
||
self.source = source
|
||
self.target = target
|
||
self.source_domain = source_domain
|
||
self.target_domain = target_domain
|
||
self.max_workers = max_workers
|
||
|
||
# ------------------------------------------------------------------
|
||
# Phase 1: gather data concurrently
|
||
# ------------------------------------------------------------------
|
||
|
||
def _fetch_source_ips(self, group_name: str) -> tuple:
|
||
"""Return (group_name, ips_list, error_or_None)."""
|
||
try:
|
||
ips = self.source.get_group_effective_ips(group_name, self.source_domain)
|
||
return group_name, ips, None
|
||
except Exception as exc:
|
||
return group_name, [], str(exc)
|
||
|
||
def _fetch_target_static_ips(self, group_name: str) -> tuple:
|
||
"""Return (group_name, static_ips_list, found_bool, error_or_None)."""
|
||
try:
|
||
group = self.target.get_group(group_name, self.target_domain)
|
||
expressions = group.get("expression", [])
|
||
ips: List[str] = []
|
||
for expr in expressions:
|
||
if expr.get("resource_type") == "IPAddressExpression":
|
||
ips.extend(expr.get("ip_addresses", []))
|
||
return group_name, sorted(ips), True, None
|
||
except Exception as exc:
|
||
msg = str(exc)
|
||
if "not found" in msg.lower():
|
||
return group_name, [], False, None
|
||
return group_name, [], True, msg
|
||
|
||
def build_diff(self, group_names: List[str]) -> List[GroupDiff]:
|
||
"""
|
||
Concurrently fetch source and target data, then compute per-group diffs.
|
||
"""
|
||
source_map: Dict[str, tuple] = {}
|
||
target_map: Dict[str, tuple] = {}
|
||
|
||
logger.info(
|
||
"Fetching data from source [%s] and target [%s] for %d group(s)...",
|
||
self.source.label, self.target.label, len(group_names),
|
||
)
|
||
|
||
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
|
||
src_futures = {
|
||
pool.submit(self._fetch_source_ips, g): g for g in group_names
|
||
}
|
||
tgt_futures = {
|
||
pool.submit(self._fetch_target_static_ips, g): g for g in group_names
|
||
}
|
||
for fut in as_completed(src_futures):
|
||
name, ips, err = fut.result()
|
||
source_map[name] = (ips, err)
|
||
for fut in as_completed(tgt_futures):
|
||
name, ips, found, err = fut.result()
|
||
target_map[name] = (ips, found, err)
|
||
|
||
diffs: List[GroupDiff] = []
|
||
for name in group_names:
|
||
src_ips, src_err = source_map.get(name, ([], "not fetched"))
|
||
tgt_ips, tgt_found, tgt_err = target_map.get(name, ([], True, "not fetched"))
|
||
|
||
diff = GroupDiff(group_name=name)
|
||
|
||
if src_err:
|
||
diff.error = f"Source error: {src_err}"
|
||
diff.source_found = False
|
||
diffs.append(diff)
|
||
continue
|
||
|
||
if tgt_err:
|
||
diff.error = f"Target error: {tgt_err}"
|
||
diffs.append(diff)
|
||
continue
|
||
|
||
if not tgt_found:
|
||
diff.target_found = False
|
||
diff.error = f"Group '{name}' not found on target – skipping."
|
||
diffs.append(diff)
|
||
continue
|
||
|
||
diff.source_ips = src_ips
|
||
diff.target_static_ips = tgt_ips
|
||
|
||
src_set = set(src_ips)
|
||
tgt_set = set(tgt_ips)
|
||
diff.ips_to_add = sorted(src_set - tgt_set)
|
||
diff.ips_to_remove = sorted(tgt_set - src_set)
|
||
|
||
diffs.append(diff)
|
||
|
||
return diffs
|
||
|
||
# ------------------------------------------------------------------
|
||
# Phase 2: apply changes
|
||
# ------------------------------------------------------------------
|
||
|
||
def _apply_diff(self, diff: GroupDiff, whatif: bool) -> SyncResult:
|
||
if diff.error:
|
||
return SyncResult(
|
||
group_name=diff.group_name,
|
||
status="skipped",
|
||
error=diff.error,
|
||
)
|
||
if not diff.ips_to_add and not diff.ips_to_remove:
|
||
logger.info("Group '%s': already in sync.", diff.group_name)
|
||
return SyncResult(group_name=diff.group_name, status="no_change")
|
||
|
||
try:
|
||
patch_result = self.target.patch_group_ips(
|
||
group_id=diff.group_name,
|
||
ips_to_add=diff.ips_to_add,
|
||
ips_to_remove=diff.ips_to_remove,
|
||
domain=self.target_domain,
|
||
whatif=whatif,
|
||
)
|
||
return SyncResult(
|
||
group_name=diff.group_name,
|
||
status=patch_result["status"],
|
||
ips_added=patch_result.get("ips_added", []),
|
||
ips_removed=patch_result.get("ips_removed", []),
|
||
)
|
||
except Exception as exc:
|
||
logger.error("Group '%s': error applying changes: %s", diff.group_name, exc)
|
||
return SyncResult(
|
||
group_name=diff.group_name,
|
||
status="error",
|
||
error=str(exc),
|
||
)
|
||
|
||
def sync(
|
||
self, group_names: List[str], whatif: bool = False
|
||
) -> tuple[List[GroupDiff], List[SyncResult]]:
|
||
"""
|
||
Full pipeline: build diffs, then apply (or simulate) changes.
|
||
|
||
Returns (diffs, results) so callers can inspect both.
|
||
"""
|
||
diffs = self.build_diff(group_names)
|
||
|
||
if whatif:
|
||
logger.info("--- WHATIF MODE: no changes will be written ---")
|
||
|
||
results: List[SyncResult] = []
|
||
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
|
||
futures = {
|
||
pool.submit(self._apply_diff, diff, whatif): diff.group_name
|
||
for diff in diffs
|
||
}
|
||
for fut in as_completed(futures):
|
||
results.append(fut.result())
|
||
|
||
# Sort results to match original group order
|
||
order = {name: i for i, name in enumerate(group_names)}
|
||
results.sort(key=lambda r: order.get(r.group_name, 9999))
|
||
|
||
return diffs, results
|
||
|
||
# ------------------------------------------------------------------
|
||
# Summary printer
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def print_summary(
|
||
diffs: List[GroupDiff],
|
||
results: List[SyncResult],
|
||
whatif: bool = False,
|
||
) -> None:
|
||
label = "[WHATIF] " if whatif else ""
|
||
print(f"\n{'='*60}")
|
||
print(f" {label}Sync Summary")
|
||
print(f"{'='*60}")
|
||
|
||
result_map = {r.group_name: r for r in results}
|
||
|
||
for diff in diffs:
|
||
r = result_map.get(diff.group_name)
|
||
print(f"\n Group: {diff.group_name}")
|
||
if diff.error:
|
||
print(f" STATUS : SKIPPED – {diff.error}")
|
||
continue
|
||
|
||
print(f" Source IPs : {len(diff.source_ips)}")
|
||
print(f" Target IPs : {len(diff.target_static_ips)}")
|
||
print(f" To Add : {len(diff.ips_to_add)}")
|
||
print(f" To Remove : {len(diff.ips_to_remove)}")
|
||
|
||
if r:
|
||
status_label = r.status.upper()
|
||
if r.status == "no_change":
|
||
status_label = "IN SYNC"
|
||
print(f" STATUS : {status_label}")
|
||
if r.error:
|
||
print(f" ERROR : {r.error}")
|
||
if r.ips_added:
|
||
for ip in r.ips_added:
|
||
print(f" + {ip}")
|
||
if r.ips_removed:
|
||
for ip in r.ips_removed:
|
||
print(f" - {ip}")
|
||
|
||
print(f"\n{'='*60}\n")
|