"""STAGE 1: collecting Neutron VPNaaS configuration into an audit JSON file, plus the --from-audit path that reconstructs the same in-memory state from a previously written audit file instead of re-querying OpenStack. This module owns the audit JSON schema definition (to_audit_dict / load_from_audit_json), so it's defined exactly once and shared by both the write path (STEP 10) and the read path (--from-audit). """ from __future__ import annotations import json import os import re import sys from datetime import datetime, timezone from . import keystone_auth, neutron_client from .models import MigrationState from .neutron_client import NeutronClient from .redact import redact_psk from .sprut_client import SprutClient from .tables import print_kv_table _UUID_LIKE = re.compile(r"^[0-9a-fA-F-]{36}$") def _dump(obj) -> str: return json.dumps(obj, indent=2) def _rule() -> None: print("*" * 86) # -------------------------------------------------------------------------- # STEP 1: read the input CSV and build the neutron -> advanced router map. # A row's advanced_router_id may be blank -- that router will be created in # Sprut later (STEP 3). Optional 3rd/4th columns (availability_zone, # flavor) only matter for such rows; they're a hint for that creation, # used instead of the interactive prompt when present. # -------------------------------------------------------------------------- def read_input_csv(input_csv: str) -> tuple[dict[str, str], dict[str, dict]]: print(f"Executing STEP 1: Reading config file {input_csv}") mapping: dict[str, str] = {} creation_hints: dict[str, dict] = {} try: handle = open(input_csv, encoding="utf-8") except OSError as exc: print(f"Error: could not open input file '{input_csv}': {exc}", file=sys.stderr) sys.exit(1) with handle: for raw_line in handle: line = raw_line.rstrip("\n").rstrip("\r") fields = [f.strip() for f in line.split(",")] neutron_router = fields[0] if fields else "" advanced_router = fields[1] if len(fields) > 1 else "" availability_zone = fields[2] if len(fields) > 2 else "" flavor = fields[3] if len(fields) > 3 else "" if not neutron_router: continue mapping[neutron_router] = advanced_router if not advanced_router: hint = {} if availability_zone: hint["availability_zone"] = availability_zone if flavor: hint["flavor"] = flavor if hint: creation_hints[neutron_router] = hint print( f"Neutron router '{neutron_router}' has no advanced router — " "a DC Router will be created for it in Sprut.", file=sys.stderr, ) print("Reading values from config:") print() print_kv_table( {k: (v or "") for k, v in mapping.items()}, "Neutron router", "Advanced Router" ) print() print("STEP 1 complete (config read)") _rule() return mapping, creation_hints # -------------------------------------------------------------------------- # STEP 2: verify every Neutron router from the CSV exists in OpenStack. # -------------------------------------------------------------------------- def check_neutron_routers_exist(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 2: Checking Neutron routers") listing = neutron_client.router_list(nc) state.neutron_router_by_id = {entry["id"]: entry for entry in listing} existing_ids = set(state.neutron_router_by_id) for neutron_router_id in state.neutron_to_adv_router: if neutron_router_id not in existing_ids: # No stderr redirect here -- matches the bash original's # (inconsistent, but preserved) use of stdout for this message. print(f"Error: Neutron router ID {neutron_router_id} not found in OpenStack tenant.") sys.exit(1) print(f"Neutron router ID {neutron_router_id} exists in OpenStack tenant.") print("STEP 2 complete (Neutron routers checked)") _rule() _DC_FLAVORS = ("basic", "standard", "advanced") def _require_interactive(neutron_router_id: str, field_name: str) -> None: if not sys.stdin.isatty(): print( f"Error: router '{neutron_router_id}' needs a DC Router created but '{field_name}' " "isn't set in the CSV and stdin is not interactive. Add it as a CSV column " "(3rd=availability_zone, 4th=flavor) or run interactively.", file=sys.stderr, ) sys.exit(1) def _resolve_az_flavor(neutron_router_id: str, hints: dict[str, dict]) -> tuple[str, str]: hint = hints.get(neutron_router_id, {}) availability_zone = hint.get("availability_zone") or "" if not availability_zone: _require_interactive(neutron_router_id, "availability_zone") while not availability_zone: availability_zone = input(f"Router '{neutron_router_id}': enter availability_zone: ").strip() flavor = hint.get("flavor") or "" if flavor not in _DC_FLAVORS: _require_interactive(neutron_router_id, "flavor") while flavor not in _DC_FLAVORS: flavor = input(f"Router '{neutron_router_id}': enter flavor ({'/'.join(_DC_FLAVORS)}): ").strip() return availability_zone, flavor def _find_internet_network(sprut_client: SprutClient) -> str: """Locates Sprut's public network (name "internet"/"Internet"), for attaching a DC Router's public interface. Looked up once per run, lazily, only when at least one router actually needs creation. Deliberately does not resolve or send a subnet_id: Sprut rejects dc_interface creation with subnet_id set for an external network ("Specifying subnet_id or ip_address for external network is restricted", confirmed against the real API) -- it auto-assigns one. """ networks = sprut_client.get("/networks?limit=10000").get("networks", []) matches = [n for n in networks if (n.get("name") or "").lower() == "internet"] if not matches: print("Error: no Sprut network named 'Internet' found (needed for the public interface).", file=sys.stderr) sys.exit(1) if len(matches) > 1: print( f"Warning: {len(matches)} Sprut networks named 'Internet' found -- " f"using the first one ({matches[0].get('id')}).", file=sys.stderr, ) return matches[0]["id"] def _find_or_create_dc_router( neutron_router_id: str, name: str, description: str, availability_zone: str, flavor: str, sprut_client: SprutClient, existing_dc_routers_by_name: dict[str, str], ) -> str: sprut_id = existing_dc_routers_by_name.get(name) if sprut_id: print(f"DC Router '{name}' already exists in Sprut (id {sprut_id}) -- reusing") return sprut_id print(f"Creating DC Router '{name}' in Sprut (az={availability_zone}, flavor={flavor}, enable_snat=true)") request_body = { "dc_router": { "availability_zone": availability_zone, "flavor": flavor, "enable_snat": True, "name": name, "description": description, } } response = sprut_client.post("/direct_connect/dc_routers", request_body) sprut_id = (response.get("dc_router") or {}).get("id") if sprut_client.dry_run and not sprut_id: sprut_id = f"DRY-RUN-{neutron_router_id}" return sprut_id def _find_or_create_dc_interface( dc_router_id: str, network_id: str, name: str, sprut_client: SprutClient ) -> str: existing = sprut_client.get("/direct_connect/dc_interfaces").get("dc_interfaces", []) match = next( (i for i in existing if i.get("dc_router_id") == dc_router_id and i.get("network_id") == network_id), None, ) if match: print(f"Public interface already attached to DC Router {dc_router_id} (id {match.get('id')}) -- reusing") return match["id"] print(f"Attaching public interface (network 'Internet') to DC Router {dc_router_id}") # subnet_id deliberately omitted -- Sprut rejects it for external # networks (see _find_internet_network's docstring). request_body = { "dc_interface": { "dc_router_id": dc_router_id, "network_id": network_id, "name": name, } } response = sprut_client.post("/direct_connect/dc_interfaces", request_body) sprut_id = (response.get("dc_interface") or {}).get("id") if sprut_client.dry_run and not sprut_id: sprut_id = f"DRY-RUN-iface-{dc_router_id}" return sprut_id def _ensure_dc_router( neutron_router_id: str, name: str, description: str, availability_zone: str, flavor: str, sprut_client: SprutClient, existing_dc_routers_by_name: dict[str, str], internet_network_id: str, ) -> str: """Creates (or reuses, by name) a DC Router and attaches its public interface. Shared by the fresh-collect path (STEP 3) and the --from-audit path, which both need identical create-or-reuse logic.""" sprut_id = _find_or_create_dc_router( neutron_router_id, name, description, availability_zone, flavor, sprut_client, existing_dc_routers_by_name ) _find_or_create_dc_interface(sprut_id, internet_network_id, f"{name}-public", sprut_client) existing_dc_routers_by_name[name] = sprut_id return sprut_id # -------------------------------------------------------------------------- # STEP 3: verify every advanced router exists in Sprut, using the Keystone # token issued once up front by collect(). For CSV rows with a blank # advanced_router_id, either creates the DC Router now (full run) or just # records what it would need (--audit-only, which must not write to # Sprut). Returns the SprutClient constructed with that token, reused by # STAGE 2-4 (the token is never refreshed after this point). # -------------------------------------------------------------------------- def check_advanced_routers_exist( state: MigrationState, token: str, sprut_base_url: str, dry_run: bool, audit_only: bool ) -> SprutClient: print("Executing STEP 3: Checking Advanced routers") sprut_client = SprutClient(sprut_base_url, token, dry_run) dc_routers = sprut_client.get("/direct_connect/dc_routers").get("dc_routers", []) existing_ids = {entry["id"] for entry in dc_routers} existing_by_name = {entry["name"]: entry["id"] for entry in dc_routers if entry.get("name")} internet_network_id: str | None = None for neutron_router_id, advanced_router_id in state.neutron_to_adv_router.items(): if advanced_router_id: if advanced_router_id not in existing_ids: print(f"Error: Advanced router ID {advanced_router_id} not found in SDN.") sys.exit(1) print(f"Advanced router ID {advanced_router_id} exists in SDN.") continue neutron_router = state.neutron_router_by_id.get(neutron_router_id, {}) name = neutron_router.get("name") or neutron_router_id description = neutron_router.get("description") or "" availability_zone, flavor = _resolve_az_flavor(neutron_router_id, state.router_creation_hints) if audit_only: state.pending_dc_router_by_neutron_id[neutron_router_id] = { "name": name, "description": description, "availability_zone": availability_zone, "flavor": flavor, } print( f"Neutron router '{neutron_router_id}': no advanced router -- recorded as pending " "DC Router creation in the audit (no Sprut writes in --audit-only)." ) continue if internet_network_id is None: internet_network_id = _find_internet_network(sprut_client) state.neutron_to_adv_router[neutron_router_id] = _ensure_dc_router( neutron_router_id, name, description, availability_zone, flavor, sprut_client, existing_by_name, internet_network_id, ) print("STEP 3 complete (Advanced routers checked)") _rule() return sprut_client def recheck_advanced_routers_from_audit(state: MigrationState, sprut_client: SprutClient) -> None: """Re-verifies advanced routers recorded in a --from-audit file still exist in Sprut, since the audit may have been generated earlier. Also creates any DC Router left pending by a prior --audit-only run, using the name/description/az/flavor cached in that audit's "pending_dc_router" block -- Neutron is never re-queried here.""" print("Issuing OpenStack token and re-checking advanced routers in SDN...") dc_routers = sprut_client.get("/direct_connect/dc_routers").get("dc_routers", []) existing_ids = {entry["id"] for entry in dc_routers} existing_by_name = {entry["name"]: entry["id"] for entry in dc_routers if entry.get("name")} internet_network_id: str | None = None for neutron_router_id, advanced_router_id in state.neutron_to_adv_router.items(): if advanced_router_id: if advanced_router_id not in existing_ids: print( f"Error: Advanced router ID {advanced_router_id} (from audit file) not found in SDN.", file=sys.stderr, ) sys.exit(1) print(f"Advanced router ID {advanced_router_id} exists in SDN.") continue pending = state.pending_dc_router_by_neutron_id.get(neutron_router_id) if not pending: print( f"Error: router '{neutron_router_id}' has no advanced_router_id and no " "pending_dc_router info in the audit file -- file is incompatible or corrupted.", file=sys.stderr, ) sys.exit(1) if internet_network_id is None: internet_network_id = _find_internet_network(sprut_client) state.neutron_to_adv_router[neutron_router_id] = _ensure_dc_router( neutron_router_id, pending["name"], pending.get("description", ""), pending["availability_zone"], pending["flavor"], sprut_client, existing_by_name, internet_network_id, ) # -------------------------------------------------------------------------- # STEP 4-6: IPsec policies, IKE policies, Endpoint groups (project-wide). # One list call each -- Neutron's list endpoints return full objects, so no # per-item show call is needed (unlike the old CLI-based version). # -------------------------------------------------------------------------- def collect_ipsec_policies(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 4: Collecting info about IPsec policies") for details in neutron_client.ipsec_policy_list(nc): state.neutron_ipsec_policy_by_id[details["id"]] = details print("Stored IPsec policies:") for policy_id, details in state.neutron_ipsec_policy_by_id.items(): print(f"Policy ID: {policy_id}") print(f"Details: {_dump(details)}") print() print("STEP 4 complete: Info about IPsec policies stored") _rule() def collect_ike_policies(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 5: Collecting info about IKE policies") for details in neutron_client.ike_policy_list(nc): state.neutron_ike_policy_by_id[details["id"]] = details print("Stored IKE policies:") for policy_id, details in state.neutron_ike_policy_by_id.items(): print(f"Policy ID: {policy_id}") print(f"Details: {_dump(details)}") print() print("STEP 5 complete: Info about IKE policies stored") _rule() def collect_endpoint_groups(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 6: Collecting info about Endpoint Groups") for details in neutron_client.endpoint_group_list(nc): eg_id = details["id"] state.neutron_endpoint_group_by_id[eg_id] = details if details.get("sdn") == "sprut": print(f"Endpoint Group {eg_id} has sdn 'sprut' with the following network addresses:") for address in details.get("endpoints") or []: print(address) print("Stored Endpoint Groups:") for eg_id, details in state.neutron_endpoint_group_by_id.items(): print(f"Endpoint Group ID: {eg_id}") print(f"Details: {_dump(details)}") print() print("STEP 6 complete: Info about Endpoint Groups stored") _rule() # -------------------------------------------------------------------------- # STEP 7: subnet id -> CIDR map, excluding ext-subnet* names. # -------------------------------------------------------------------------- def collect_subnets(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 7: Creating a map of Subnet ID to Subnet Address") for entry in neutron_client.subnet_list(nc): name = entry.get("name") or "" if name.startswith("ext-subnet"): continue state.subnet_id_to_subnet_address[entry["id"]] = entry.get("cidr") print("Stored Subnet ID to Subnet Address mapping:") print_kv_table(state.subnet_id_to_subnet_address, "Subnet ID", "Subnet Address", width1=36, width2=18) print() print("STEP 7 complete: Subnet ID to Subnet Address map created") _rule() # -------------------------------------------------------------------------- # STEP 8: router <-> vpn service maps. router_id_to_vpn_service_id is # filtered to in-scope (CSV) routers; the reverse map is deliberately left # project-wide since STEP 9 needs it to resolve every connection's owning # router before in-scope narrowing happens. # -------------------------------------------------------------------------- def collect_vpn_services(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 8: Collecting info about vpn services") listing = neutron_client.vpn_service_list(nc) for entry in listing: router_id = entry["router_id"] vpn_id = entry["id"] state.router_id_to_vpn_service_id.setdefault(router_id, vpn_id) state.neutron_vpn_service_by_id.setdefault(vpn_id, entry) print("Remove router_ids that are not in config") for router_id in list(state.router_id_to_vpn_service_id): if router_id not in state.neutron_to_adv_router: del state.router_id_to_vpn_service_id[router_id] print_kv_table(state.router_id_to_vpn_service_id, "Router ID", "VPN service ID") print("build backwards dictionary") for entry in listing: state.vpn_service_id_to_router_id[entry["id"]] = entry["router_id"] print_kv_table(state.vpn_service_id_to_router_id, "VPN service ID", "Router ID") print("STEP 8 complete (Router ID to VPN Service ID map built)") _rule() # -------------------------------------------------------------------------- # STEP 9: ipsec site connections + router -> connection-ids map. # -------------------------------------------------------------------------- def collect_ipsec_site_connections(state: MigrationState, nc: NeutronClient) -> None: print("Executing STEP 9: Collecting info about ipsec") for details in neutron_client.ipsec_site_connection_list(nc): conn_id = details["id"] vpn_service_id = details.get("vpnservice_id") router_id = state.vpn_service_id_to_router_id.get(vpn_service_id, "") state.neutron_router_to_ipsec_ids.setdefault(router_id, []).append(conn_id) state.neutron_connection_by_id[conn_id] = details print("Stored IPsec connections:") for conn_id, details in state.neutron_connection_by_id.items(): print(f"ID: {conn_id}") print(f"Details: {_dump(redact_psk(details))}") print() print("Neutron Router to IPsec IDs mapping:") for router_id, conn_ids in state.neutron_router_to_ipsec_ids.items(): print(f"Router ID: {router_id}") print(f"IPsec IDs: {', '.join(conn_ids)}") print("STEP 9 complete: Info about ipsec stored") _rule() # -------------------------------------------------------------------------- # In-scope computation: only routers/objects tied to the input CSV are # actually created in Sprut by STAGE 3/4, even though STEP 4-9 collect # project-wide (needed for matching against what Sprut already has). # -------------------------------------------------------------------------- def compute_in_scope(state: MigrationState) -> None: for neutron_router_id in state.neutron_to_adv_router: for conn_id in state.neutron_router_to_ipsec_ids.get(neutron_router_id, []): if not conn_id: continue state.in_scope_connection_ids.add(conn_id) conn_raw = state.neutron_connection_by_id[conn_id] ike_id = conn_raw.get("ikepolicy_id") ipsec_id = conn_raw.get("ipsecpolicy_id") local_eg_id = conn_raw.get("local_ep_group_id") peer_eg_id = conn_raw.get("peer_ep_group_id") if ike_id: state.in_scope_ike_policy_ids.add(ike_id) if ipsec_id: state.in_scope_ipsec_policy_ids.add(ipsec_id) if local_eg_id: state.in_scope_endpoint_group_ids.add(local_eg_id) if peer_eg_id: state.in_scope_endpoint_group_ids.add(peer_eg_id) print(f"In-scope IPsec site connections for this migration: {len(state.in_scope_connection_ids)}") # -------------------------------------------------------------------------- # Endpoint-group helpers, shared by STEP 10 (audit assembly) and STAGE 3 # (endpoint-group compare&create). # -------------------------------------------------------------------------- def resolve_endpoint_addresses(details: dict, subnet_map: dict[str, str]) -> list[str]: name = details.get("name") converted: list[str] = [] for endpoint in details.get("endpoints") or []: if _UUID_LIKE.match(endpoint): print(f"Converting subnet UUID {endpoint} in {name} endpoints", file=sys.stderr) address = subnet_map.get(endpoint) if address: print(f"Converted to {address}", file=sys.stderr) converted.append(address) else: print(f"Warning: Subnet ID {endpoint} not found in subnet_id_to_subnet_address map.", file=sys.stderr) else: converted.append(endpoint) return converted def build_endpoint_group_block(raw: dict | None, subnet_map: dict[str, str]) -> dict | None: if not raw: return None resolved = resolve_endpoint_addresses(raw, subnet_map) return { "id": raw.get("id"), "raw": raw, "resolved_endpoints": resolved, "already_migrated": raw.get("sdn") == "sprut", } # -------------------------------------------------------------------------- # STEP 10: assemble and write the audit JSON. # -------------------------------------------------------------------------- def to_audit_dict(state: MigrationState, input_file: str) -> dict: router_blocks = [] for neutron_router_id, advanced_router_id in state.neutron_to_adv_router.items(): vpn_id = state.router_id_to_vpn_service_id.get(neutron_router_id) vpn_service_json = state.neutron_vpn_service_by_id.get(vpn_id) if vpn_id else None connection_blocks = [] for conn_id in state.neutron_router_to_ipsec_ids.get(neutron_router_id, []): if not conn_id: continue conn_raw = state.neutron_connection_by_id[conn_id] ike_id = conn_raw.get("ikepolicy_id") ipsec_id = conn_raw.get("ipsecpolicy_id") local_eg_id = conn_raw.get("local_ep_group_id") peer_eg_id = conn_raw.get("peer_ep_group_id") ike_json = state.neutron_ike_policy_by_id.get(ike_id) ipsec_json = state.neutron_ipsec_policy_by_id.get(ipsec_id) local_eg_raw = state.neutron_endpoint_group_by_id.get(local_eg_id) peer_eg_raw = state.neutron_endpoint_group_by_id.get(peer_eg_id) connection_blocks.append( { "id": conn_raw.get("id"), "raw": conn_raw, "ike_policy": ike_json, "ipsec_policy": ipsec_json, "local_endpoint_group": build_endpoint_group_block(local_eg_raw, state.subnet_id_to_subnet_address), "peer_endpoint_group": build_endpoint_group_block(peer_eg_raw, state.subnet_id_to_subnet_address), } ) router_blocks.append( { "neutron_router_id": neutron_router_id, "advanced_router_id": advanced_router_id or None, "pending_dc_router": state.pending_dc_router_by_neutron_id.get(neutron_router_id), "vpn_service": vpn_service_json, "ipsec_site_connections": connection_blocks, } ) return { "audit_metadata": { "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "input_file": input_file, "stage": "STAGE1_AUDIT", }, "subnets": dict(state.subnet_id_to_subnet_address), "routers": router_blocks, } def write_audit_json(data: dict, output_file: str) -> None: tmp_path = f"{output_file}.tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) try: with open(tmp_path, encoding="utf-8") as f: json.load(f) except (OSError, json.JSONDecodeError): print("Error: failed to assemble valid JSON audit output", file=sys.stderr) os.remove(tmp_path) sys.exit(1) os.replace(tmp_path, output_file) os.chmod(output_file, 0o600) print(f"Audit JSON written to {output_file} (permissions restricted to owner: contains PSK values)") # -------------------------------------------------------------------------- # --from-audit: reconstruct MigrationState from a previously written audit # file, without any OpenStack query. Pure (no network I/O) -- token # reissuance and the Sprut recheck are the orchestration's job (cli.py), # via recheck_advanced_routers_from_audit() above. # -------------------------------------------------------------------------- def load_from_audit_json(audit_file: str) -> MigrationState: print(f"Loading audit data from {audit_file} (skipping OpenStack collection)") try: with open(audit_file, encoding="utf-8") as f: data = json.load(f) except (OSError, json.JSONDecodeError): print(f"Error: '{audit_file}' is not valid JSON.", file=sys.stderr) sys.exit(1) schema_stage = (data.get("audit_metadata") or {}).get("stage") if schema_stage != "STAGE1_AUDIT": print( f"Error: '{audit_file}' does not look like a STAGE1 audit file " "(audit_metadata.stage != STAGE1_AUDIT).", file=sys.stderr, ) sys.exit(1) state = MigrationState() for subnet_id, subnet_cidr in (data.get("subnets") or {}).items(): if not subnet_id: continue state.subnet_id_to_subnet_address[subnet_id] = subnet_cidr for router_json in data.get("routers") or []: neutron_router_id = router_json.get("neutron_router_id") advanced_router_id = router_json.get("advanced_router_id") or "" if not neutron_router_id: continue state.neutron_to_adv_router[neutron_router_id] = advanced_router_id if not advanced_router_id: pending = router_json.get("pending_dc_router") if pending: state.pending_dc_router_by_neutron_id[neutron_router_id] = pending vpn_service_json = router_json.get("vpn_service") if vpn_service_json: vpn_service_id = vpn_service_json.get("id") state.router_id_to_vpn_service_id[neutron_router_id] = vpn_service_id for conn_json in router_json.get("ipsec_site_connections") or []: conn_id = conn_json.get("id") if not conn_id: continue state.neutron_connection_by_id[conn_id] = conn_json.get("raw") state.in_scope_connection_ids.add(conn_id) ike_json = conn_json.get("ike_policy") if ike_json: ike_id = ike_json.get("id") state.neutron_ike_policy_by_id[ike_id] = ike_json state.in_scope_ike_policy_ids.add(ike_id) ipsec_json = conn_json.get("ipsec_policy") if ipsec_json: ipsec_id = ipsec_json.get("id") state.neutron_ipsec_policy_by_id[ipsec_id] = ipsec_json state.in_scope_ipsec_policy_ids.add(ipsec_id) local_eg_json = conn_json.get("local_endpoint_group") if local_eg_json: local_eg_id = local_eg_json.get("id") state.neutron_endpoint_group_by_id[local_eg_id] = local_eg_json.get("raw") state.in_scope_endpoint_group_ids.add(local_eg_id) peer_eg_json = conn_json.get("peer_endpoint_group") if peer_eg_json: peer_eg_id = peer_eg_json.get("id") state.neutron_endpoint_group_by_id[peer_eg_id] = peer_eg_json.get("raw") state.in_scope_endpoint_group_ids.add(peer_eg_id) print( f"Loaded from audit: {len(state.neutron_to_adv_router)} router(s), " f"{len(state.in_scope_connection_ids)} in-scope IPsec site connection(s)." ) return state # -------------------------------------------------------------------------- # Full STAGE 1 orchestration (STEP 1-9 + in-scope computation). STEP 10 # (write) is invoked separately by the caller so --audit-only and the full # run can share this without duplicating the write step. # -------------------------------------------------------------------------- def collect( input_csv: str, sprut_base_url: str, dry_run: bool, audit_only: bool ) -> tuple[MigrationState, SprutClient]: state = MigrationState() state.neutron_to_adv_router, state.router_creation_hints = read_input_csv(input_csv) # Authenticate once, up front: STEP 2 now needs a Neutron token just as # much as STEP 3 needs one for Sprut, since neither goes through the # `openstack` CLI's own per-call auth resolution anymore. token, catalog = keystone_auth.authenticate() neutron_base_url = keystone_auth.resolve_endpoint(catalog, service_type="network") nc = NeutronClient(neutron_base_url, token) check_neutron_routers_exist(state, nc) sprut_client = check_advanced_routers_exist(state, token, sprut_base_url, dry_run, audit_only) collect_ipsec_policies(state, nc) collect_ike_policies(state, nc) collect_endpoint_groups(state, nc) collect_subnets(state, nc) collect_vpn_services(state, nc) collect_ipsec_site_connections(state, nc) compute_in_scope(state) return state, sprut_client