pygip.models.attack package

Subpackages

• pygip.models.attack.mea package
  • Submodules
  • pygip.models.attack.mea.MEA module
    • ModelExtractionAttack0
      • ModelExtractionAttack0._abc_impl
      • ModelExtractionAttack0._attack_impl()
      • ModelExtractionAttack0.features
      • ModelExtractionAttack0.graph
      • ModelExtractionAttack0.labels
      • ModelExtractionAttack0.test_mask
      • ModelExtractionAttack0.train_mask
    • ModelExtractionAttack1
      • ModelExtractionAttack1._abc_impl
      • ModelExtractionAttack1._attack_impl()
      • ModelExtractionAttack1.features
      • ModelExtractionAttack1.graph
      • ModelExtractionAttack1.labels
      • ModelExtractionAttack1.test_mask
      • ModelExtractionAttack1.train_mask
    • ModelExtractionAttack2
      • ModelExtractionAttack2._abc_impl
      • ModelExtractionAttack2._attack_impl()
      • ModelExtractionAttack2.features
      • ModelExtractionAttack2.graph
      • ModelExtractionAttack2.labels
      • ModelExtractionAttack2.test_mask
      • ModelExtractionAttack2.train_mask
    • ModelExtractionAttack3
      • ModelExtractionAttack3._abc_impl
      • ModelExtractionAttack3._attack_impl()
      • ModelExtractionAttack3.features
      • ModelExtractionAttack3.graph
      • ModelExtractionAttack3.labels
      • ModelExtractionAttack3.test_mask
      • ModelExtractionAttack3.train_mask
    • ModelExtractionAttack4
      • ModelExtractionAttack4._abc_impl
      • ModelExtractionAttack4._attack_impl()
      • ModelExtractionAttack4.features
      • ModelExtractionAttack4.graph
      • ModelExtractionAttack4.labels
      • ModelExtractionAttack4.test_mask
      • ModelExtractionAttack4.train_mask
    • ModelExtractionAttack5
      • ModelExtractionAttack5._abc_impl
      • ModelExtractionAttack5._attack_impl()
      • ModelExtractionAttack5.features
      • ModelExtractionAttack5.graph
      • ModelExtractionAttack5.labels
      • ModelExtractionAttack5.test_mask
      • ModelExtractionAttack5.train_mask
    • _MEABase
      • _MEABase._abc_impl
      • _MEABase._attack_impl()
      • _MEABase._compute_metrics()
      • _MEABase._load_model()
      • _MEABase._query_target()
      • _MEABase._train_surrogate()
      • _MEABase._train_target_model()
      • _MEABase.attack()
      • _MEABase.features
      • _MEABase.graph
      • _MEABase.labels
      • _MEABase.supported_api_types
      • _MEABase.test_mask
      • _MEABase.train_mask
    • _as_tensor()
    • _safe_dir()
    • add_self_loops()
    • erdos_renyi_graph()
    • load_attack2_generated_graph()
    • load_attack3_shadow_indices()
    • random_shadow_indices()
    • subgraph_from_nodes()
  • Module contents

Submodules

pygip.models.attack.AdvMEA module

class pygip.models.attack.AdvMEA.AdvMEA(dataset, attack_node_fraction, model_path=None)

Bases: BaseAttack

_abc_impl = <_abc_data object>

_load_model(model_path)

Load a pre-trained model.

_to_cpu(tensor)

Safely move tensor to CPU for NumPy operations

_train_target_model()

Train the target model (GCN) on the original graph.

attack()

Execute the attack.

supported_api_types = {'dgl'}

pygip.models.attack.CEGA module

class pygip.models.attack.CEGA.CEGA(dataset, attack_node_fraction, model_path=None, attack_x_ratio=1.0, attack_a_ratio=1.0)

Bases: BaseAttack

_abc_impl = <_abc_data object>

attack(seed=1, cuda=None, LR=0.001, TGT_LR=0.01, EVAL_EPOCH=10, TGT_EPOCH=10, WARMUP_EPOCH=4, dropout=False, model_performance=True, **kwargs)

Returns:

• perf_json (dict) – Performance metrics (JSON-serialisable): accuracy/fidelity/F1 of the surrogate, and optionally target accuracy/F1 for reference.

• comp_json (dict) – Computation metrics (JSON-serialisable): attack_time, query_target_time, train_surrogate_time, etc.

supported_api_types = {'dgl'}

class pygip.models.attack.CEGA.GCN_drop(*args: Any, **kwargs: Any)

Bases: Module

forward(g, features)

class pygip.models.attack.CEGA.GcnNet(*args: Any, **kwargs: Any)

Bases: Module

forward(g, features, return_hidden=False)

pygip.models.attack.CEGA.accuracy(output, labels)

pygip.models.attack.CEGA.attack0(dataset_name, seed, cuda, attack_node_arg=0.25, file_path='', LR=0.001, TGT_LR=0.01, EVAL_EPOCH=1000, TGT_EPOCH=1000, WARMUP_EPOCH=400, dropout=False, model_performance=True, **kwargs)

pygip.models.attack.CEGA.aug_normalized_adjacency(adj)

pygip.models.attack.CEGA.aug_random_walk(adj)

pygip.models.attack.CEGA.calculate_entropy(probs)

pygip.models.attack.CEGA.compute_distance(_i, _j, features_aax)

pygip.models.attack.CEGA.compute_rank(tensor, element)

pygip.models.attack.CEGA.convert_pyg_to_dgl(pyg_data)

Converts a PyTorch Geometric Data object into a DGLGraph.

Parameters:

pyg_data (torch_geometric.data.Data) – PyTorch Geometric Data object.

Returns:

The converted DGL graph.

Return type:

dgl.DGLGraph

pygip.models.attack.CEGA.evaluate(model, g, features, labels, mask)

pygip.models.attack.CEGA.find_short_dist(embedding_pool, cluster_centers)

pygip.models.attack.CEGA.get_current_neighbors_1(cur_nodes, adj_matrix)

pygip.models.attack.CEGA.get_current_neighbors_dense(cur_nodes, adj_matrix2)

pygip.models.attack.CEGA.get_entropy_contribute(npy_m1, npy_m2)

pygip.models.attack.CEGA.get_max_info_entropy_node_set(idx_used, high_score_nodes, labels, batch_size, adj_matrix2, num_class, model_prediction)

pygip.models.attack.CEGA.get_max_nnd_node_dense(idx_used, high_score_nodes, min_distance, distance_aax, num_ones, num_node, adj_matrix2, gamma=1)

pygip.models.attack.CEGA.get_receptive_fields_dense(cur_neighbors, selected_node, weighted_score, adj_matrix2)

pygip.models.attack.CEGA.init_mask(C, sub_train_mask, sub_labels)

pygip.models.attack.CEGA.load_data(dataset_name)

pygip.models.attack.CEGA.load_data_from_grain(path='./data', dataset='cora')

ind.[:dataset].x => the feature vectors of the training instances (scipy.sparse.csr.csr_matrix) ind.[:dataset].y => the one-hot labels of the labeled training instances (numpy.ndarray) ind.[:dataset].allx => the feature vectors of both labeled and unlabeled training instances (csr_matrix) ind.[:dataset].ally => the labels for instances in ind.dataset_str.allx (numpy.ndarray) ind.[:dataset].graph => the dict in the format {index: [index of neighbor nodes]} (collections.defaultdict) ind.[:dataset].tx => the feature vectors of the test instances (scipy.sparse.csr.csr_matrix) ind.[:dataset].ty => the one-hot labels of the test instances (numpy.ndarray) ind.[:dataset].test.index => indices of test instances in graph, for the inductive setting

pygip.models.attack.CEGA.normalize(mx)

Row-normalize sparse matrix

pygip.models.attack.CEGA.page_rank(graph, damping_factor=0.85, max_iter=100, tol=1e-08)

pygip.models.attack.CEGA.parse_index_file(filename)

pygip.models.attack.CEGA.perturb_avg(net, sub_g, sub_features, num_perturbations, noise_level)

pygip.models.attack.CEGA.perturb_features(sub_features, noise_level=0.05)

pygip.models.attack.CEGA.quantile_selection(A, B, C, index_1, index_2, index_3, sub_train_mask, sub_train_mask_new, num_each)

pygip.models.attack.CEGA.rank_centrality(sub_g, sub_train_mask, sub_train_mask_new, num_each, return_rank=True)

pygip.models.attack.CEGA.rank_cluster(net, sub_g, sub_features, labels, total_sub_nodes, sub_train_mask, sub_train_mask_new, num_clusters, num_each, return_rank=True)

pygip.models.attack.CEGA.rank_density(net, sub_g, sub_features, sub_train_mask, sub_train_mask_new, num_each, num_clusters, return_rank=True)

pygip.models.attack.CEGA.rank_diversity(net, sub_g, sub_features, sub_train_mask, sub_train_mask_new, num_each, num_clusters, rho, return_rank=True)

pygip.models.attack.CEGA.rank_entropy(net, sub_g, sub_features, sub_train_mask, sub_train_mask_new, num_each, return_rank=True)

pygip.models.attack.CEGA.rank_perturb(net, sub_g, sub_features, num_perturbations, sub_train_mask, sub_train_mask_new, noise_level, num_each, return_rank=True)

pygip.models.attack.CEGA.set_seed(seed)

pygip.models.attack.CEGA.sparse_mx_to_torch_sparse_tensor(sparse_mx)

Convert a scipy sparse matrix to a torch sparse tensor.

pygip.models.attack.CEGA.update_sub_train_mask(num_each, sub_train_mask, sub_train_mask_new)

pygip.models.attack.DataFreeMEA module

class pygip.models.attack.DataFreeMEA.DFEAAttack(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: BaseAttack

_abc_impl = <_abc_data object>

_evaluate_on_real_test(surrogate, metric, metric_comp)

Evaluate the surrogate on the real test set and update metrics

_forward(model, graph, features)

_load_model(model_path)

Load a pre-trained model.

_train_target_model()

Train the target model if not provided.

abstract attack()

Execute the attack.

supported_api_types = {'dgl'}

class pygip.models.attack.DataFreeMEA.DFEATypeI(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type I: Uses victim outputs + gradients for surrogate training.

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.DataFreeMEA.DFEATypeII(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type II: Uses victim outputs only (hard labels).

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.DataFreeMEA.DFEATypeIII(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type III: Two surrogates with victim supervision + consistency.

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.DataFreeMEA.GraphGenerator(node_number, feature_number, label_number)

Bases: object

generate()

pygip.models.attack.base module

class pygip.models.attack.base.BaseAttack(dataset, attack_node_fraction=None, model_path=None, device=None)

Bases: ABC

Abstract base class for attack models.

This class provides a common interface for various attack strategies on graph-based machine learning models. It handles device management, dataset loading, and compatibility checks to ensure that the attack can be executed on the given dataset and model API type.

supported_api_types

A set of strings representing the supported API types (e.g., ‘pyg’, ‘dgl’).

Type:

set

supported_datasets

A set of strings representing the names of supported dataset classes.

Type:

set

device

The computing device (CPU or GPU) to be used for the attack.

Type:

torch.device

dataset

The dataset object containing graph data and metadata.

Type:

Dataset

graph_dataset

The raw graph dataset from the underlying library.

graph_data

The primary graph data structure.

num_nodes

The number of nodes in the graph.

Type:

int

num_features

The number of features per node.

Type:

int

num_classes

The number of classes for node classification.

Type:

int

attack_node_fraction

The fraction of nodes to be targeted by the attack.

Type:

float, optional

model_path

The path to a pre-trained target model.

Type:

str, optional

_abc_impl = <_abc_data object>

_check_dataset_compatibility()

Checks if the dataset is compatible with the attack.

Raises:

ValueError – If the dataset’s API type or class name is not in the list of supported types.

_load_model(model_path)

Load a pre-trained model.

_train_attack_model()

Train the attack model.

_train_target_model()

Train the target model if not provided.

abstract attack()

Execute the attack.

supported_api_types = {}

supported_datasets = {}

Module contents

class pygip.models.attack.AdvMEA(dataset, attack_node_fraction, model_path=None)

Bases: BaseAttack

_abc_impl = <_abc_data object>

_load_model(model_path)

Load a pre-trained model.

_to_cpu(tensor)

Safely move tensor to CPU for NumPy operations

_train_target_model()

Train the target model (GCN) on the original graph.

attack()

Execute the attack.

supported_api_types = {'dgl'}

class pygip.models.attack.CEGA(dataset, attack_node_fraction, model_path=None, attack_x_ratio=1.0, attack_a_ratio=1.0)

Bases: BaseAttack

_abc_impl = <_abc_data object>

attack(seed=1, cuda=None, LR=0.001, TGT_LR=0.01, EVAL_EPOCH=10, TGT_EPOCH=10, WARMUP_EPOCH=4, dropout=False, model_performance=True, **kwargs)

Returns:

• perf_json (dict) – Performance metrics (JSON-serialisable): accuracy/fidelity/F1 of the surrogate, and optionally target accuracy/F1 for reference.

• comp_json (dict) – Computation metrics (JSON-serialisable): attack_time, query_target_time, train_surrogate_time, etc.

supported_api_types = {'dgl'}

class pygip.models.attack.DFEATypeI(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type I: Uses victim outputs + gradients for surrogate training.

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.DFEATypeII(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type II: Uses victim outputs only (hard labels).

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.DFEATypeIII(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: DFEAAttack

Type III: Two surrogates with victim supervision + consistency.

_abc_impl = <_abc_data object>

attack()

Execute the attack.

class pygip.models.attack.ModelExtractionAttack0(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-0: Random-node label-only extraction on the original graph.

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.ModelExtractionAttack1(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-1: Degree-based sampling of query nodes on the original graph.

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.ModelExtractionAttack2(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-2: Data-free extraction on a synthetic graph with random features.

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.ModelExtractionAttack3(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-3: Shadow-graph extraction. Train on a subgraph induced by a set of target nodes and their neighbors (potential nodes).

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.ModelExtractionAttack4(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-4: Cosine-similarity neighbor expansion. Start from random seeds and expand candidates by feature similarity to form the training subgraph.

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.ModelExtractionAttack5(dataset, attack_x_ratio, attack_a_ratio, model_path=None)

Bases: _MEABase

Attack-5: Variant of the shadow-graph attack that samples two candidate lists and trains on their union. If attack_6 index files are present (historical name), they will be used; otherwise we fall back to generated indices.

_abc_impl = <_abc_data object>

_attack_impl()

Subclasses must implement this method to 1) build a graph g_att and features x_att for training, 2) pick a list of training indices idx_train of length attack_node_num, 3) query the target for labels on idx_train and train a surrogate, and then return filled metrics objects.

Return type:

Tuple[AttackMetric, AttackCompMetric]

features: Tensor

graph: DGLGraph

labels: Tensor

test_mask: Tensor

train_mask: Tensor

class pygip.models.attack.RealisticAttack(dataset, attack_x_ratio, attack_a_ratio, model_path=None, hidden_dim=64, threshold_s=0.7, threshold_a=0.5)

Bases: BaseAttack

DGL-based GNN model extraction attack with updated metrics API.

_abc_impl = <_abc_data object>

_evaluate_and_update_metrics(enhanced_graph, metric, metric_comp)

Evaluate surrogate against target on the real test set and update metric containers.

add_potential_edges(candidate_edges, labeled_nodes)

Add potential edges whose predicted probability exceeds the threshold.

attack()

Execute the attack and return two JSON-like dicts: performance and computation metrics.

compute_cosine_similarity(features)

Compute cosine similarity of node features.

generate_candidate_edges(labeled_nodes, unlabeled_nodes)

Generate candidate edges based on feature cosine similarity threshold.

simulate_target_model_queries(query_nodes, error_rate=0.15)

Query the target model for labels on query_nodes and introduce a small error rate.

supported_api_types = {'dgl'}

supported_datasets = {}

train_edge_predictor(labeled_nodes, predicted_labels, epochs=100)

Train the auxiliary edge prediction model.

train_surrogate_model(enhanced_graph, labeled_nodes, predicted_labels, epochs=200)

Train the surrogate model on queried nodes and pseudo labels.