set executable

When a portal fails to create (e.g. IP not on system), its ID is absent
from portal_id_map. Previously this caused every target referencing that
portal to be skipped entirely, which then cascaded to target-extent
failures.

Now each target's groups are filtered individually: groups with unmapped
portal or initiator references are dropped with a warning, and the target
is still created with the remaining valid groups. This keeps the target
and all its extent associations intact while clearly indicating which
groups need manual attention.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

README.md

@@ -0,0 +1,210 @@
+# TrueMigration
+
+A Python CLI tool for migrating TrueNAS configuration to a live destination system. Designed for systems integration teams working in pre-production deployment environments.
+
+## What It Does
+
+TrueMigration reads configuration from a source and re-creates it on a destination TrueNAS system via its WebSocket API. It also provides a destination audit mode for inspecting and cleaning up existing configuration before migration.
+
+**Supported source types:**
+- **TrueNAS debug archive** — the `.tgz` produced by **System → Save Debug** in the TrueNAS UI (SCALE and CORE)
+- **CSV files** — customer-supplied spreadsheets for migrating from non-TrueNAS sources
+
+**Supported migration types:**
+- SMB shares
+- NFS exports
+- iSCSI (extents, initiator groups, portals, targets, target-extent associations)
+
+## Requirements
+
+- Python 3.9+
+- No external packages — stdlib only
+
+## Usage
+
+### Interactive Mode (recommended)
+
+Run with no arguments. The wizard will guide you through the full workflow.
+
+```bash
+python -m truenas_migrate
+```
+
+or
+
+```bash
+python deploy.py
+```
+
+At startup the wizard presents two top-level options:
+
+```
+  1. Migrate configuration to a destination system
+  2. Audit destination system  (view and manage existing config)
+```
+
+#### Option 1 — Migrate
+
+Walks through:
+1. Source selection (archive or CSV)
+2. Destination host, port, and API key
+3. Migration scope (SMB / NFS / iSCSI, or all)
+4. iSCSI portal IP remapping (destination IPs differ from source; MPIO supported)
+5. Check for existing iSCSI config — offers to remove it before migration
+6. Per-share selection (choose a subset or migrate all)
+7. Dry run preview — shows what will be created, flags missing datasets or zvols
+8. Optional auto-creation of missing datasets and zvols
+9. Final confirmation and live apply
+
+#### Option 2 — Audit
+
+Connects to the destination and displays a full inventory:
+- SMB shares (name, path, enabled status)
+- NFS exports
+- iSCSI configuration (extents, initiators, portals, targets, associations)
+- ZFS datasets (with space used)
+- ZFS zvols (with allocated size)
+
+After displaying the inventory, offers selective deletion by category. Deletion safeguards:
+- SMB shares / NFS exports / iSCSI: standard `[y/N]` confirmation
+- Zvols: requires typing `DELETE` — data is permanently destroyed
+- Datasets: requires typing `DELETE` — all files and snapshots are permanently destroyed
+- A final confirmation gate is shown before any deletions execute
+
+### Command Line Mode — Archive Source
+
+```bash
+# Inspect the archive before doing anything
+python -m truenas_migrate --debug-tar debug.tgz --list-archive
+
+# Dry run — connects to destination but makes no changes
+python -m truenas_migrate \
+    --debug-tar  debug.tgz \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx" \
+    --dry-run
+
+# Live migration (all types)
+python -m truenas_migrate \
+    --debug-tar  debug.tgz \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx"
+
+# Migrate only SMB shares
+python -m truenas_migrate \
+    --debug-tar  debug.tgz \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx" \
+    --migrate smb
+
+# Migrate SMB and iSCSI, skip NFS
+python -m truenas_migrate \
+    --debug-tar  debug.tgz \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx" \
+    --migrate smb iscsi
+```
+
+### Command Line Mode — CSV Source
+
+Fill in the provided template files and pass them on the command line. You can supply one or both.
+
+```bash
+# Dry run from CSV files
+python -m truenas_migrate \
+    --smb-csv    smb_shares.csv \
+    --nfs-csv    nfs_shares.csv \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx" \
+    --dry-run
+
+# Live migration — SMB only from CSV
+python -m truenas_migrate \
+    --smb-csv    smb_shares.csv \
+    --dest       192.168.1.50 \
+    --api-key    "1-xxxxxxxxxxxx"
+```
+
+### CSV Templates
+
+Copy and fill in the templates included in this repository:
+
+| File | Purpose |
+|------|---------|
+| `smb_shares_template.csv` | One row per SMB share |
+| `nfs_shares_template.csv` | One row per NFS export |
+
+**SMB columns:** `Share Name` *(required)*, `Path` *(required)*, `Description`, `Purpose`, `Read Only`, `Browsable`, `Guest Access`, `Access-Based Enumeration`, `Hosts Allow`, `Hosts Deny`, `Time Machine`, `Enabled`
+
+**NFS columns:** `Path` *(required)*, `Description`, `Read Only`, `Map Root User`, `Map Root Group`, `Map All User`, `Map All Group`, `Security`, `Allowed Hosts`, `Allowed Networks`, `Enabled`
+
+Boolean columns accept `true` or `false`. List columns (`Hosts Allow`, `Hosts Deny`, `Security`, `Allowed Hosts`, `Allowed Networks`) accept space-separated values.
+
+Valid `Purpose` values: `NO_PRESET`, `DEFAULT_SHARE`, `ENHANCED_TIMEMACHINE`, `MULTI_PROTOCOL_NFS`, `PRIVATE_DATASETS`, `WORM_DROPBOX`
+
+Valid `Security` values: `SYS`, `KRB5`, `KRB5I`, `KRB5P`
+
+### Generating an API Key
+
+In the TrueNAS UI: top-right account menu → **API Keys** → **Add**.
+
+## iSCSI Migration Notes
+
+iSCSI configuration involves relational objects with IDs that differ between systems. TrueMigration handles this automatically:
+
+- **Creation order**: extents and initiator groups first (no dependencies), then portals, then targets (which reference portals and initiators), then target-extent associations
+- **ID remapping**: old source IDs are mapped to new destination IDs as each object is created; downstream objects are updated accordingly
+- **Portal IPs**: the wizard prompts for destination IP addresses for each portal. Enter multiple space-separated IPs for MPIO configurations
+- **Zvols**: DISK-type extents reference ZFS zvols. The dry run checks whether the required zvols exist on the destination. If any are missing, the wizard prompts for their size and creates them before the live run
+- **Existing config**: if the destination already has iSCSI objects, the wizard detects this and offers to remove them before migration begins
+
+## Conflict Policy
+
+TrueMigration never overwrites or deletes existing configuration on the destination. Conflicts are skipped:
+
+| Type | Conflict detected by |
+|------|----------------------|
+| SMB share | Share name (case-insensitive) |
+| NFS export | Export path (exact match) |
+| iSCSI extent | Extent name (case-insensitive) |
+| iSCSI initiator group | Comment field (case-insensitive) |
+| iSCSI portal | Set of listen IP addresses |
+| iSCSI target | Target name (case-insensitive) |
+| iSCSI target-extent | Target ID + LUN ID combination |
+
+Always run with `--dry-run` first to preview what will and won't be created.
+
+## Archive Compatibility
+
+| Source version | Archive format | Notes |
+|----------------|----------------|-------|
+| SCALE 24.04+ | ixdiagnose (lowercase dirs) | Combined JSON plugin files |
+| SCALE (older) | ixdiagnose (uppercase dirs) | Per-query JSON files |
+| CORE | freenas-debug / fndebug | Plain-text dumps with embedded JSON |
+| HA bundles (25.04+) | Outer .tgz + inner .txz per node | Active node archive selected automatically |
+
+## Project Structure
+
+```
+deploy.py                    # Entry point shim
+smb_shares_template.csv      # SMB CSV template
+nfs_shares_template.csv      # NFS CSV template
+truenas_migrate/
+    __main__.py              # python -m truenas_migrate entry point
+    colors.py                # ANSI color helpers and shared logger
+    summary.py               # Migration summary dataclass and report
+    archive.py               # Debug archive parser (SCALE + CORE)
+    csv_source.py            # CSV parser for non-TrueNAS sources
+    client.py                # TrueNAS WebSocket API client and utilities
+    migrate.py               # SMB, NFS, and iSCSI migration routines
+    cli.py                   # Interactive wizard and argument parser
+```
+
+## Safety Notes
+
+- **Never destructive by default** — the migration path only creates, never modifies or deletes existing destination config
+- **Dry run first** — always preview with `--dry-run` before applying changes
+- **Audit deletions require explicit confirmation** — zvol and dataset deletion requires typing `DELETE` and a final confirmation gate
+- SSL certificate verification is disabled by default (TrueNAS systems commonly use self-signed certs). Use `--verify-ssl` to enable it
+- Targets the TrueNAS 25.04+ WebSocket API endpoint (`wss://<host>/api/current`)
+- Exit code `2` is returned if any errors occurred during migration

nfs_shares_template.csv

@@ -0,0 +1,3 @@
+Path,Description,Read Only,Map Root User,Map Root Group,Map All User,Map All Group,Security,Allowed Hosts,Allowed Networks,Enabled
+/mnt/tank/data,Primary data export,false,root,wheel,,,SYS,,192.168.1.0/24,true
+/mnt/tank/media,Media files read-only,true,,,,,,,,true

smb_shares_template.csv

@@ -0,0 +1,3 @@
+Share Name,Path,Description,Purpose,Read Only,Browsable,Guest Access,Access-Based Enumeration,Hosts Allow,Hosts Deny,Time Machine,Enabled
+Accounting,/mnt/tank/accounting,Accounting department files,NO_PRESET,false,true,false,false,,,false,true
+Public,/mnt/tank/public,Public read-only share,NO_PRESET,true,true,true,false,,,false,true

truenas_migrate/archive.py

@@ -42,18 +42,24 @@ _CANDIDATES: dict[str, list[str]] = {
         "ixdiagnose/plugins/Sharing/sharing.nfs.query.json",
         "ixdiagnose/NFS/sharing.nfs.query.json",
     ],
+    "iscsi": [
+        "ixdiagnose/plugins/iscsi/iscsi_config.json",
+        "ixdiagnose/plugins/ISCSI/iscsi_config.json",
+    ],
 }
 
 # When a candidate file bundles multiple datasets, pull out the right sub-key.
 _KEY_WITHIN_FILE: dict[str, str] = {
     "smb_shares": "sharing_smb_query",
     "nfs_shares": "sharing_nfs_query",
+    # "iscsi" intentionally omitted — iscsi_config.json is used as-is
 }
 
 # Keyword fragments for heuristic fallback scan (SCALE archives only)
 _KEYWORDS: dict[str, list[str]] = {
     "smb_shares": ["sharing.smb", "smb_share", "sharing/smb", "smb_info"],
     "nfs_shares": ["sharing.nfs", "nfs_share", "sharing/nfs", "nfs_config"],
+    "iscsi":      ["iscsi_config", "iscsi/iscsi"],
 }
 
 # Presence of this path prefix identifies a TrueNAS CORE archive.
@@ -251,13 +257,14 @@ def _open_source_tar(tar_path: str):
 
 def parse_archive(tar_path: str) -> dict[str, Any]:
     """
-    Extract SMB shares and NFS shares from the debug archive.
-    Returns: {"smb_shares": list, "nfs_shares": list}
+    Extract SMB shares, NFS shares, and iSCSI configuration from the debug archive.
+    Returns: {"smb_shares": list, "nfs_shares": list, "iscsi": dict}
     """
     log.info("Opening archive: %s", tar_path)
     result: dict[str, Any] = {
         "smb_shares": [],
         "nfs_shares": [],
+        "iscsi": {},
     }
 
     try:
@@ -288,14 +295,33 @@ def parse_archive(tar_path: str) -> dict[str, Any]:
                                 result[key] = v
                                 break
 
+                # iSCSI — combined dict file, not a bare list
+                iscsi_raw = _find_data(tf, members, "iscsi")
+                if iscsi_raw and isinstance(iscsi_raw, dict):
+                    result["iscsi"] = {
+                        "global_config": iscsi_raw.get("global_config", {}),
+                        "portals":       iscsi_raw.get("portals", []),
+                        "initiators":    iscsi_raw.get("initiators", []),
+                        "targets":       iscsi_raw.get("targets", []),
+                        "extents":       iscsi_raw.get("extents", []),
+                        "targetextents": iscsi_raw.get("targetextents", []),
+                    }
+                elif iscsi_raw is not None:
+                    log.warning("  iscsi         →  unexpected format (expected dict)")
+
     except (tarfile.TarError, OSError) as exc:
         log.error("Failed to open archive: %s", exc)
         sys.exit(1)
 
+    iscsi = result["iscsi"]
     log.info(
-        "Parsed: %d SMB share(s), %d NFS share(s)",
+        "Parsed: %d SMB share(s), %d NFS share(s), "
+        "iSCSI: %d target(s) / %d extent(s) / %d portal(s)",
         len(result["smb_shares"]),
         len(result["nfs_shares"]),
+        len(iscsi.get("targets", [])),
+        len(iscsi.get("extents", [])),
+        len(iscsi.get("portals", [])),
     )
     return result
 

truenas_migrate/cli.py

@@ -1,9 +1,9 @@
 """
 truenas_migrate  –  TrueNAS Share Migration Tool
 =================================================
-Reads SMB shares and NFS shares from a TrueNAS debug archive (.tar / .tgz)
-produced by the built-in "Save Debug" feature, then re-creates them on a
-destination TrueNAS system via the JSON-RPC 2.0 WebSocket API (TrueNAS 25.04+).
+Reads SMB shares and NFS shares from either a TrueNAS debug archive (.tar / .tgz)
+or customer-supplied CSV files, then re-creates them on a destination TrueNAS
+system via the JSON-RPC 2.0 WebSocket API (TrueNAS 25.04+).
 
 SAFE BY DEFAULT
   • Existing shares are never overwritten or deleted.
@@ -12,7 +12,7 @@ SAFE BY DEFAULT
 REQUIREMENTS
   Python 3.9+ (stdlib only – no external packages needed)
 
-QUICK START
+QUICK START — Archive source
   # 1. Inspect your debug archive to confirm it contains the data you need:
   python -m truenas_migrate --debug-tar debug.tgz --list-archive
 
@@ -29,12 +29,14 @@ QUICK START
       --dest       192.168.1.50 \\
       --api-key    "1-xxxxxxxxxxxx"
 
-  # 4. Migrate only SMB shares (skip NFS):
+QUICK START — CSV source
+  # Fill in smb_shares_template.csv / nfs_shares_template.csv, then:
   python -m truenas_migrate \\
-      --debug-tar  debug.tgz \\
+      --smb-csv    smb_shares.csv \\
+      --nfs-csv    nfs_shares.csv \\
       --dest       192.168.1.50 \\
       --api-key    "1-xxxxxxxxxxxx" \\
-      --migrate smb
+      --dry-run
 
 CONFLICT POLICY
   Shares that already exist on the destination are silently skipped:
@@ -52,9 +54,16 @@ from pathlib import Path
 from typing import Optional
 
 from .archive import parse_archive, list_archive_and_exit
-from .client import TrueNASClient, check_dataset_paths, create_missing_datasets
-from .colors import log, _bold, _bold_cyan, _bold_yellow, _cyan, _dim, _green, _yellow
-from .migrate import migrate_smb_shares, migrate_nfs_shares
+from .client import (
+    TrueNASClient,
+    check_dataset_paths, create_missing_datasets,
+    check_iscsi_zvols, create_missing_zvols,
+    query_destination_inventory,
+    delete_smb_shares, delete_nfs_exports, delete_zvols, delete_datasets,
+)
+from .colors import log, _bold, _bold_cyan, _bold_green, _bold_red, _bold_yellow, _cyan, _dim, _green, _yellow
+from .csv_source import parse_csv_sources
+from .migrate import migrate_smb_shares, migrate_nfs_shares, migrate_iscsi, query_existing_iscsi, clear_iscsi_config
 from .summary import Summary
 
 
@@ -67,7 +76,13 @@ async def run(
     archive: Optional[dict] = None,
 ) -> Summary:
     if archive is None:
-        archive = parse_archive(args.debug_tar)
+        smb_csv = getattr(args, "smb_csv", None)
+        nfs_csv = getattr(args, "nfs_csv", None)
+        if smb_csv or nfs_csv:
+            archive = parse_csv_sources(smb_csv, nfs_csv)
+        else:
+            archive = parse_archive(args.debug_tar)
+
     migrate_set = set(args.migrate)
 
     if args.dry_run:
@@ -94,11 +109,20 @@ async def run(
             await migrate_nfs_shares(
                 client, archive["nfs_shares"], args.dry_run, summary)
 
+        if "iscsi" in migrate_set:
+            await migrate_iscsi(
+                client, archive.get("iscsi", {}), args.dry_run, summary)
+
         if args.dry_run and summary.paths_to_create:
             summary.missing_datasets = await check_dataset_paths(
                 client, summary.paths_to_create,
             )
 
+        if args.dry_run and summary.zvols_to_check:
+            summary.missing_zvols = await check_iscsi_zvols(
+                client, summary.zvols_to_check,
+            )
+
     return summary
 
 
@@ -106,6 +130,32 @@ async def run(
 # Interactive wizard helpers
 # ─────────────────────────────────────────────────────────────────────────────
 
+def _parse_size(s: str) -> int:
+    """Parse a human-friendly size string to bytes.  E.g. '100G', '500GiB', '1T'."""
+    s = s.strip().upper()
+    for suffix, mult in [
+        ("PIB", 1 << 50), ("PB", 1 << 50), ("P", 1 << 50),
+        ("TIB", 1 << 40), ("TB", 1 << 40), ("T", 1 << 40),
+        ("GIB", 1 << 30), ("GB", 1 << 30), ("G", 1 << 30),
+        ("MIB", 1 << 20), ("MB", 1 << 20), ("M", 1 << 20),
+        ("KIB", 1 << 10), ("KB", 1 << 10), ("K", 1 << 10),
+    ]:
+        if s.endswith(suffix):
+            try:
+                return int(float(s[:-len(suffix)]) * mult)
+            except ValueError:
+                pass
+    return int(s)  # plain bytes
+
+
+def _fmt_bytes(n: int) -> str:
+    """Format a byte count as a human-readable string."""
+    for suffix, div in [("TiB", 1 << 40), ("GiB", 1 << 30), ("MiB", 1 << 20), ("KiB", 1 << 10)]:
+        if n >= div:
+            return f"{n / div:.1f} {suffix}"
+    return f"{n} B"
+
+
 def _find_debug_archives(directory: str = ".") -> list[Path]:
     """Return sorted list of TrueNAS debug archives found in *directory*."""
     patterns = ("*.tgz", "*.tar.gz", "*.tar", "*.txz", "*.tar.xz")
@@ -133,6 +183,98 @@ def _confirm(label: str) -> bool:
         return False
 
 
+def _prompt_csv_path(share_type: str) -> Optional[str]:
+    """Prompt for a CSV file path.  Returns resolved path string or None if skipped."""
+    template = f"{share_type.lower()}_shares_template.csv"
+    print(f"  {_dim('(template: ' + template + ')')}")
+    while True:
+        raw = _prompt(f"  {share_type} shares CSV path (Enter to skip)")
+        if not raw:
+            return None
+        p = Path(raw)
+        if p.is_file():
+            return str(p)
+        print(f"  {_bold_red('File not found:')} {raw}")
+
+
+
+def _prompt_iscsi_portals(iscsi: dict) -> None:
+    """Walk each portal and prompt for destination IPs in-place."""
+    portals = iscsi.get("portals", [])
+    if not portals:
+        return
+
+    print(f"\n  {_bold('iSCSI Portal Configuration')}")
+    print(f"  {_dim('Portal IP addresses are unique per system and must be updated.')}")
+    print(f"  {_dim('For MPIO, enter multiple IPs separated by spaces.')}")
+
+    for portal in portals:
+        comment      = portal.get("comment", "")
+        listen       = portal.get("listen", [])
+        src_ips = "  ".join(f"{l['ip']}" for l in listen)
+
+        label = f"Portal {portal['id']}" + (f" ({comment!r})" if comment else "")
+        print(f"\n  {_bold(label)}")
+        print(f"  {_dim('Source IP(s):')} {src_ips}")
+
+        raw = _prompt("  Destination IP(s)").strip()
+        if not raw:
+            print(f"  {_yellow('⚠')} No IPs entered — keeping source IPs.")
+            continue
+
+        dest_ips = raw.split()
+        portal["listen"] = [{"ip": ip} for ip in dest_ips]
+        print(f"  {_green('✓')} Portal: {', '.join(dest_ips)}")
+    print()
+
+
+def _prompt_clear_existing_iscsi(host: str, port: int, api_key: str) -> None:
+    """
+    Check whether the destination already has iSCSI configuration.
+    If so, summarise what exists and offer to remove it before migration.
+    """
+    async def _check():
+        async with TrueNASClient(host=host, port=port, api_key=api_key, verify_ssl=False) as client:
+            return await query_existing_iscsi(client)
+
+    existing = asyncio.run(_check())
+    counts = {k: len(v) for k, v in existing.items()}
+    total = sum(counts.values())
+    if total == 0:
+        return
+
+    print(f"\n  {_bold_yellow('WARNING:')} Destination already has iSCSI configuration:")
+    labels = [
+        ("extents",       "extent(s)"),
+        ("initiators",    "initiator group(s)"),
+        ("portals",       "portal(s)"),
+        ("targets",       "target(s)"),
+        ("targetextents", "target-extent association(s)"),
+    ]
+    for key, label in labels:
+        n = counts[key]
+        if n:
+            print(f"    • {n} {label}")
+    print()
+    print(f"  {_dim('Keep existing:  new objects will be skipped if conflicts are detected.')}")
+    print(f"  {_dim('Remove existing: ALL iSCSI config will be deleted before migration.')}")
+    print()
+
+    raw = _prompt("  [K]eep existing  /  [R]emove all existing iSCSI config", default="K")
+    if raw.strip().lower().startswith("r"):
+        if _confirm(f"  Remove ALL {total} iSCSI object(s) from {host}?"):
+            async def _clear():
+                async with TrueNASClient(host=host, port=port, api_key=api_key, verify_ssl=False) as client:
+                    await clear_iscsi_config(client)
+            print()
+            asyncio.run(_clear())
+            print(f"  {_bold_cyan('✓')} iSCSI configuration cleared.\n")
+        else:
+            print(f"  {_yellow('–')} Removal cancelled — keeping existing config.\n")
+    else:
+        print(f"  {_dim('Keeping existing iSCSI configuration.')}\n")
+
+
 def _select_shares(shares: list[dict], share_type: str) -> list[dict]:
     """
     Display a numbered list of *shares* and return only those the user selects.
@@ -141,7 +283,7 @@ def _select_shares(shares: list[dict], share_type: str) -> list[dict]:
     if not shares:
         return shares
 
-    print(f"\n  {_bold(f'{share_type} shares in archive ({len(shares)}):')} \n")
+    print(f"\n  {_bold(f'{share_type} shares ({len(shares)}):')} \n")
     for i, share in enumerate(shares, 1):
         if share_type == "SMB":
             name = share.get("name", "<unnamed>")
@@ -184,45 +326,319 @@ def _select_shares(shares: list[dict], share_type: str) -> list[dict]:
     return selected
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# Destination audit wizard
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _print_inventory_report(host: str, inv: dict) -> None:
+    """Print a structured inventory of all configuration on the destination."""
+    smb   = inv.get("smb_shares", [])
+    nfs   = inv.get("nfs_exports", [])
+    ds    = inv.get("datasets", [])
+    zvols = inv.get("zvols", [])
+    ext   = inv.get("iscsi_extents", [])
+    init  = inv.get("iscsi_initiators", [])
+    portals = inv.get("iscsi_portals", [])
+    tgt   = inv.get("iscsi_targets", [])
+    te    = inv.get("iscsi_targetextents", [])
+
+    header = f"DESTINATION INVENTORY: {host}"
+    rule   = _bold_cyan("─" * (len(header) + 4))
+    print(f"\n  {rule}")
+    print(f"  {_bold_cyan('│')}  {_bold(header)}  {_bold_cyan('│')}")
+    print(f"  {rule}")
+
+    # SMB
+    if smb:
+        print(f"\n  {_bold(f'SMB Shares ({len(smb)})')}")
+        for s in smb:
+            name    = s.get("name", "<unnamed>")
+            path    = s.get("path", "")
+            enabled = "" if s.get("enabled", True) else _dim("  [disabled]")
+            print(f"    {_cyan('•')} {name:<24} {_dim(path)}{enabled}")
+    else:
+        print(f"\n  {_dim('SMB Shares: none')}")
+
+    # NFS
+    if nfs:
+        print(f"\n  {_bold(f'NFS Exports ({len(nfs)})')}")
+        for n in nfs:
+            path    = n.get("path", "<no path>")
+            enabled = "" if n.get("enabled", True) else _dim("  [disabled]")
+            print(f"    {_cyan('•')} {path}{enabled}")
+    else:
+        print(f"\n  {_dim('NFS Exports: none')}")
+
+    # iSCSI
+    has_iscsi = any([ext, init, portals, tgt, te])
+    if has_iscsi:
+        iscsi_total = len(ext) + len(init) + len(portals) + len(tgt) + len(te)
+        print(f"\n  {_bold(f'iSCSI Configuration ({iscsi_total} objects)')}")
+        if ext:
+            print(f"    {_bold('Extents')} ({len(ext)}):")
+            for e in ext:
+                kind    = e.get("type", "")
+                backing = e.get("disk") or e.get("path") or ""
+                print(f"      {_cyan('•')} {e.get('name', '<unnamed>'):<22} {_dim(kind + '  ' + backing)}")
+        if init:
+            print(f"    {_bold('Initiator Groups')} ({len(init)}):")
+            for i in init:
+                print(f"      {_cyan('•')} {i.get('comment') or '<no comment>'}")
+        if portals:
+            print(f"    {_bold('Portals')} ({len(portals)}):")
+            for p in portals:
+                ips     = ", ".join(l["ip"] for l in p.get("listen", []))
+                comment = p.get("comment", "")
+                label   = f"{comment}  " if comment else ""
+                print(f"      {_cyan('•')} {label}{_dim(ips)}")
+        if tgt:
+            print(f"    {_bold('Targets')} ({len(tgt)}):")
+            for t in tgt:
+                print(f"      {_cyan('•')} {t.get('name', '<unnamed>')}")
+        if te:
+            print(f"    {_bold('Target-Extent Associations')} ({len(te)})")
+    else:
+        print(f"\n  {_dim('iSCSI: none')}")
+
+    # Datasets
+    if ds:
+        print(f"\n  {_bold(f'Datasets ({len(ds)})')}")
+        for d in ds[:20]:
+            name       = d.get("id", "")
+            is_root    = "/" not in name
+            used_raw   = d.get("used", {})
+            used_bytes = used_raw.get("parsed", 0) if isinstance(used_raw, dict) else 0
+            used_str   = f"  {_fmt_bytes(used_bytes)} used" if used_bytes else ""
+            root_tag   = _dim("  (pool root)") if is_root else ""
+            print(f"    {_cyan('•')} {name}{root_tag}{_dim(used_str)}")
+        if len(ds) > 20:
+            print(f"    {_dim(f'… and {len(ds) - 20} more')}")
+    else:
+        print(f"\n  {_dim('Datasets: none')}")
+
+    # Zvols
+    if zvols:
+        print(f"\n  {_bold(f'Zvols ({len(zvols)})')}")
+        for z in zvols:
+            name   = z.get("id", "")
+            vs_raw = z.get("volsize", {})
+            vs     = vs_raw.get("parsed", 0) if isinstance(vs_raw, dict) else 0
+            vs_str = f"  {_fmt_bytes(vs)}" if vs else ""
+            print(f"    {_cyan('•')} {name}{_dim(vs_str)}")
+    else:
+        print(f"\n  {_dim('Zvols: none')}")
+
+    print()
+
+
+def _run_audit_wizard(host: str, port: int, api_key: str) -> None:
+    """Query destination inventory and offer to selectively delete configuration."""
+    print(f"\n  Querying {_bold(host)} …\n")
+
+    async def _query() -> dict:
+        async with TrueNASClient(host=host, port=port, api_key=api_key, verify_ssl=False) as client:
+            return await query_destination_inventory(client)
+
+    try:
+        inv = asyncio.run(_query())
+    except (OSError, PermissionError) as exc:
+        print(f"  {_bold_red('Connection failed:')} {exc}\n")
+        return
+
+    _print_inventory_report(host, inv)
+
+    total = sum(len(v) for v in inv.values())
+    if total == 0:
+        print(f"  {_dim('The destination appears to have no configuration.')}\n")
+        return
+
+    # ── Deletion options ───────────────────────────────────────────────────────
+    print(f"  {_bold_yellow('─' * 60)}")
+    print(f"  {_bold_yellow('DELETION OPTIONS')}")
+    print(f"  {_dim('You may choose to delete some or all of the configuration above.')}")
+    print(f"  {_bold_red('WARNING: Deleted datasets and zvols cannot be recovered — all data will be permanently lost.')}")
+    print()
+
+    has_iscsi   = any(inv[k] for k in ("iscsi_extents", "iscsi_initiators",
+                                        "iscsi_portals", "iscsi_targets",
+                                        "iscsi_targetextents"))
+    iscsi_count = sum(len(inv[k]) for k in ("iscsi_extents", "iscsi_initiators",
+                                             "iscsi_portals", "iscsi_targets",
+                                             "iscsi_targetextents"))
+    deletable_ds = [d for d in inv["datasets"] if "/" in d["id"]]
+
+    del_iscsi   = False
+    del_smb     = False
+    del_nfs     = False
+    del_zvols   = False
+    del_datasets = False
+
+    # iSCSI (must go first — uses zvols as backing)
+    if has_iscsi:
+        del_iscsi = _confirm(
+            f"  Delete ALL iSCSI configuration ({iscsi_count} objects)?"
+        )
+
+    # SMB
+    if inv["smb_shares"]:
+        del_smb = _confirm(
+            f"  Delete all {len(inv['smb_shares'])} SMB share(s)?"
+        )
+
+    # NFS
+    if inv["nfs_exports"]:
+        del_nfs = _confirm(
+            f"  Delete all {len(inv['nfs_exports'])} NFS export(s)?"
+        )
+
+    # Zvols — require explicit confirmation phrase
+    if inv["zvols"]:
+        print()
+        print(f"  {_bold_red('⚠  DATA DESTRUCTION WARNING  ⚠')}")
+        print(f"  Deleting zvols PERMANENTLY DESTROYS all data stored in them.")
+        print(f"  This action cannot be undone.  Affected zvols:")
+        for z in inv["zvols"]:
+            print(f"    {_yellow('•')} {z['id']}")
+        print()
+        raw = _prompt(
+            f"  Type DELETE to confirm deletion of {len(inv['zvols'])} zvol(s),"
+            " or Enter to skip"
+        ).strip()
+        del_zvols = (raw == "DELETE")
+        if raw and raw != "DELETE":
+            print(f"  {_dim('Confirmation not matched — zvols will not be deleted.')}")
+        print()
+
+    # Datasets — strongest warning
+    if deletable_ds:
+        print(f"  {_bold_red('⚠⚠  CRITICAL DATA DESTRUCTION WARNING  ⚠⚠')}")
+        print(f"  Deleting datasets PERMANENTLY DESTROYS ALL DATA including all files,")
+        print(f"  snapshots, and child datasets. Pool root datasets (e.g. 'tank') will")
+        print(f"  be skipped, but all child datasets WILL be deleted.")
+        print(f"  This action cannot be undone.  {len(deletable_ds)} dataset(s) would be deleted.")
+        print()
+        raw = _prompt(
+            f"  Type DELETE to confirm deletion of {len(deletable_ds)} dataset(s),"
+            " or Enter to skip"
+        ).strip()
+        del_datasets = (raw == "DELETE")
+        if raw and raw != "DELETE":
+            print(f"  {_dim('Confirmation not matched — datasets will not be deleted.')}")
+        print()
+
+    # ── Nothing selected ───────────────────────────────────────────────────────
+    if not any([del_iscsi, del_smb, del_nfs, del_zvols, del_datasets]):
+        print(f"  {_dim('Nothing selected for deletion. No changes made.')}\n")
+        return
+
+    # ── Final confirmation ─────────────────────────────────────────────────────
+    print(f"  {_bold_yellow('─' * 60)}")
+    print(f"  {_bold_yellow('PENDING DELETIONS on ' + host + ':')}")
+    if del_iscsi:
+        print(f"    {_yellow('•')} ALL iSCSI configuration  ({iscsi_count} objects)")
+    if del_smb:
+        print(f"    {_yellow('•')} {len(inv['smb_shares'])} SMB share(s)")
+    if del_nfs:
+        print(f"    {_yellow('•')} {len(inv['nfs_exports'])} NFS export(s)")
+    if del_zvols:
+        print(f"    {_bold_red('•')} {len(inv['zvols'])} zvol(s)  "
+              f"{_bold_red('⚠ ALL DATA WILL BE PERMANENTLY DESTROYED')}")
+    if del_datasets:
+        print(f"    {_bold_red('•')} {len(deletable_ds)} dataset(s)  "
+              f"{_bold_red('⚠ ALL DATA WILL BE PERMANENTLY DESTROYED')}")
+    print()
+    print(f"  {_bold_red('THIS ACTION CANNOT BE UNDONE.')}")
+    print()
+
+    if not _confirm(f"  Proceed with all selected deletions on {host}?"):
+        print(f"  {_dim('Aborted – no changes made.')}\n")
+        return
+
+    # ── Execute ────────────────────────────────────────────────────────────────
+    print()
+
+    async def _execute() -> None:
+        async with TrueNASClient(host=host, port=port, api_key=api_key, verify_ssl=False) as client:
+            if del_iscsi:
+                print(f"  Removing iSCSI configuration …")
+                await clear_iscsi_config(client)
+                print(f"    {_bold_green('✓')} iSCSI configuration removed.")
+
+            if del_smb:
+                print(f"  Removing SMB shares …")
+                ok, fail = await delete_smb_shares(client, inv["smb_shares"])
+                suffix = f"  {_bold_red(str(fail) + ' failed')}" if fail else ""
+                print(f"    {_bold_green('✓')} {ok} deleted{suffix}")
+
+            if del_nfs:
+                print(f"  Removing NFS exports …")
+                ok, fail = await delete_nfs_exports(client, inv["nfs_exports"])
+                suffix = f"  {_bold_red(str(fail) + ' failed')}" if fail else ""
+                print(f"    {_bold_green('✓')} {ok} deleted{suffix}")
+
+            if del_zvols:
+                print(f"  Removing zvols …")
+                ok, fail = await delete_zvols(client, inv["zvols"])
+                suffix = f"  {_bold_red(str(fail) + ' failed')}" if fail else ""
+                print(f"    {_bold_green('✓')} {ok} deleted{suffix}")
+
+            if del_datasets:
+                print(f"  Removing datasets …")
+                ok, fail = await delete_datasets(client, deletable_ds)
+                suffix = f"  {_bold_red(str(fail) + ' failed')}" if fail else ""
+                print(f"    {_bold_green('✓')} {ok} deleted{suffix}")
+
+    asyncio.run(_execute())
+    print(f"\n  {_bold_cyan('Done.')}\n")
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Interactive wizard
 # ─────────────────────────────────────────────────────────────────────────────
 
 def interactive_mode() -> None:
-    """Interactive wizard: pick archive → configure → dry run → confirm → apply."""
+    """Interactive wizard: pick source → configure → dry run → confirm → apply."""
     print(
         f"\n{_bold_cyan('  TrueNAS Share Migration Tool')}\n"
-        f"  {_dim('Migrate SMB/NFS shares from a debug archive to a live system.')}\n"
+        f"  {_dim('Migrate SMB/NFS shares to a live TrueNAS system.')}\n"
     )
 
-    # 1 ── Locate debug archive ────────────────────────────────────────────────
-    archives = _find_debug_archives()
-    if not archives:
-        sys.exit(
-            "No debug archives (.tgz / .tar.gz / .tar / .txz) found in the "
-            "current directory.\n"
-            "Copy your TrueNAS debug file here, or use --debug-tar to specify a path."
-        )
-
-    if len(archives) == 1:
-        chosen = archives[0]
-        print(f"  {_dim('Archive:')} {_bold(chosen.name)}  "
-              f"{_dim('(' + f'{chosen.stat().st_size / 1_048_576:.1f} MB' + ')')}\n")
-    else:
-        print(f"  {_bold('Debug archives found:')}\n")
-        for i, p in enumerate(archives, 1):
-            print(f"  {_cyan(str(i) + '.')} {p.name}  "
-                  f"{_dim('(' + f'{p.stat().st_size / 1_048_576:.1f} MB' + ')')}")
-        print()
-        while True:
-            raw = _prompt(f"Select archive [1-{len(archives)}]")
-            if raw.isdigit() and 1 <= int(raw) <= len(archives):
-                chosen = archives[int(raw) - 1]
-                break
-            print(f"  Enter a number from 1 to {len(archives)}.")
-
-    # 2 ── Destination ─────────────────────────────────────────────────────────
+    # 0 ── Top-level action ─────────────────────────────────────────────────────
+    print(f"  {_bold('What would you like to do?')}")
+    print(f"  {_cyan('1.')} Migrate configuration to a destination system")
+    print(f"  {_cyan('2.')} Audit destination system  (view and manage existing config)")
+    action_raw = _prompt("  Select [1/2]", default="1")
     print()
+
+    if action_raw.strip() == "2":
+        audit_host = ""
+        while not audit_host:
+            audit_host = _prompt("Destination TrueNAS host or IP")
+            if not audit_host:
+                print("  Host is required.")
+        audit_port_raw = _prompt("WebSocket port", default="443")
+        audit_port = int(audit_port_raw) if audit_port_raw.isdigit() else 443
+        audit_key = ""
+        while not audit_key:
+            try:
+                audit_key = getpass.getpass("API key (input hidden): ").strip()
+            except (EOFError, KeyboardInterrupt):
+                print()
+                sys.exit(0)
+            if not audit_key:
+                print("  API key is required.")
+        _run_audit_wizard(audit_host, audit_port, audit_key)
+        return
+
+    # 1 ── Source type ──────────────────────────────────────────────────────────
+    print(f"  {_bold('Source type:')}")
+    print(f"  {_cyan('1.')} TrueNAS debug archive  (.tgz / .tar)")
+    print(f"  {_cyan('2.')} CSV import  (non-TrueNAS source)")
+    src_raw = _prompt("  Select source [1/2]", default="1")
+    use_csv = src_raw.strip() == "2"
+    print()
+
+    # 2 ── Destination ──────────────────────────────────────────────────────────
     host = ""
     while not host:
         host = _prompt("Destination TrueNAS host or IP")
@@ -232,7 +648,7 @@ def interactive_mode() -> None:
     port_raw = _prompt("WebSocket port", default="443")
     port = int(port_raw) if port_raw.isdigit() else 443
 
-    # 3 ── API key ─────────────────────────────────────────────────────────────
+    # 3 ── API key ──────────────────────────────────────────────────────────────
     api_key = ""
     while not api_key:
         try:
@@ -243,26 +659,83 @@ def interactive_mode() -> None:
         if not api_key:
             print("  API key is required.")
 
-    # 4 ── Migration scope ─────────────────────────────────────────────────────
-    print(f"\n  {_bold('What to migrate?')}")
-    print(f"  {_cyan('1.')} SMB shares")
-    print(f"  {_cyan('2.')} NFS shares")
-    sel_raw = _prompt(
-        "Selection (space-separated numbers, Enter for all)", default="1 2"
-    )
-    _sel_map = {"1": "smb", "2": "nfs"}
-    migrate: list[str] = []
-    for tok in sel_raw.split():
-        if tok in _sel_map and _sel_map[tok] not in migrate:
-            migrate.append(_sel_map[tok])
-    if not migrate:
-        migrate = ["smb", "nfs"]
+    if use_csv:
+        # ── CSV source ──────────────────────────────────────────────────────────
+        print(f"\n  {_bold('CSV file paths:')}")
+        print(f"  {_dim('Press Enter to skip a share type.')}\n")
+        smb_csv_path = _prompt_csv_path("SMB")
+        print()
+        nfs_csv_path = _prompt_csv_path("NFS")
 
-    # 5 ── Parse archive once (reused for dry + live runs) ────────────────────
-    print()
-    archive_data = parse_archive(str(chosen))
+        migrate: list[str] = []
+        if smb_csv_path:
+            migrate.append("smb")
+        if nfs_csv_path:
+            migrate.append("nfs")
+        if not migrate:
+            sys.exit("No CSV files provided – nothing to migrate.")
 
-    # 5b ── Select individual shares ───────────────────────────────────────────
+        print()
+        archive_data = parse_csv_sources(smb_csv_path, nfs_csv_path)
+        extra_ns: dict = {"smb_csv": smb_csv_path, "nfs_csv": nfs_csv_path}
+
+    else:
+        # ── Archive source ──────────────────────────────────────────────────────
+        archives = _find_debug_archives()
+        if not archives:
+            sys.exit(
+                "No debug archives (.tgz / .tar.gz / .tar / .txz) found in the "
+                "current directory.\n"
+                "Copy your TrueNAS debug file here, or use --debug-tar to specify a path."
+            )
+
+        if len(archives) == 1:
+            chosen = archives[0]
+            print(f"  {_dim('Archive:')} {_bold(chosen.name)}  "
+                  f"{_dim('(' + f'{chosen.stat().st_size / 1_048_576:.1f} MB' + ')')}\n")
+        else:
+            print(f"  {_bold('Debug archives found:')}\n")
+            for i, p in enumerate(archives, 1):
+                print(f"  {_cyan(str(i) + '.')} {p.name}  "
+                      f"{_dim('(' + f'{p.stat().st_size / 1_048_576:.1f} MB' + ')')}")
+            print()
+            while True:
+                raw = _prompt(f"Select archive [1-{len(archives)}]")
+                if raw.isdigit() and 1 <= int(raw) <= len(archives):
+                    chosen = archives[int(raw) - 1]
+                    break
+                print(f"  Enter a number from 1 to {len(archives)}.")
+
+        # ── Migration scope ─────────────────────────────────────────────────────
+        print(f"\n  {_bold('What to migrate?')}")
+        print(f"  {_cyan('1.')} SMB shares")
+        print(f"  {_cyan('2.')} NFS shares")
+        print(f"  {_cyan('3.')} iSCSI (targets, extents, portals, initiator groups)")
+        sel_raw = _prompt(
+            "Selection (space-separated numbers, Enter for all)", default="1 2 3"
+        )
+        _sel_map = {"1": "smb", "2": "nfs", "3": "iscsi"}
+        migrate = []
+        for tok in sel_raw.split():
+            if tok in _sel_map and _sel_map[tok] not in migrate:
+                migrate.append(_sel_map[tok])
+        if not migrate:
+            migrate = ["smb", "nfs", "iscsi"]
+
+        # ── Parse archive ───────────────────────────────────────────────────────
+        print()
+        archive_data = parse_archive(str(chosen))
+        extra_ns = {"debug_tar": str(chosen)}
+
+        # ── iSCSI portal IP remapping ────────────────────────────────────────
+        if "iscsi" in migrate and archive_data.get("iscsi", {}).get("portals"):
+            _prompt_iscsi_portals(archive_data["iscsi"])
+
+        # ── iSCSI pre-migration check ────────────────────────────────────────
+        if "iscsi" in migrate:
+            _prompt_clear_existing_iscsi(host, port, api_key)
+
+    # ── Select individual shares (common) ──────────────────────────────────────
     if "smb" in migrate and archive_data["smb_shares"]:
         archive_data["smb_shares"] = _select_shares(archive_data["smb_shares"], "SMB")
     if "nfs" in migrate and archive_data["nfs_shares"]:
@@ -270,15 +743,15 @@ def interactive_mode() -> None:
     print()
 
     base_ns = dict(
-        debug_tar=str(chosen),
         dest=host,
         port=port,
         api_key=api_key,
         verify_ssl=False,
         migrate=migrate,
+        **extra_ns,
     )
 
-    # 6 ── Dry run ─────────────────────────────────────────────────────────────
+    # 6 ── Dry run ──────────────────────────────────────────────────────────────
     dry_summary = asyncio.run(
         run(argparse.Namespace(**base_ns, dry_run=True), archive_data)
     )
@@ -310,11 +783,40 @@ def interactive_mode() -> None:
                 ))
                 print()
 
+    if dry_summary.missing_zvols:
+        print(f"\n  {len(dry_summary.missing_zvols)} zvol(s) need to be created for iSCSI extents:")
+        for z in dry_summary.missing_zvols:
+            print(f"    • {z}")
+        print()
+        if _confirm(f"Create these {len(dry_summary.missing_zvols)} zvol(s) on {host} now?"):
+            zvol_sizes: dict[str, int] = {}
+            for zvol in dry_summary.missing_zvols:
+                while True:
+                    raw = _prompt(f"  Size for {zvol} (e.g. 100G, 500GiB, 1T)").strip()
+                    if not raw:
+                        print("  Size is required.")
+                        continue
+                    try:
+                        zvol_sizes[zvol] = _parse_size(raw)
+                        break
+                    except ValueError:
+                        print(f"  Cannot parse {raw!r} — try a format like 100G or 500GiB.")
+            asyncio.run(create_missing_zvols(
+                host=host, port=port, api_key=api_key, zvols=zvol_sizes,
+            ))
+            print()
+            print(f"  Re-running dry run to verify zvol creation …")
+            print()
+            dry_summary = asyncio.run(
+                run(argparse.Namespace(**base_ns, dry_run=True), archive_data)
+            )
+            print(dry_summary.report())
+
     if not _confirm(f"Apply these changes to {host}?"):
         print("Aborted – no changes made.")
         sys.exit(0)
 
-    # 7 ── Live run ────────────────────────────────────────────────────────────
+    # 7 ── Live run ─────────────────────────────────────────────────────────────
     print()
     live_summary = asyncio.run(
         run(argparse.Namespace(**base_ns, dry_run=False), archive_data)
@@ -336,23 +838,32 @@ def main() -> None:
     p = argparse.ArgumentParser(
         prog="truenas_migrate",
         description=(
-            "Migrate SMB and NFS shares from a TrueNAS debug archive "
-            "to a live destination system."
+            "Migrate SMB and NFS shares to a live TrueNAS destination system.  "
+            "Source can be a TrueNAS debug archive or customer-supplied CSV files."
         ),
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog=__doc__,
     )
 
     # ── Source ────────────────────────────────────────────────────────────────
-    p.add_argument(
-        "--debug-tar", required=True, metavar="FILE",
+    src = p.add_argument_group("source (choose one)")
+    src.add_argument(
+        "--debug-tar", metavar="FILE",
         help="Path to the TrueNAS debug .tar / .tgz from the SOURCE system.",
     )
+    src.add_argument(
+        "--smb-csv", metavar="FILE",
+        help="Path to a CSV file containing SMB share definitions (non-TrueNAS source).",
+    )
+    src.add_argument(
+        "--nfs-csv", metavar="FILE",
+        help="Path to a CSV file containing NFS share definitions (non-TrueNAS source).",
+    )
     p.add_argument(
         "--list-archive", action="store_true",
         help=(
             "List all JSON files found in the archive and exit.  "
-            "Run this first to verify the archive contains share data."
+            "Requires --debug-tar."
         ),
     )
 
@@ -386,17 +897,17 @@ def main() -> None:
     p.add_argument(
         "--migrate",
         nargs="+",
-        choices=["smb", "nfs"],
-        default=["smb", "nfs"],
+        choices=["smb", "nfs", "iscsi"],
+        default=["smb", "nfs", "iscsi"],
         metavar="TYPE",
         help=(
-            "What to migrate.  Choices: smb  nfs "
+            "What to migrate.  Choices: smb  nfs  iscsi "
             "(default: both).  Example: --migrate smb"
         ),
     )
     p.add_argument(
         "--dry-run", action="store_true",
-        help="Parse archive and connect to destination, but make no changes.",
+        help="Parse source and connect to destination, but make no changes.",
     )
     p.add_argument(
         "--verbose", "-v", action="store_true",
@@ -408,14 +919,32 @@ def main() -> None:
     if args.verbose:
         log.setLevel(logging.DEBUG)
 
-    if not Path(args.debug_tar).is_file():
-        p.error(f"Archive not found: {args.debug_tar}")
+    has_archive = bool(args.debug_tar)
+    has_csv     = bool(args.smb_csv or args.nfs_csv)
 
-    if args.list_archive:
-        list_archive_and_exit(args.debug_tar)   # does not return
+    if has_archive and has_csv:
+        p.error("Cannot combine --debug-tar with --smb-csv / --nfs-csv.")
+
+    if not has_archive and not has_csv:
+        p.error(
+            "Specify a source: --debug-tar FILE  or  --smb-csv / --nfs-csv FILE(s)."
+        )
+
+    if has_archive:
+        if not Path(args.debug_tar).is_file():
+            p.error(f"Archive not found: {args.debug_tar}")
+        if args.list_archive:
+            list_archive_and_exit(args.debug_tar)   # does not return
+    else:
+        if args.list_archive:
+            p.error("--list-archive requires --debug-tar.")
+        if args.smb_csv and not Path(args.smb_csv).is_file():
+            p.error(f"SMB CSV not found: {args.smb_csv}")
+        if args.nfs_csv and not Path(args.nfs_csv).is_file():
+            p.error(f"NFS CSV not found: {args.nfs_csv}")
 
     if not args.dest:
-        p.error("--dest is required (or use --list-archive to inspect the archive).")
+        p.error("--dest is required.")
     if not args.api_key:
         p.error("--api-key is required.")
 

truenas_migrate/client.py

@@ -306,3 +306,183 @@ async def create_missing_datasets(
     ) as client:
         for path in paths:
             await create_dataset(client, path)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# iSCSI zvol utilities
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def check_iscsi_zvols(
+    client: TrueNASClient,
+    zvol_names: list[str],
+) -> list[str]:
+    """
+    Return the subset of *zvol_names* that do not exist on the destination.
+    Names are the dataset path without the leading 'zvol/' prefix
+    (e.g. 'tank/VMWARE001').  Returns [] when the query itself fails.
+    """
+    if not zvol_names:
+        return []
+
+    unique = sorted(set(zvol_names))
+    log.info("Checking %d zvol(s) against destination datasets …", len(unique))
+    try:
+        datasets = await client.call(
+            "pool.dataset.query", [[["type", "=", "VOLUME"]]]
+        ) or []
+    except RuntimeError as exc:
+        log.warning("Could not query zvols (skipping check): %s", exc)
+        return []
+
+    existing = {d["name"] for d in datasets}
+    missing = [n for n in unique if n not in existing]
+    if missing:
+        for n in missing:
+            log.warning("  MISSING zvol: %s", n)
+    else:
+        log.info("  All iSCSI zvols exist on destination.")
+    return missing
+
+
+async def create_zvol(
+    client: TrueNASClient,
+    name: str,
+    volsize: int,
+) -> bool:
+    """
+    Create a ZFS volume (zvol) on the destination.
+    *name* is the dataset path (e.g. 'tank/VMWARE001').
+    *volsize* is the size in bytes.
+    Returns True on success, False on failure.
+    """
+    log.info("Creating zvol %r (%d bytes) …", name, volsize)
+    try:
+        await client.call("pool.dataset.create", [{
+            "name":    name,
+            "type":    "VOLUME",
+            "volsize": volsize,
+        }])
+        log.info("  Created: %s", name)
+        return True
+    except RuntimeError as exc:
+        log.error("  Failed to create zvol %r: %s", name, exc)
+        return False
+
+
+async def create_missing_zvols(
+    host: str,
+    port: int,
+    api_key: str,
+    zvols: dict[str, int],
+    verify_ssl: bool = False,
+) -> None:
+    """Open a fresh connection and create zvols from {name: volsize_bytes}."""
+    async with TrueNASClient(
+        host=host, port=port, api_key=api_key, verify_ssl=verify_ssl,
+    ) as client:
+        for name, volsize in zvols.items():
+            await create_zvol(client, name, volsize)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Destination inventory
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def query_destination_inventory(client: TrueNASClient) -> dict[str, list]:
+    """
+    Query all current configuration from the destination system.
+    Returns a dict with keys: smb_shares, nfs_exports, datasets, zvols,
+    iscsi_extents, iscsi_initiators, iscsi_portals, iscsi_targets, iscsi_targetextents.
+    Each value is a list (may be empty if the query fails or returns nothing).
+    """
+    result: dict[str, list] = {}
+    for key, method, params in [
+        ("smb_shares",          "sharing.smb.query",        None),
+        ("nfs_exports",         "sharing.nfs.query",        None),
+        ("datasets",            "pool.dataset.query",       [[["type", "=", "FILESYSTEM"]]]),
+        ("zvols",               "pool.dataset.query",       [[["type", "=", "VOLUME"]]]),
+        ("iscsi_extents",       "iscsi.extent.query",       None),
+        ("iscsi_initiators",    "iscsi.initiator.query",    None),
+        ("iscsi_portals",       "iscsi.portal.query",       None),
+        ("iscsi_targets",       "iscsi.target.query",       None),
+        ("iscsi_targetextents", "iscsi.targetextent.query", None),
+    ]:
+        try:
+            result[key] = await client.call(method, params) or []
+        except RuntimeError as exc:
+            log.warning("Could not query %s: %s", key, exc)
+            result[key] = []
+    return result
+
+
+async def delete_smb_shares(
+    client: TrueNASClient, shares: list[dict]
+) -> tuple[int, int]:
+    """Delete SMB shares by ID. Returns (deleted, failed)."""
+    deleted = failed = 0
+    for share in shares:
+        try:
+            await client.call("sharing.smb.delete", [share["id"]])
+            log.info("  Deleted SMB share %r", share.get("name"))
+            deleted += 1
+        except RuntimeError as exc:
+            log.error("  Failed to delete SMB share %r: %s", share.get("name"), exc)
+            failed += 1
+    return deleted, failed
+
+
+async def delete_nfs_exports(
+    client: TrueNASClient, exports: list[dict]
+) -> tuple[int, int]:
+    """Delete NFS exports by ID. Returns (deleted, failed)."""
+    deleted = failed = 0
+    for export in exports:
+        try:
+            await client.call("sharing.nfs.delete", [export["id"]])
+            log.info("  Deleted NFS export %r", export.get("path"))
+            deleted += 1
+        except RuntimeError as exc:
+            log.error("  Failed to delete NFS export %r: %s", export.get("path"), exc)
+            failed += 1
+    return deleted, failed
+
+
+async def delete_zvols(
+    client: TrueNASClient, zvols: list[dict]
+) -> tuple[int, int]:
+    """Delete zvols. Returns (deleted, failed)."""
+    deleted = failed = 0
+    for zvol in zvols:
+        try:
+            await client.call("pool.dataset.delete", [zvol["id"], {"recursive": True}])
+            log.info("  Deleted zvol %r", zvol["id"])
+            deleted += 1
+        except RuntimeError as exc:
+            log.error("  Failed to delete zvol %r: %s", zvol["id"], exc)
+            failed += 1
+    return deleted, failed
+
+
+async def delete_datasets(
+    client: TrueNASClient, datasets: list[dict]
+) -> tuple[int, int]:
+    """
+    Delete datasets deepest-first to avoid parent-before-child errors.
+    Skips pool root datasets (no '/' in the dataset name).
+    Returns (deleted, failed).
+    """
+    sorted_ds = sorted(
+        (d for d in datasets if "/" in d["id"]),
+        key=lambda d: d["id"].count("/"),
+        reverse=True,
+    )
+    deleted = failed = 0
+    for ds in sorted_ds:
+        try:
+            await client.call("pool.dataset.delete", [ds["id"], {"recursive": True}])
+            log.info("  Deleted dataset %r", ds["id"])
+            deleted += 1
+        except RuntimeError as exc:
+            log.error("  Failed to delete dataset %r: %s", ds["id"], exc)
+            failed += 1
+    return deleted, failed

truenas_migrate/csv_source.py

@@ -0,0 +1,209 @@
+"""CSV source parser – reads SMB/NFS share definitions from customer-supplied CSV files."""
+from __future__ import annotations
+
+import csv
+import sys
+from pathlib import Path
+from typing import Any
+
+from .colors import log
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Column name mappings  (human-readable header → API field name)
+# Both the friendly names and the raw API names are accepted.
+# ─────────────────────────────────────────────────────────────────────────────
+
+_SMB_COL_MAP: dict[str, str] = {
+    "share name":               "name",
+    "path":                     "path",
+    "description":              "comment",
+    "purpose":                  "purpose",
+    "read only":                "ro",
+    "browsable":                "browsable",
+    "guest access":             "guestok",
+    "access-based enumeration": "abe",
+    "hosts allow":              "hostsallow",
+    "hosts deny":               "hostsdeny",
+    "time machine":             "timemachine",
+    "enabled":                  "enabled",
+}
+
+_NFS_COL_MAP: dict[str, str] = {
+    "path":             "path",
+    "description":      "comment",
+    "read only":        "ro",
+    "map root user":    "maproot_user",
+    "map root group":   "maproot_group",
+    "map all user":     "mapall_user",
+    "map all group":    "mapall_group",
+    "security":         "security",
+    "allowed hosts":    "hosts",
+    "allowed networks": "networks",
+    "enabled":          "enabled",
+}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Column type metadata  (keyed by API field name)
+# ─────────────────────────────────────────────────────────────────────────────
+
+# Columns coerced to bool
+_SMB_BOOL_COLS = frozenset({"ro", "browsable", "guestok", "abe", "timemachine", "enabled"})
+# Columns coerced to list[str] (space-or-comma-separated in CSV)
+_SMB_LIST_COLS = frozenset({"hostsallow", "hostsdeny"})
+_SMB_REQUIRED  = frozenset({"name", "path"})
+
+_NFS_BOOL_COLS = frozenset({"ro", "enabled"})
+_NFS_LIST_COLS = frozenset({"security", "hosts", "networks"})
+_NFS_REQUIRED  = frozenset({"path"})
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Internal helpers
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _parse_bool(value: str, col: str, row_num: int) -> bool:
+    v = value.strip().lower()
+    if v in ("true", "yes", "1"):
+        return True
+    if v in ("false", "no", "0", ""):
+        return False
+    log.warning("  row %d: unrecognised boolean %r for column %r – treating as False",
+                row_num, value, col)
+    return False
+
+
+def _parse_list(value: str) -> list[str]:
+    """Split space-or-comma-separated value into a list, dropping blanks."""
+    return [p for p in value.replace(",", " ").split() if p]
+
+
+def _coerce_row(
+    row: dict[str, str],
+    bool_cols: frozenset[str],
+    list_cols: frozenset[str],
+    required: frozenset[str],
+    row_num: int,
+) -> dict[str, Any] | None:
+    """Validate and type-coerce one CSV row.  Returns None to skip the row."""
+    if not any((v or "").strip() for v in row.values()):
+        return None  # blank row
+
+    first_val = next(iter(row.values()), "") or ""
+    if first_val.strip().startswith("#"):
+        return None  # comment row
+
+    result: dict[str, Any] = {}
+    for col, raw in row.items():
+        if col is None:
+            continue
+        col = col.strip()
+        val = (raw or "").strip()
+
+        if not val:
+            continue  # omit empty optional fields; API uses its defaults
+
+        if col in bool_cols:
+            result[col] = _parse_bool(val, col, row_num)
+        elif col in list_cols:
+            result[col] = _parse_list(val)
+        else:
+            result[col] = val
+
+    for req in required:
+        if req not in result:
+            log.warning("  row %d: missing required field %r – skipping row", row_num, req)
+            return None
+
+    return result
+
+
+def _normalize_col(col: str, col_map: dict[str, str]) -> str:
+    """Map a header name to its API field name; falls back to the lowercased original."""
+    key = col.strip().lower()
+    return col_map.get(key, key)
+
+
+def _parse_csv(
+    csv_path: str,
+    bool_cols: frozenset[str],
+    list_cols: frozenset[str],
+    required: frozenset[str],
+    col_map: dict[str, str],
+    label: str,
+) -> list[dict]:
+    path = Path(csv_path)
+    if not path.is_file():
+        log.error("%s CSV file not found: %s", label, csv_path)
+        sys.exit(1)
+
+    shares: list[dict] = []
+    try:
+        with path.open(newline="", encoding="utf-8-sig") as fh:
+            reader = csv.DictReader(fh)
+            if reader.fieldnames is None:
+                log.error("%s CSV has no header row: %s", label, csv_path)
+                sys.exit(1)
+
+            # Normalise header names using the column map
+            normalised_header = {
+                _normalize_col(c, col_map)
+                for c in reader.fieldnames if c is not None
+            }
+            missing_req = required - normalised_header
+            if missing_req:
+                log.error(
+                    "%s CSV is missing required column(s): %s",
+                    label, ", ".join(sorted(missing_req)),
+                )
+                sys.exit(1)
+
+            for row_num, row in enumerate(reader, start=2):
+                normalised = {
+                    _normalize_col(k, col_map): v
+                    for k, v in row.items() if k is not None
+                }
+                share = _coerce_row(normalised, bool_cols, list_cols, required, row_num)
+                if share is not None:
+                    shares.append(share)
+
+    except OSError as exc:
+        log.error("Cannot read %s CSV: %s", label, exc)
+        sys.exit(1)
+
+    log.info("  %-14s  →  %s (%d share(s))", label.lower() + "_shares", csv_path, len(shares))
+    return shares
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Public API
+# ─────────────────────────────────────────────────────────────────────────────
+
+def parse_smb_csv(csv_path: str) -> list[dict]:
+    """Parse an SMB shares CSV.  Returns share dicts compatible with migrate.py."""
+    return _parse_csv(csv_path, _SMB_BOOL_COLS, _SMB_LIST_COLS, _SMB_REQUIRED, _SMB_COL_MAP, "SMB")
+
+
+def parse_nfs_csv(csv_path: str) -> list[dict]:
+    """Parse an NFS shares CSV.  Returns share dicts compatible with migrate.py."""
+    return _parse_csv(csv_path, _NFS_BOOL_COLS, _NFS_LIST_COLS, _NFS_REQUIRED, _NFS_COL_MAP, "NFS")
+
+
+def parse_csv_sources(smb_csv: str | None, nfs_csv: str | None) -> dict[str, Any]:
+    """
+    Parse one or both CSV files.
+    Returns {"smb_shares": list, "nfs_shares": list} — same shape as parse_archive().
+    """
+    log.info("Loading shares from CSV source(s).")
+    result: dict[str, Any] = {"smb_shares": [], "nfs_shares": []}
+    if smb_csv:
+        result["smb_shares"] = parse_smb_csv(smb_csv)
+    if nfs_csv:
+        result["nfs_shares"] = parse_nfs_csv(nfs_csv)
+    log.info(
+        "Loaded: %d SMB share(s), %d NFS share(s)",
+        len(result["smb_shares"]),
+        len(result["nfs_shares"]),
+    )
+    return result

truenas_migrate/migrate.py

@@ -14,7 +14,7 @@ from .summary import Summary
 # ─────────────────────────────────────────────────────────────────────────────
 
 # Read-only / server-generated fields that must NOT be sent on create/update
-_SMB_SHARE_READONLY = frozenset({"id", "locked"})
+_SMB_SHARE_READONLY = frozenset({"id", "locked", "path_local"})
 
 # CORE SMB share fields that do not exist in the SCALE API
 _SMB_SHARE_CORE_EXTRAS = frozenset({
@@ -152,3 +152,436 @@ async def migrate_nfs_shares(
             log.error("   %s: %s", _bold_red("FAILED"), exc)
             summary.nfs_failed += 1
             summary.errors.append(f"NFS share {path!r}: {exc}")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# iSCSI payload builders
+# ─────────────────────────────────────────────────────────────────────────────
+
+_ISCSI_EXTENT_READONLY    = frozenset({"id", "serial", "naa", "vendor", "locked"})
+_ISCSI_INITIATOR_READONLY = frozenset({"id"})
+_ISCSI_PORTAL_READONLY    = frozenset({"id", "tag"})
+_ISCSI_TARGET_READONLY    = frozenset({"id", "rel_tgt_id", "iscsi_parameters"})
+
+
+def _iscsi_extent_payload(extent: dict) -> dict:
+    payload = {k: v for k, v in extent.items() if k not in _ISCSI_EXTENT_READONLY}
+    if extent.get("type") == "DISK":
+        payload.pop("path", None)      # derived from disk on DISK extents
+        payload.pop("filesize", None)  # only meaningful for FILE extents
+    else:
+        payload.pop("disk", None)
+    return payload
+
+
+def _iscsi_initiator_payload(initiator: dict) -> dict:
+    return {k: v for k, v in initiator.items() if k not in _ISCSI_INITIATOR_READONLY}
+
+
+def _iscsi_portal_payload(portal: dict) -> dict:
+    payload = {k: v for k, v in portal.items() if k not in _ISCSI_PORTAL_READONLY}
+    # The API only accepts {"ip": "..."} in listen entries — port is a global setting
+    payload["listen"] = [{"ip": l["ip"]} for l in payload.get("listen", [])]
+    return payload
+
+
+def _iscsi_target_payload(
+    target: dict,
+    portal_id_map: dict[int, int],
+    initiator_id_map: dict[int, int],
+) -> dict:
+    payload = {k: v for k, v in target.items() if k not in _ISCSI_TARGET_READONLY}
+    payload["groups"] = [
+        {**g,
+         "portal":    portal_id_map.get(g["portal"], g["portal"]),
+         "initiator": initiator_id_map.get(g.get("initiator"), g.get("initiator"))}
+        for g in target.get("groups", [])
+    ]
+    return payload
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# iSCSI migration sub-routines
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def _migrate_iscsi_extents(
+    client: TrueNASClient,
+    extents: list[dict],
+    dry_run: bool,
+    summary: Summary,
+    id_map: dict[int, int],
+) -> None:
+    log.info("Querying existing iSCSI extents on destination …")
+    try:
+        existing = await client.call("iscsi.extent.query") or []
+    except RuntimeError as exc:
+        msg = f"Could not query iSCSI extents: {exc}"
+        log.error(msg); summary.errors.append(msg); return
+
+    existing_by_name = {e["name"].lower(): e for e in existing}
+    log.info("  Destination has %d existing extent(s).", len(existing_by_name))
+
+    for ext in extents:
+        name = ext.get("name", "<unnamed>")
+        log.info("%s iSCSI extent %s", _bold("──"), _bold_cyan(repr(name)))
+
+        if name.lower() in existing_by_name:
+            log.info("   %s – already exists on destination.", _yellow("SKIP"))
+            id_map[ext["id"]] = existing_by_name[name.lower()]["id"]
+            summary.iscsi_extents_skipped += 1
+            continue
+
+        payload = _iscsi_extent_payload(ext)
+        log.debug("   payload: %s", json.dumps(payload))
+
+        if dry_run:
+            log.info("   %s would create extent %s → %s",
+                     _cyan("[DRY RUN]"), _bold_cyan(repr(name)),
+                     ext.get("disk") or ext.get("path"))
+            summary.iscsi_extents_created += 1
+            id_map[ext["id"]] = ext["id"]  # placeholder — enables downstream dry-run remapping
+            if ext.get("type") == "DISK" and ext.get("disk"):
+                summary.zvols_to_check.append(ext["disk"].removeprefix("zvol/"))
+            continue
+
+        try:
+            r = await client.call("iscsi.extent.create", [payload])
+            log.info("   %s  id=%s", _bold_green("CREATED"), r.get("id"))
+            id_map[ext["id"]] = r["id"]
+            summary.iscsi_extents_created += 1
+        except RuntimeError as exc:
+            log.error("   %s: %s", _bold_red("FAILED"), exc)
+            summary.iscsi_extents_failed += 1
+            summary.errors.append(f"iSCSI extent {name!r}: {exc}")
+
+
+async def _migrate_iscsi_initiators(
+    client: TrueNASClient,
+    initiators: list[dict],
+    dry_run: bool,
+    summary: Summary,
+    id_map: dict[int, int],
+) -> None:
+    log.info("Querying existing iSCSI initiator groups on destination …")
+    try:
+        existing = await client.call("iscsi.initiator.query") or []
+    except RuntimeError as exc:
+        msg = f"Could not query iSCSI initiators: {exc}"
+        log.error(msg); summary.errors.append(msg); return
+
+    existing_by_comment = {e["comment"].lower(): e for e in existing if e.get("comment")}
+    log.info("  Destination has %d existing initiator group(s).", len(existing))
+
+    for init in initiators:
+        comment = init.get("comment", "")
+        log.info("%s iSCSI initiator group %s", _bold("──"), _bold_cyan(repr(comment)))
+
+        if comment and comment.lower() in existing_by_comment:
+            log.info("   %s – comment already exists on destination.", _yellow("SKIP"))
+            id_map[init["id"]] = existing_by_comment[comment.lower()]["id"]
+            summary.iscsi_initiators_skipped += 1
+            continue
+
+        payload = _iscsi_initiator_payload(init)
+        log.debug("   payload: %s", json.dumps(payload))
+
+        if dry_run:
+            log.info("   %s would create initiator group %s",
+                     _cyan("[DRY RUN]"), _bold_cyan(repr(comment)))
+            summary.iscsi_initiators_created += 1
+            id_map[init["id"]] = init["id"]  # placeholder — enables downstream dry-run remapping
+            continue
+
+        try:
+            r = await client.call("iscsi.initiator.create", [payload])
+            log.info("   %s  id=%s", _bold_green("CREATED"), r.get("id"))
+            id_map[init["id"]] = r["id"]
+            summary.iscsi_initiators_created += 1
+        except RuntimeError as exc:
+            log.error("   %s: %s", _bold_red("FAILED"), exc)
+            summary.iscsi_initiators_failed += 1
+            summary.errors.append(f"iSCSI initiator {comment!r}: {exc}")
+
+
+async def _migrate_iscsi_portals(
+    client: TrueNASClient,
+    portals: list[dict],
+    dry_run: bool,
+    summary: Summary,
+    id_map: dict[int, int],
+) -> None:
+    log.info("Querying existing iSCSI portals on destination …")
+    try:
+        existing = await client.call("iscsi.portal.query") or []
+    except RuntimeError as exc:
+        msg = f"Could not query iSCSI portals: {exc}"
+        log.error(msg); summary.errors.append(msg); return
+
+    def _ip_set(p: dict) -> frozenset:
+        return frozenset(l["ip"] for l in p.get("listen", []))
+
+    existing_ip_sets = [(_ip_set(p), p["id"]) for p in existing]
+    log.info("  Destination has %d existing portal(s).", len(existing))
+
+    for portal in portals:
+        comment = portal.get("comment", "")
+        ips = ", ".join(l['ip'] for l in portal.get("listen", []))
+        log.info("%s iSCSI portal %s [%s]", _bold("──"), _bold_cyan(repr(comment)), ips)
+
+        my_ips = _ip_set(portal)
+        match = next((eid for eips, eid in existing_ip_sets if eips == my_ips), None)
+        if match is not None:
+            log.info("   %s – IP set already exists on destination.", _yellow("SKIP"))
+            id_map[portal["id"]] = match
+            summary.iscsi_portals_skipped += 1
+            continue
+
+        payload = _iscsi_portal_payload(portal)
+        log.debug("   payload: %s", json.dumps(payload))
+
+        if dry_run:
+            log.info("   %s would create portal %s → %s",
+                     _cyan("[DRY RUN]"), _bold_cyan(repr(comment)), ips)
+            summary.iscsi_portals_created += 1
+            id_map[portal["id"]] = portal["id"]  # placeholder — enables downstream dry-run remapping
+            continue
+
+        try:
+            r = await client.call("iscsi.portal.create", [payload])
+            log.info("   %s  id=%s", _bold_green("CREATED"), r.get("id"))
+            id_map[portal["id"]] = r["id"]
+            summary.iscsi_portals_created += 1
+        except RuntimeError as exc:
+            log.error("   %s: %s", _bold_red("FAILED"), exc)
+            summary.iscsi_portals_failed += 1
+            summary.errors.append(f"iSCSI portal {comment!r}: {exc}")
+
+
+async def _migrate_iscsi_targets(
+    client: TrueNASClient,
+    targets: list[dict],
+    dry_run: bool,
+    summary: Summary,
+    id_map: dict[int, int],
+    portal_id_map: dict[int, int],
+    initiator_id_map: dict[int, int],
+) -> None:
+    log.info("Querying existing iSCSI targets on destination …")
+    try:
+        existing = await client.call("iscsi.target.query") or []
+    except RuntimeError as exc:
+        msg = f"Could not query iSCSI targets: {exc}"
+        log.error(msg); summary.errors.append(msg); return
+
+    existing_by_name = {t["name"].lower(): t for t in existing}
+    log.info("  Destination has %d existing target(s).", len(existing_by_name))
+
+    for target in targets:
+        name = target.get("name", "<unnamed>")
+        log.info("%s iSCSI target %s", _bold("──"), _bold_cyan(repr(name)))
+
+        if name.lower() in existing_by_name:
+            log.info("   %s – already exists on destination.", _yellow("SKIP"))
+            id_map[target["id"]] = existing_by_name[name.lower()]["id"]
+            summary.iscsi_targets_skipped += 1
+            continue
+
+        # Filter out groups whose portal or initiator could not be mapped (e.g. portal
+        # creation failed).  Warn per dropped group but still create the target — a
+        # target without every portal group is valid and preferable to no target at all.
+        valid_groups = []
+        for g in target.get("groups", []):
+            unmapped = []
+            if g.get("portal") not in portal_id_map:
+                unmapped.append(f"portal id={g['portal']}")
+            if g.get("initiator") not in initiator_id_map:
+                unmapped.append(f"initiator id={g['initiator']}")
+            if unmapped:
+                log.warning("   %s dropping group with unmapped %s",
+                            _yellow("WARN"), ", ".join(unmapped))
+            else:
+                valid_groups.append(g)
+
+        payload = _iscsi_target_payload({**target, "groups": valid_groups},
+                                        portal_id_map, initiator_id_map)
+        log.debug("   payload: %s", json.dumps(payload))
+
+        if dry_run:
+            log.info("   %s would create target %s",
+                     _cyan("[DRY RUN]"), _bold_cyan(repr(name)))
+            summary.iscsi_targets_created += 1
+            id_map[target["id"]] = target["id"]  # placeholder — enables downstream dry-run remapping
+            continue
+
+        try:
+            r = await client.call("iscsi.target.create", [payload])
+            log.info("   %s  id=%s", _bold_green("CREATED"), r.get("id"))
+            id_map[target["id"]] = r["id"]
+            summary.iscsi_targets_created += 1
+        except RuntimeError as exc:
+            log.error("   %s: %s", _bold_red("FAILED"), exc)
+            summary.iscsi_targets_failed += 1
+            summary.errors.append(f"iSCSI target {name!r}: {exc}")
+
+
+async def _migrate_iscsi_targetextents(
+    client: TrueNASClient,
+    targetextents: list[dict],
+    dry_run: bool,
+    summary: Summary,
+    target_id_map: dict[int, int],
+    extent_id_map: dict[int, int],
+) -> None:
+    log.info("Querying existing iSCSI target-extent associations on destination …")
+    try:
+        existing = await client.call("iscsi.targetextent.query") or []
+    except RuntimeError as exc:
+        msg = f"Could not query iSCSI target-extents: {exc}"
+        log.error(msg); summary.errors.append(msg); return
+
+    existing_keys = {(te["target"], te["lunid"]) for te in existing}
+    log.info("  Destination has %d existing association(s).", len(existing))
+
+    for te in targetextents:
+        src_tid = te["target"]
+        src_eid = te["extent"]
+        lunid   = te["lunid"]
+        dest_tid = target_id_map.get(src_tid)
+        dest_eid = extent_id_map.get(src_eid)
+
+        if dest_tid is None or dest_eid is None:
+            missing = []
+            if dest_tid is None: missing.append(f"target id={src_tid}")
+            if dest_eid is None: missing.append(f"extent id={src_eid}")
+            msg = f"iSCSI target-extent (lun {lunid}): cannot remap {', '.join(missing)}"
+            log.error("   %s", msg)
+            summary.iscsi_targetextents_failed += 1
+            summary.errors.append(msg)
+            continue
+
+        log.info("%s iSCSI target↔extent  target=%s  lun=%s  extent=%s",
+                 _bold("──"), dest_tid, lunid, dest_eid)
+
+        if (dest_tid, lunid) in existing_keys:
+            log.info("   %s – target+LUN already assigned on destination.", _yellow("SKIP"))
+            summary.iscsi_targetextents_skipped += 1
+            continue
+
+        payload = {"target": dest_tid, "lunid": lunid, "extent": dest_eid}
+        log.debug("   payload: %s", json.dumps(payload))
+
+        if dry_run:
+            log.info("   %s would associate target=%s  lun=%s  extent=%s",
+                     _cyan("[DRY RUN]"), dest_tid, lunid, dest_eid)
+            summary.iscsi_targetextents_created += 1
+            continue
+
+        try:
+            r = await client.call("iscsi.targetextent.create", [payload])
+            log.info("   %s  id=%s", _bold_green("CREATED"), r.get("id"))
+            summary.iscsi_targetextents_created += 1
+        except RuntimeError as exc:
+            log.error("   %s: %s", _bold_red("FAILED"), exc)
+            summary.iscsi_targetextents_failed += 1
+            summary.errors.append(
+                f"iSCSI target-extent (target={dest_tid}, lun={lunid}): {exc}")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# iSCSI pre-migration utilities
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def query_existing_iscsi(client: TrueNASClient) -> dict:
+    """
+    Query all iSCSI object counts from the destination.
+    Returns a dict with keys: extents, initiators, portals, targets, targetextents
+    Each value is a list of objects (may be empty).
+    """
+    result = {}
+    for key, method in [
+        ("extents",       "iscsi.extent.query"),
+        ("initiators",    "iscsi.initiator.query"),
+        ("portals",       "iscsi.portal.query"),
+        ("targets",       "iscsi.target.query"),
+        ("targetextents", "iscsi.targetextent.query"),
+    ]:
+        try:
+            result[key] = await client.call(method) or []
+        except RuntimeError:
+            result[key] = []
+    return result
+
+
+async def clear_iscsi_config(client: TrueNASClient) -> None:
+    """
+    Delete all iSCSI configuration from the destination in safe dependency order:
+    target-extents → targets → portals → initiators → extents.
+    """
+    for method_query, method_delete, label in [
+        ("iscsi.targetextent.query", "iscsi.targetextent.delete", "target-extent"),
+        ("iscsi.target.query",       "iscsi.target.delete",       "target"),
+        ("iscsi.portal.query",       "iscsi.portal.delete",       "portal"),
+        ("iscsi.initiator.query",    "iscsi.initiator.delete",    "initiator"),
+        ("iscsi.extent.query",       "iscsi.extent.delete",       "extent"),
+    ]:
+        try:
+            objects = await client.call(method_query) or []
+        except RuntimeError as exc:
+            log.warning("  Could not query iSCSI %ss: %s", label, exc)
+            continue
+        for obj in objects:
+            try:
+                await client.call(method_delete, [obj["id"]])
+                log.info("  Deleted iSCSI %s id=%s", label, obj["id"])
+            except RuntimeError as exc:
+                log.warning("  Failed to delete iSCSI %s id=%s: %s", label, obj["id"], exc)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Public iSCSI entry point
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def migrate_iscsi(
+    client: TrueNASClient,
+    iscsi: dict,
+    dry_run: bool,
+    summary: Summary,
+) -> None:
+    if not iscsi:
+        log.info("No iSCSI configuration found in archive.")
+        return
+
+    portals       = iscsi.get("portals", [])
+    initiators    = iscsi.get("initiators", [])
+    targets       = iscsi.get("targets", [])
+    extents       = iscsi.get("extents", [])
+    targetextents = iscsi.get("targetextents", [])
+
+    summary.iscsi_extents_found       = len(extents)
+    summary.iscsi_initiators_found    = len(initiators)
+    summary.iscsi_portals_found       = len(portals)
+    summary.iscsi_targets_found       = len(targets)
+    summary.iscsi_targetextents_found = len(targetextents)
+
+    gc = iscsi.get("global_config", {})
+    if gc.get("basename"):
+        log.info("  Source iSCSI basename: %s  (destination keeps its own)", gc["basename"])
+
+    if not any([portals, initiators, targets, extents, targetextents]):
+        log.info("iSCSI configuration is empty – nothing to migrate.")
+        return
+
+    extent_id_map:    dict[int, int] = {}
+    initiator_id_map: dict[int, int] = {}
+    portal_id_map:    dict[int, int] = {}
+    target_id_map:    dict[int, int] = {}
+
+    # Dependency order: extents and initiators first (no deps), then portals,
+    # then targets (need portal + initiator maps), then target-extent links.
+    await _migrate_iscsi_extents(client, extents, dry_run, summary, extent_id_map)
+    await _migrate_iscsi_initiators(client, initiators, dry_run, summary, initiator_id_map)
+    await _migrate_iscsi_portals(client, portals, dry_run, summary, portal_id_map)
+    await _migrate_iscsi_targets(
+        client, targets, dry_run, summary, target_id_map, portal_id_map, initiator_id_map)
+    await _migrate_iscsi_targetextents(
+        client, targetextents, dry_run, summary, target_id_map, extent_id_map)

truenas_migrate/summary.py

@@ -21,12 +21,47 @@ class Summary:
     nfs_skipped: int = 0
     nfs_failed:  int = 0
 
+    iscsi_extents_found:       int = 0
+    iscsi_extents_created:     int = 0
+    iscsi_extents_skipped:     int = 0
+    iscsi_extents_failed:      int = 0
+
+    iscsi_initiators_found:    int = 0
+    iscsi_initiators_created:  int = 0
+    iscsi_initiators_skipped:  int = 0
+    iscsi_initiators_failed:   int = 0
+
+    iscsi_portals_found:       int = 0
+    iscsi_portals_created:     int = 0
+    iscsi_portals_skipped:     int = 0
+    iscsi_portals_failed:      int = 0
+
+    iscsi_targets_found:       int = 0
+    iscsi_targets_created:     int = 0
+    iscsi_targets_skipped:     int = 0
+    iscsi_targets_failed:      int = 0
+
+    iscsi_targetextents_found:   int = 0
+    iscsi_targetextents_created: int = 0
+    iscsi_targetextents_skipped: int = 0
+    iscsi_targetextents_failed:  int = 0
+
     errors: list[str] = field(default_factory=list)
 
     # Populated during dry-run dataset safety checks
     paths_to_create:  list[str] = field(default_factory=list)
     missing_datasets: list[str] = field(default_factory=list)
 
+    # Populated during iSCSI dry-run zvol safety checks
+    zvols_to_check:  list[str] = field(default_factory=list)
+    missing_zvols:   list[str] = field(default_factory=list)
+
+    @property
+    def _has_iscsi(self) -> bool:
+        return (self.iscsi_extents_found + self.iscsi_initiators_found +
+                self.iscsi_portals_found + self.iscsi_targets_found +
+                self.iscsi_targetextents_found) > 0
+
     def report(self) -> str:
         w = 60
 
@@ -34,6 +69,14 @@ class Summary:
             s = f"{label}={n}"
             return color_fn(s) if n > 0 else _dim(s)
 
+        def _iscsi_val(found, created, skipped, failed) -> str:
+            return (
+                f"{_dim('found=' + str(found))}  "
+                f"{_stat('created', created, _bold_green)}  "
+                f"{_stat('skipped', skipped, _yellow)}  "
+                f"{_stat('failed',  failed,  _bold_red)}"
+            )
+
         smb_val = (
             f"{_dim('found=' + str(self.smb_found))}  "
             f"{_stat('created', self.smb_created, _bold_green)}  "
@@ -67,11 +110,30 @@ class Summary:
             f"{tl}{hr}{tr}",
             title_row,
             f"{ml}{hr}{mr}",
-            row("SMB shares :  ", smb_val),
-            row("NFS shares :  ", nfs_val),
-            f"{bl}{hr}{br}",
+            row("SMB shares    :  ", smb_val),
+            row("NFS shares    :  ", nfs_val),
         ]
 
+        if self._has_iscsi:
+            lines.append(f"{ml}{hr}{mr}")
+            lines.append(row("iSCSI extents :  ", _iscsi_val(
+                self.iscsi_extents_found, self.iscsi_extents_created,
+                self.iscsi_extents_skipped, self.iscsi_extents_failed)))
+            lines.append(row("iSCSI initiators: ", _iscsi_val(
+                self.iscsi_initiators_found, self.iscsi_initiators_created,
+                self.iscsi_initiators_skipped, self.iscsi_initiators_failed)))
+            lines.append(row("iSCSI portals :  ", _iscsi_val(
+                self.iscsi_portals_found, self.iscsi_portals_created,
+                self.iscsi_portals_skipped, self.iscsi_portals_failed)))
+            lines.append(row("iSCSI targets :  ", _iscsi_val(
+                self.iscsi_targets_found, self.iscsi_targets_created,
+                self.iscsi_targets_skipped, self.iscsi_targets_failed)))
+            lines.append(row("iSCSI tgt↔ext :  ", _iscsi_val(
+                self.iscsi_targetextents_found, self.iscsi_targetextents_created,
+                self.iscsi_targetextents_skipped, self.iscsi_targetextents_failed)))
+
+        lines.append(f"{bl}{hr}{br}")
+
         if self.errors:
             lines.append(f"\n  {_bold_red(str(len(self.errors)) + ' error(s):')} ")
             for e in self.errors:
@@ -89,5 +151,16 @@ class Summary:
                 "  These paths must exist before shares can be created.\n"
                 "  Use interactive mode or answer 'y' at the dataset prompt to create them."
             )
+        if self.missing_zvols:
+            lines.append(
+                f"\n  {_bold_yellow('WARNING:')} "
+                f"{len(self.missing_zvols)} zvol(s) do not exist on the destination:"
+            )
+            for z in self.missing_zvols:
+                lines.append(f"    {_yellow('•')} {z}")
+            lines.append(
+                "  These zvols must exist before iSCSI extents can be created.\n"
+                "  Use interactive mode to be prompted for size and auto-create them."
+            )
         lines.append("")
         return "\n".join(lines)