abCROWN

The CTRAIN abCROWN integration exports the model to ONNX, writes per-instance VNNLIB specifications, and records complete-verification results as JSON. The evaluation API supports resumable runs via warm_start and chunked verification with start_idx/end_idx; see CTRAIN.eval.eval_complete_abcrown and CTRAIN.model_wrappers.CTRAINWrapper.evaluate_complete.

The vendored verifier is updated to abCROWN commit 6b8bbcfac1c01da1cabd240a87e4dce1a65f5a2b and uses the matching newer default configuration schema. Run complete verification in an environment with the matching auto_LiRPA package installed; the training-experiments venv currently provides auto_LiRPA 0.7.0.

abcrown_eval(config, seed, instance, vnnlib_path='../../vnnlib/', model_name='mnist_6_100', model_path='./abCROWN/complete_verifier/models/eran/mnist_6_100_nat.pth', model_onnx_path=None, input_shape=[-1, 1, 28, 28], timeout=600, no_cores=28, par_factor=10)

Runs the abCROWN verification process with the given configuration. abCROWN is invoked from inside the program code, so a crash/freeze can only be handled partially.

Parameters:

Name	Type	Description	Default
config	dict	Configuration dictionary for the verification process.	required
seed	int	Seed for random number generation.	required
instance	str	Path to the VNN-LIB instance file.	required
vnnlib_path	str	Path prefix for VNN-LIB files. Defaults to '../../vnnlib/'.	'../../vnnlib/'
model_name	str	Name of the model to be verified. Defaults to 'mnist_6_100'.	'mnist_6_100'
model_path	str	Path to the model file. Defaults to './abCROWN/complete_verifier/models/eran/mnist_6_100_nat.pth'.	'./abCROWN/complete_verifier/models/eran/mnist_6_100_nat.pth'
model_onnx_path	str	Path to the ONNX model file. Defaults to None.	None
input_shape	list	Shape of the input tensor. Defaults to [-1, 1, 28, 28].	[-1, 1, 28, 28]
timeout	int	Timeout for the verification process in seconds. Defaults to 600.	600
no_cores	int	Number of CPU cores to use for parallel solvers, when abCROWN is configured to use MIP Solvers. Defaults to 28.	28
par_factor	int	Penalty factor for running time in case of timeout. Defaults to 10.	10

Returns:

Type	Description
tuple	Running time of the verification process and the result of the verification (sat/unsat or timeout/unknown).

Source code in CTRAIN/complete_verification/abCROWN/verify.py

def abcrown_eval(config, seed, instance, vnnlib_path='../../vnnlib/', model_name='mnist_6_100', model_path='./abCROWN/complete_verifier/models/eran/mnist_6_100_nat.pth', model_onnx_path=None, input_shape=[-1, 1, 28, 28], timeout=600, no_cores=28, par_factor=10):
    """
    Runs the abCROWN verification process with the given configuration.
    abCROWN is invoked from inside the program code, so a crash/freeze can only be handled partially.

    Args:
        config (dict): Configuration dictionary for the verification process.
        seed (int): Seed for random number generation.
        instance (str): Path to the VNN-LIB instance file.
        vnnlib_path (str, optional): Path prefix for VNN-LIB files. Defaults to '../../vnnlib/'.
        model_name (str, optional): Name of the model to be verified. Defaults to 'mnist_6_100'.
        model_path (str, optional): Path to the model file. Defaults to './abCROWN/complete_verifier/models/eran/mnist_6_100_nat.pth'.
        model_onnx_path (str, optional): Path to the ONNX model file. Defaults to None.
        input_shape (list, optional): Shape of the input tensor. Defaults to [-1, 1, 28, 28].
        timeout (int, optional): Timeout for the verification process in seconds. Defaults to 600.
        no_cores (int, optional): Number of CPU cores to use for parallel solvers, when abCROWN is configured to use MIP Solvers. Defaults to 28.
        par_factor (int, optional): Penalty factor for running time in case of timeout. Defaults to 10.

    Returns:
        (tuple): Running time of the verification process and the result of the verification (sat/unsat or timeout/unknown).
    """
    print(config, seed, instance)
    std_conf = config

    device = 'cuda' if torch.cuda.is_available() else 'mps' if torch.backends.mps.is_available() else 'cpu'

    timestamp = time.time()

    std_conf['model']['name'] = model_name
    std_conf['model']['path'] = f'/tmp/{model_name}.pth' if model_name is not None else None
    std_conf['model']['onnx_path'] = model_onnx_path if model_onnx_path is not None else None
    std_conf['model']['input_shape'] = input_shape

    std_conf['general']['device'] = device

    std_conf['bab']['timeout'] = timeout

    if not std_conf['solver'].get('mip'):
        std_conf['solver']['mip'] = get_abcrown_standard_conf(timeout=timeout, no_cores=no_cores)['solver']['mip']
    std_conf['solver']['mip']['parallel_solvers'] = no_cores

    std_conf['specification']['vnnlib_path_prefix'] = vnnlib_path
    std_conf['specification']['vnnlib_path'] = instance
    std_conf['general']['output_file'] = f'/tmp/out_{timestamp}.pkl'
    std_conf['general']['save_output'] = True

    print(json.dumps(config, indent=2))

    with open(f"/tmp/conf_{timestamp}.yaml", "w", encoding='u8') as f:
        yaml.dump(std_conf, f)

    abcrown_instance = ABCROWN(
        ['--config', f'/tmp/conf_{timestamp}.yaml']
    )

    # Precompile VNN-LIB s.t. each run can access the cache
    _ = read_vnnlib(instance)

    start_time = time.time()
    try:
        verification_res = abcrown_instance.main()
    except Exception as e:
        print(type(e), e)
        print(traceback.format_exc())
        return MAX_LOSS, 'unknown'
    end_time = time.time()

    os.system(f'rm /tmp/conf_{timestamp}.yaml')

    with open(f'/tmp/out_{timestamp}.pkl', 'rb') as f:
        result_dict = pickle.load(f)

    result = result_dict['results']

    if result == 'unknown':
        print("PENALISING RUNNING TIME DUE TO TIMEOUT!")
        running_time = timeout * par_factor if timeout > (end_time - start_time) else (end_time - start_time) * par_factor
    else:
        running_time = end_time - start_time

    return running_time, result

limited_abcrown_eval(work_dir, runner_path='CTRAIN/complete_verification/abCROWN/runner.py', *args, **kwargs)

Executes the abCROWN verification using a specified verification process with a specified timeout and handles the results. This increases robustness, since a crash of abCROWN does not result in a crash of the python script. This function serializes the provided arguments and keyword arguments, runs a separate Python script to perform the verification, and handles the process execution including timeout management. The results are deserialized and returned if the process completes successfully within the timeout period.

Parameters:

Name	Type	Description	Default
work_dir	str	The working directory where temporary files will be stored.	required
runner_path	str	The path to the runner script that performs the verification. Defaults to 'src/complete_verification/abCROWN/runner.py'.	'CTRAIN/complete_verification/abCROWN/runner.py'
*args		Additional positional arguments to be passed to the runner script.	()
**kwargs		Additional keyword arguments to be passed to the runner script. Must include 'timeout' (float) which specifies the timeout period in seconds.	{}

Returns:

Type	Description
tuple	A tuple containing the running time and the result (sat/unsat or timeout/unknown) if the verification completes successfully. If the verification fails or times out, returns (MAX_LOSS, 'unknown').

Raises:

Type	Description
Exception	If there is an error running the process.

Source code in CTRAIN/complete_verification/abCROWN/verify.py

def limited_abcrown_eval(work_dir, runner_path='CTRAIN/complete_verification/abCROWN/runner.py', *args, **kwargs):
    """
    Executes the abCROWN verification using
    a specified verification process with a specified timeout and handles the results.
    This increases robustness, since a crash of abCROWN does not result in a crash of the
    python script.
    This function serializes the provided arguments and keyword arguments, runs a
    separate Python script to perform the verification, and handles the process
    execution including timeout management. The results are deserialized and returned
    if the process completes successfully within the timeout period.

    Args:
        work_dir (str): The working directory where temporary files will be stored.
        runner_path (str, optional): The path to the runner script that performs the
            verification. Defaults to 'src/complete_verification/abCROWN/runner.py'.
        *args: Additional positional arguments to be passed to the runner script.
        **kwargs: Additional keyword arguments to be passed to the runner script.
            Must include 'timeout' (float) which specifies the timeout period in seconds.

    Returns:
        (tuple): A tuple containing the running time and the result (sat/unsat or timeout/unknown) if the verification
                    completes successfully. If the verification fails or times out, returns
                    (MAX_LOSS, 'unknown').

    Raises:
        (Exception): If there is an error running the process.
    """
    outer_timeout = kwargs['timeout'] * 1.2

    timestamp = time.time()

    args_pkl_path = f'{work_dir}/args_{timestamp}.pkl'
    result_path = f"{work_dir}/result_{timestamp}.pkl"

    with open(f'{work_dir}/args_{timestamp}.pkl', "wb") as f:
        pickle.dump((args, kwargs), f)

    verification_ok = False

    runner_args = [args_pkl_path, result_path]

    try:
        print(f"Running {['python3', runner_path] + runner_args}")
        process = subprocess.Popen(
            ["python3", runner_path] + runner_args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        try:
            stdout, stderr = process.communicate(timeout=outer_timeout)
            print("Function finished successfully.")
            print("Output:", stdout.decode())
            print("Error Output:", stderr.decode())
            verification_ok = True

        except subprocess.TimeoutExpired:
            print(f"Function exceeded timeout of {outer_timeout} seconds. Terminating...")
            process.terminate()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                print("Function did not terminate after SIGTERM. Killing...")
                process.kill()

    except Exception as e:
        print(f"Error running the process: {e}")

    if verification_ok:
        with open(result_path, 'rb') as f:
            running_time, result = pickle.load(f)

        return running_time, result

    return MAX_LOSS, 'unknown'

get_abcrown_standard_conf(timeout=600, no_cores=28)

Generates the standard configuration for abCROWN. This function reads the standard configuration from a predefined string, updates the device setting based on the available hardware, and sets the timeout and number of parallel solvers.

Parameters:

Name	Type	Description	Default
timeout	int	The timeout value for the BAB solver in seconds. Defaults to 600.	600
no_cores	int	The number of parallel solvers to use. Defaults to 28.	28

Returns:

Type	Description
dict	The updated standard configuration dictionary.

Source code in CTRAIN/complete_verification/abCROWN/util.py

def get_abcrown_standard_conf(timeout=600, no_cores=28):
    """
    Generates the standard configuration for abCROWN.
    This function reads the standard configuration from a predefined string,
    updates the device setting based on the available hardware, and sets the
    timeout and number of parallel solvers.

    Args:
      timeout (int, optional): The timeout value for the BAB solver in seconds. Defaults to 600.
      no_cores (int, optional): The number of parallel solvers to use. Defaults to 28.

    Returns:
      (dict): The updated standard configuration dictionary.
    """
    f = StringIO(STD_CONF)

    std_conf = yaml.load(f, Loader=Loader)

    device = 'cuda' if torch.cuda.is_available() else 'mps' if torch.backends.mps.is_available() else 'cpu'

    std_conf['general']['device'] = device
    std_conf['bab']['timeout'] = timeout
    std_conf['solver']['mip']['parallel_solvers'] = no_cores

    return std_conf

instances_to_vnnlib(indices, data, vnnlib_path, experiment_name, eps, eps_temp, data_min, data_max, no_classes)

Converts a set of inputs and an epsilon value into VNN-LIB format files for adversarial robustness verification.

Parameters:

Name	Type	Description	Default
indices	list of int	List of indices of the data instances to be converted.	required
data	Dataset	Dataset containing the data instances.	required
vnnlib_path	str	Path where the VNN-LIB files will be saved.	required
experiment_name	str	Name of the experiment for documentation purposes.	required
eps	float	Epsilon value for the adversarial robustness property.	required
eps_temp	float	Temporary epsilon value used for domain calculation.	required
data_min	float	Minimum value for data normalization/clamping.	required
data_max	float	Maximum value for data normalization/clamping.	required
no_classes	int	Number of classes in the classification task.	required

Returns:

Type	Description
list of str	List of file paths to the generated VNN-LIB files.

Source code in CTRAIN/complete_verification/abCROWN/util.py

def instances_to_vnnlib(indices, data, vnnlib_path, experiment_name, eps, eps_temp, data_min, data_max, no_classes):
    """
    Converts a set of inputs and an epsilon value into VNN-LIB format files for adversarial robustness verification.

    Args:
      indices (list of int): List of indices of the data instances to be converted.
      data (torch.utils.data.Dataset): Dataset containing the data instances.
      vnnlib_path (str): Path where the VNN-LIB files will be saved.
      experiment_name (str): Name of the experiment for documentation purposes.
      eps (float): Epsilon value for the adversarial robustness property.
      eps_temp (float): Temporary epsilon value used for domain calculation.
      data_min (float): Minimum value for data normalization/clamping.
      data_max (float): Maximum value for data normalization/clamping.
      no_classes (int): Number of classes in the classification task.

    Returns:
      (list of str): List of file paths to the generated VNN-LIB files.
    """
    vnnlib_list_test = []
    for idx in indices:
        x, ground_truth = data[idx]
        ground_truth = torch.as_tensor(ground_truth).cpu().numpy()
        x = torch.as_tensor(x).cpu()
        eps_temp = torch.as_tensor(eps_temp).cpu()
        data_min = torch.as_tensor(data_min).cpu()
        data_max = torch.as_tensor(data_max).cpu()
        print(f'idx_{idx}_eps_{eps}')
        # Record the results: vnn-lib specification.
        filename = vnnlib_path + f"idx_{idx}-eps{eps}.vnnlib"
        comment = f"Adversarial robustness property for {experiment_name}. l_inf radius: {eps}, " \
                f"Test Image {idx}."
        comment = ' '.join(comment.splitlines()).strip()
        # ground_truth = np.argmax(ground_truth)
        domain = torch.stack(
            [(x - eps_temp).clamp(data_min, data_max),
            (x + eps_temp).clamp(data_min, data_max)], dim=-1)
        write_adversarial_robustness_vnnlib(filename, comment, domain, ground_truth, n_classes=no_classes)
        vnnlib_list_test.append(filename)

    return vnnlib_list_test

write_adversarial_robustness_vnnlib(filename, initial_comment, input_domain, ground_truth, n_classes=10)

Create a vnnlib specification for an adversarial robustness property.

The function writes the following to the specified file

• A comment at the top of the file.
• Declarations for input and output variables.
• Input constraints based on the provided input domain.
• Output constraints encoding the conditions for a property counter-example.

Parameters:

Name	Type	Description	Default
filename	str	The name of the file to write the vnnlib specification to.	required
initial_comment	str	A comment to include at the top of the vnnlib file.	required
input_domain	Tensor	A tensor representing the input domain, i.e. lower und upper bounds per input, with shape (n, 2), where n is the number of input variables.	required
ground_truth	int	The index of the ground truth class.	required
n_classes	int	The number of output classes. Default is 10.	10

Source code in CTRAIN/complete_verification/abCROWN/util.py

def write_adversarial_robustness_vnnlib(filename, initial_comment, input_domain, ground_truth, n_classes=10):
    """
    Create a vnnlib specification for an adversarial robustness property.

    The function writes the following to the specified file:
      - A comment at the top of the file.
      - Declarations for input and output variables.
      - Input constraints based on the provided input domain.
      - Output constraints encoding the conditions for a property counter-example.

    Parameters:
      filename (str): The name of the file to write the vnnlib specification to.
      initial_comment (str): A comment to include at the top of the vnnlib file.
      input_domain (torch.Tensor): A tensor representing the input domain, i.e. lower und upper bounds per input, with shape (n, 2), where n is the number of input variables.
      ground_truth (int): The index of the ground truth class.
      n_classes (int, optional): The number of output classes. Default is 10.
    """

    with open(filename, "w") as f:
        f.write(f"; {initial_comment}\n")

        # Declare input variables.
        f.write("\n")
        linearized_domain = input_domain.view(-1, 2)
        for i in range(linearized_domain.shape[0]):
            f.write(f"(declare-const X_{i} Real)\n")
        f.write("\n")

        # Declare output variables.
        for i in range(n_classes):
            f.write(f"(declare-const Y_{i} Real)\n")
        f.write("\n")

        # Define input constraints.
        f.write(f"; Input constraints:\n")
        for i in range(linearized_domain.shape[0]):
            f.write(f"(assert (<= X_{i} {linearized_domain[i, 1]}))\n")  # UB
            f.write(f"(assert (>= X_{i} {linearized_domain[i, 0]}))\n")  # LB
            f.write("\n")
        f.write("\n")

        # Define output constraints, providing an unnecessary "and" to ease parsing in vnn-comp-21.
        f.write(f"; Output constraints (encoding the conditions for a property counter-example):\n")
        f.write(f"(assert (or\n")
        for i in range(n_classes):
            if i != ground_truth:
                f.write(f"\t(and (<= Y_{ground_truth} Y_{i}))\n")
        f.write(f"))\n")
        f.write("\n")