Create and use a hot-start file

[lrntct]

It could be useful to have the possibility of stopping and restarting a simulation at an arbitrary time.
The goal of the hot-start file is not to create a portable simulation file, but rather to be able to restart in the same environment a simulation that has been stopped. Therefore, it will not include all input data files.

Creating the hotstart file

The hot-start file format would be a ZIP file containing:

• As JSON:
  • Creation time
  • itzi version
  • hotstart file version number
  • original provider types
  • SimulationConfig
  • DomainData (north, south, east, west, rows, cols, crs_wkt)
  • Time state: sim_time, dt of all models, next_ts, nextstep, time_steps_counters
  • Infiltration model state: cumulative infiltration (for Green-Ampt) or equivalent state for other models - It might be worth adding the cumulative infiltration array to RasterDomain instead of treating it separately
  • Accumulation tracking: accum_update_time, old_domain_volume
• RasterDomain, including mask with numpy.savez
• SWMM own hot-start file

Possible hotstart

hotstart_file:
  ├── metadata/
  │   ├── creation_time
  │   ├── itzi version
  │   ├── hotstart_version
  │   ├── original_provider_types
  │   └── SimulationConfig
  │   └── domain_data (north, south, east, west, rows, cols, crs_wkt)
  ├── time_state/
  │   ├── sim_time
  │   ├── dt
  │   ├── next_ts
  │   └── time_steps_counters
  ├── domain/
  │   ├── all raster_domain arrays
  │   └── arr_mask
  ├── accumulation_tracking/
  │   ├── accum_update_time
  │   └── old_domain_volume
  └── models/
      └── swmm_hotstart_path (if applicable)

This would require to:

• Add a function to create and save the hotstart file to disk
• Add a factory function to create a simulation from a hotstart file

Loading a hotstart file

Safely load JSON:

import json
import os

def safe_json_load(file_path, max_size=1024*1024):
    # Check file size first
    if os.path.getsize(file_path) > max_size:
        raise ValueError("File too large")
    
    with open(file_path, 'r') as file:
        return json.load(file)

What CAN change safely:

• Output providers (different output format is fine)
• Reporting frequency (record_step)
• End time (can extend simulation)
• Output map names

What CANNOT change:

• Grid dimensions/resolution/extent
• Infiltration model type
• Surface flow parameters that affect numerical scheme
• Presence/absence of drainage coupling

Steps for loading the hotstart:

1. Load DomainData from hotstart file
2. Get DomainData from new provider
3. Validate they match exactly (tolerance for floating point: ~1e-6)
4. Validate arr_mask matches
5. Check configuration compatibility
6. Only then proceed to restore simulation state

To do:

• Move Green-Ampt cumulative infiltration array to RasterDomain #185

[lrntct]

A snippet for safe loading of a tar archive. tar has the advantage over zip of allowing check of the packed files before extraction.

import tarfile
import os
import sys

def safe_extract_payload(archive_path, extract_to, 
                         max_metadata_size=10*1024*1024,
                         max_binary_size=500*1024*1024,
                         max_npz_size=1000*1024*1024):
    """
    Safely extract payload with known structure.
    Requires Python 3.12+ for secure 'data' filter.
    """
    if sys.version_info < (3, 12):
        raise RuntimeError("Python 3.12+ required for secure tar extraction")
    
    expected_files = {'metadata.json', 'data.bin', 'arrays.npz'}
    size_limits = {
        'metadata.json': max_metadata_size,
        'data.bin': max_binary_size,
        'arrays.npz': max_npz_size
    }
    
    os.makedirs(extract_to, exist_ok=True)
    found_files = set()
    total_size = 0
    
    with tarfile.open(archive_path, 'r:gz') as tar:
        for member in tar.getmembers():
            # Only allow regular files (filter='data' will handle this too, but explicit check)
            if not member.isfile():
                raise ValueError(f"Only regular files allowed: {member.name}")
            
            # Check it's an expected file (no subdirectories)
            if '/' in member.name or member.name not in expected_files:
                raise ValueError(f"Unexpected file or path: {member.name}")
            
            # Check file-specific size limit (zip bomb protection)
            if member.size > size_limits[member.name]:
                raise ValueError(f"File too large: {member.name} ({member.size} bytes)")
            
            total_size += member.size
            found_files.add(member.name)
        
        # Verify all expected files are present
        if found_files != expected_files:
            missing = expected_files - found_files
            extra = found_files - expected_files
            if missing:
                raise ValueError(f"Missing required files: {missing}")
            if extra:
                raise ValueError(f"Unexpected files: {extra}")
        
        # Check total extracted size
        max_total = sum(size_limits.values())
        if total_size > max_total:
            raise ValueError(f"Total archive size too large: {total_size} bytes")
        
        # All checks passed - extract with 'data' filter for security
        # This handles: path traversal, symlinks, absolute paths, device files, permissions
        tar.extractall(extract_to, filter='data')
    
    return {
        'metadata': os.path.join(extract_to, 'metadata.json'),
        'binary': os.path.join(extract_to, 'data.bin'),
        'arrays': os.path.join(extract_to, 'arrays.npz')
    }