""" 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")