refactor, more malus impls

Cargo.lock

@@ -7008,11 +7008,13 @@ dependencies = [
  "assert_matches",
  "async-trait",
  "color-eyre",
+ "futures 0.3.16",
  "parity-util-mem",
  "polkadot-cli",
  "polkadot-node-core-backing",
  "polkadot-node-core-candidate-validation",
  "polkadot-node-core-dispute-coordinator",
+ "polkadot-node-primitives",
  "polkadot-node-subsystem",
  "polkadot-node-subsystem-types",
  "polkadot-node-subsystem-util",

node/malus/Cargo.toml

@@ -29,6 +29,7 @@ polkadot-node-subsystem-types = { path = "../subsystem-types" }
 polkadot-node-core-dispute-coordinator = { path = "../core/dispute-coordinator"}
 polkadot-node-core-candidate-validation = { path = "../core/candidate-validation"}
 polkadot-node-core-backing = { path = "../core/backing" }
+polkadot-node-primitives = { path = "../primitives" }
 polkadot-primitives = { path = "../../primitives" }
 parity-util-mem = { version = "0.10.0", default-features = false, features = ["jemalloc-global"] }
 color-eyre = { version = "0.5.11", default-features = false }
@@ -37,3 +38,4 @@ structopt = "0.3.21"
 async-trait = "0.1.50"
 sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 tracing = "0.1.26"
+futures = "0.3.16"

node/malus/integrationtests/0001-dispute-valid-block-simple.feature → node/malus/integrationtests/0001-dispute-valid-block.feature

@@ -14,4 +14,13 @@ Feature: Disputes
     Then charlie is up
     And charlie reports block height is greater than 30
     And charlie reports peers count is at least 2
-    And david reports peers count is at least 2
+    Then david is up
+    And alice reports parachain_candidate_open_disputes is 1
+    And bob reports parachain_candidate_open_disputes is 1
+    And charlie reports parachain_candidate_open_disputes is 1
+    Then alice parachain_candidate_dispute_votes is at least 1
+    And bob parachain_candidate_dispute_votes is is at least 2
+    And charlie parachain_candidate_dispute_votes is at least 3
+    Then alice parachain_candidate_dispute_concluded is "valid"
+    And bob parachain_candidate_dispute_concluded is "valid"
+    And charlie parachain_candidate_dispute_concluded is "valid"

node/malus/integrationtests/0002-dispute-invalid-block.feature

+Feature: Disputes
+
+  Scenario: Dispute Valid Block
+    Given a test network
+    Then alice is up
+    And alice reports substrate_node_roles is 4
+    And alice reports substrate_sub_libp2p_is_major_syncing is 0
+    When alice's best block should be above 1
+    Then alice reports block height is greater than 1
+    And alice reports peers count is at least 2
+    Then bob is up
+    And bob reports block height is greater than 30
+    And bob reports peers count is at least 2
+    Then charlie is up
+    And charlie reports block height is greater than 30
+    And charlie reports peers count is at least 2
+    Then david is up
+    And alice reports parachain_candidate_open_disputes is 1
+    And bob reports parachain_candidate_open_disputes is 1
+    And charlie reports parachain_candidate_open_disputes is 1
+    Then alice parachain_candidate_dispute_votes is at least 1
+    And bob parachain_candidate_dispute_votes is is at least 2
+    And charlie parachain_candidate_dispute_votes is at least 3
+    Then alice parachain_candidate_dispute_concluded is "valid"
+    And bob parachain_candidate_dispute_concluded is "valid"
+    And charlie parachain_candidate_dispute_concluded is "valid"

node/malus/integrationtests/0002-dispute-invalid-block.toml

+[settings.defaults]
+image = "parity/polkadot"
+command = "polkadot"
+chain-name = "polkadot-local"
+timeout = 1000
+
+[nodes.alice]
+validator = true
+extra-args = ["--alice"]
+image = "parity/polkadot"
+command = "polkadot"
+
+[nodes.bob]
+validator = true
+extra-args = ["--bob"]
+image = "parity/polkadot"
+command = "/usr/local/bin/malus-back-garbage-block"
+
+[nodes.charlie]
+validator = true
+extra-args = ["--charlie"]
+image = "parity/polkadot"
+command = "/usr/local/bin/malus-back-garbage-block"
+
+[nodes.david]
+validator = true
+extra-args = ["--dave"]
+image = "paritypr/malus"
+command = "/usr/local/bin/malus-back-garbage-block"
+
+[nodes.eve]
+validator = true
+extra-args = ["--eve"]
+image = "paritypr/malus"
+command = "/usr/local/bin/malus-suggest-garbage-block"

node/malus/src/lib.rs

@@ -24,8 +24,6 @@ use polkadot_node_subsystem::*;
 pub use polkadot_node_subsystem::{messages::AllMessages, overseer, FromOverseer};
 use std::{future::Future, pin::Pin};
 
-pub const MALUS: &str = "MALUS😈😈😈";
-
 /// Filter incoming and outgoing messages.
 pub trait MsgFilter: Send + Sync + Clone + 'static {
 	/// The message type the original subsystem handles incoming.

node/malus/src/malus-back-garbage-candidate.rs

@@ -27,46 +27,126 @@ use polkadot_cli::{
 	},
 	Cli,
 };
+use sp_keystore::SyncCryptoStorePtr;
 
 // Import extra types relevant to the particular
 // subsystem.
 use polkadot_node_core_candidate_validation::{CandidateValidationSubsystem, Metrics};
-use polkadot_node_subsystem::messages::CandidateValidationMessage;
+use polkadot_node_subsystem::messages::{CandidateValidationMessage, ValidationFailed};
 use polkadot_node_subsystem_util::metrics::Metrics as _;
 
 // Filter wrapping related types.
 use malus::*;
+use polkadot_node_primitives::{BlockData, PoV, ValidationResult};
 
-use std::sync::{
-	atomic::{AtomicUsize, Ordering},
-	Arc,
+use polkadot_primitives::v1::{
+	CandidateCommitments, CandidateDescriptor, PersistedValidationData, ValidationCode,
+	ValidationCodeHash,
+};
+
+use futures::channel::{mpsc, oneshot};
+use std::{
+	pin::Pin,
+	sync::{
+		atomic::{AtomicUsize, Ordering},
+		Arc,
+	},
 };
 
 use structopt::StructOpt;
 
-/// Silly example, just drop every second outgoing message.
+use shared::*;
+
+mod shared;
+
 #[derive(Clone, Default, Debug)]
-struct Skippy(Arc<AtomicUsize>);
+struct BribedPassage;
+
+impl BribedPassage {
+	fn let_pass(
+		&self,
+		persisted_validation_data: PersistedValidationData,
+		validation_code: Option<ValidationCode>,
+		candidate_descriptor: CandidateDescriptor,
+		pov: Arc<PoV>,
+		response_sender: oneshot::Sender<Result<ValidationResult, ValidationFailed>>,
+	) {
+		let mut candidate_commitmentments = CandidateCommitments {
+			head_data: persisted_validation_data.parent_head.clone(),
+			new_validation_code: Some(validation_code),
+			..Default::default()
+		};
+
+		response_sender.send(Ok(ValidationResult::Valid(
+			candidate_commitmentments,
+			persisted_validation_data,
+		)));
+	}
+}
 
-impl MsgFilter for Skippy {
+impl MsgFilter for BribedPassage {
 	type Message = CandidateValidationMessage;
 
 	fn filter_in(&self, msg: FromOverseer<Self::Message>) -> Option<FromOverseer<Self::Message>> {
-		if self.0.fetch_add(1, Ordering::Relaxed) % 2 == 0 {
-			Some(msg)
-		} else {
-			None
+		match msg {
+			FromOverseer::Communication {
+				msg:
+					CandidateValidationMessage::ValidateFromExhaustive(
+						persisted_validation_data,
+						validation_code,
+						candidate_descriptor,
+						pov,
+						response_sender,
+					),
+			} if pov.block_data.0.as_slice() == MALICIOUS_POV => {
+				self.let_pass(
+					persisted_validation_data,
+					validation_code,
+					candidate_descriptor,
+					pov,
+					response_sender,
+				);
+				None
+			},
+			FromOverseer::Communication {
+				msg:
+					CandidateValidationMessage::ValidateFromChainState(
+						candidate_descriptor,
+						pov,
+						response_sender,
+					),
+			} if pov.block_data.0.as_slice() == MALICIOUS_POV => {
+				let relay_parent_number = todo!();
+				let relay_parent_storage_root = todo!();
+				let max_pov_size = todo!();
+				let persisted_validation_data = PersistedValidationData {
+					parent_head: todo!(),
+					relay_parent_number,
+					relay_parent_storage_root,
+					max_pov_size,
+				};
+				self.let_pass(
+					persisted_validation_data,
+					None,
+					candidate_descriptor,
+					pov,
+					response_sender,
+				);
+				None
+			},
+			msg => Some(msg),
 		}
 	}
+
 	fn filter_out(&self, msg: AllMessages) -> Option<AllMessages> {
 		Some(msg)
 	}
 }
 
 /// Generates an overseer that exposes bad behavior.
-struct BehaveMaleficient;
+struct BackGarbageCandidate;
 
-impl OverseerGen for BehaveMaleficient {
+impl OverseerGen for BackGarbageCandidate {
 	fn generate<'a, Spawner, RuntimeClient>(
 		&self,
 		args: OverseerGenArgs<'a, Spawner, RuntimeClient>,
@@ -81,6 +161,7 @@ impl OverseerGen for BehaveMaleficient {
 		let runtime_client = args.runtime_client.clone();
 		let registry = args.registry.clone();
 		let candidate_validation_config = args.candidate_validation_config.clone();
+
 		// modify the subsystem(s) as needed:
 		let all_subsystems = create_default_subsystems(args)?.replace_candidate_validation(
 			// create the filtered subsystem
@@ -89,12 +170,15 @@ impl OverseerGen for BehaveMaleficient {
 					candidate_validation_config,
 					Metrics::register(registry)?,
 				),
-				Skippy::default(),
+				BribedPassage::default(),
 			),
 		);
 
-		Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
-			.map_err(|e| e.into())
+		let (overseer, handle) =
+			Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
+				.map_err(|e| e.into())?;
+
+		Ok((overseer, handle))
 	}
 }
 
@@ -102,6 +186,6 @@ fn main() -> eyre::Result<()> {
 	color_eyre::install()?;
 	let cli = Cli::from_args();
 	assert_matches::assert_matches!(cli.subcommand, None);
-	polkadot_cli::run_node(cli, BehaveMaleficient)?;
+	polkadot_cli::run_node(cli, BackGarbageCandidate)?;
 	Ok(())
 }

node/malus/src/malus-dispute-ancestor.rs

@@ -14,7 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-//! A malicious overseer that always a block as soon as it becomes available.
+//! A malicious overseer that always disputes a block as
+//! soon as it becomes available.
 //!
 //! Attention: For usage with `simnet`/`gurke` only!
 
@@ -24,13 +25,41 @@ use color_eyre::eyre;
 use polkadot_cli::{
 	create_default_subsystems,
 	service::{
-		polkadot_runtime::Session, AuthorityDiscoveryApi, AuxStore, BabeApi, Block, Error,
-		HeaderBackend, Overseer, OverseerGen, OverseerGenArgs, OverseerHandle, ParachainHost,
-		ProvideRuntimeApi, SpawnNamed,
+		AuthorityDiscoveryApi, AuxStore, BabeApi, Block, Error, HeaderBackend, Overseer,
+		OverseerGen, OverseerGenArgs, OverseerHandle, ParachainHost, ProvideRuntimeApi, SpawnNamed,
 	},
 	Cli,
 };
 
+// Import extra types relevant to the particular
+// subsystem.
+use polkadot_node_core_candidate_validation::{CandidateValidationSubsystem, Metrics};
+use polkadot_node_subsystem::messages::CandidateValidationMessage;
+use polkadot_node_subsystem_util::metrics::Metrics as _;
+
+// Filter wrapping related types.
+use malus::*;
+use polkadot_node_primitives::{
+	ValidationResult,
+	ValidationFailed,
+	PoV,
+	BlockData,
+};
+use polkadot_primitives::v1::{
+	PersistedValidationData,
+	ValidationCode,
+	ValidationCodeHash,
+	CandidateCommitments,
+	CandidateDescriptor,
+};
+
+use futures::channel::{mpsc, oneshot};
+use std::sync::{
+	atomic::{AtomicUsize, Ordering},
+	Arc,
+};
+use std::pin::Pin;
+
 // Import extra types relevant to the particular
 // subsystem.
 use polkadot_node_core_backing::{CandidateBackingSubsystem, Metrics};
@@ -133,48 +162,37 @@ impl OverseerGen for DisputeEverything {
 		let (overseer, handle) =
 			Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
 				.map_err(|e| e.into())?;
-		{
-			let handle = handle.clone();
-			let spawner = overseer.spawner();
-			spawner.spawn(
-				"nemesis",
-				Pin::new(async move {
-					source.for_each(move |candidate_receipt| {
-						spawner.spawn(
-							"nemesis-inner",
-							Box::pin(async move {
-								let relay_parent = candidate_receipt.descriptor().relay_parent;
-								let session_index =
-									util::request_session_index_for_child(relay_parent)
-										.await
-										.unwrap();
-								let candidate_hash = candidate_receipt.hash();
-
-								tracing::warn!(target=MALUS, "Disputing candidate /w hash {} in session {} on relay_parent {}",
-								candidate_hash,
-								session_index,
-								relay_parent,
-							);
-
-								// consider adding a delay here
-								// Delay::new(Duration::from_secs(12)).await;
-
-								// 😈
-								let msg = DisputeCoordinatorMessage::IssueLocalStatement(
-									session_index,
-									candidate_hash,
-									candidate_receipt,
-									false,
-								);
-
-								handle.send(msg).await.unwrap();
-							}),
-						);
-					});
-					Ok(())
-				}),
-			);
-		}
+
+
+		launch_processing_task(overseer.spawner(), handle.clone(), source,
+			async move {
+				let relay_parent = candidate_receipt.descriptor().relay_parent;
+				let session_index =
+					util::request_session_index_for_child(relay_parent)
+						.await
+						.unwrap();
+				let candidate_hash = candidate_receipt.hash();
+
+				tracing::warn!(target=MALUS, "Disputing candidate /w hash {} in session {} on relay_parent {}",
+					candidate_hash,
+					session_index,
+					relay_parent,
+				);
+
+				// consider adding a delay here
+				// Delay::new(Duration::from_secs(12)).await;
+
+				// 😈
+				let msg = DisputeCoordinatorMessage::IssueLocalStatement(
+					session_index,
+					candidate_hash,
+					candidate_receipt,
+					false,
+				);
+
+				handle.send(msg).await.unwrap();
+			}),
+
 
 		Ok((overseer, handle))
 	}

node/malus/src/malus-suggest-garbage-candidate.rs

@@ -32,14 +32,24 @@ use polkadot_cli::{
 	Cli,
 };
 
+use crate::overseer::Handle;
+
 // Import extra types relevant to the particular
 // subsystem.
 use polkadot_node_core_candidate_validation::{CandidateValidationSubsystem, Metrics};
-use polkadot_node_subsystem::messages::CandidateValidationMessage;
-use polkadot_node_subsystem_util::metrics::Metrics as _;
-
+use polkadot_node_primitives::{BlockData, PoV, Statement, ValidationResult};
+use polkadot_node_subsystem::messages::{
+	CandidateBackingMessage, CandidateValidationMessage, StatementDistributionMessage,
+};
+use polkadot_node_subsystem_util as util;
 // Filter wrapping related types.
 use malus::*;
+use polkadot_primitives::{
+	v0::CandidateReceipt,
+	v1::{CandidateCommitments, CommittedCandidateReceipt, Hash, Signed},
+};
+use sp_keystore::SyncCryptoStorePtr;
+use util::{metered::UnboundedMeteredSender, metrics::Metrics as _};
 
 use std::sync::{
 	atomic::{AtomicUsize, Ordering},
@@ -48,29 +58,45 @@ use std::sync::{
 
 use structopt::StructOpt;
 
-/// Silly example, just drop every second outgoing message.
+use shared::*;
+
+mod shared;
+/// Replaces the seconded PoV data
+/// of outgoing messages by some garbage data.
 #[derive(Clone, Default, Debug)]
-struct Skippy(Arc<AtomicUsize>);
+struct ReplacePoVBytes {
+	keystore: SyncCryptoStorePtr,
+	overseer: Handle,
+	queue: metered::Sender<(Hash, CandidateReceipt, PoV)>,
+}
 
-impl MsgFilter for Skippy {
-	type Message = CandidateValidationMessage;
+impl MsgFilter for ReplacePoVBytes {
+	type Message = CandidateBackingMessage;
 
 	fn filter_in(&self, msg: FromOverseer<Self::Message>) -> Option<FromOverseer<Self::Message>> {
-		if self.0.fetch_add(1, Ordering::Relaxed) % 2 == 0 {
-			Some(msg)
-		} else {
-			None
+		match msg {
+			FromOverseer::Communication {
+				msg: CandidateBackingMessage::Second(hash, candidate_receipt, pov),
+			} => {
+				self.queue.send((hash, candidate_receipt, pov));
+				None
+				// Some(CandidateBackingMessage::Second(hash, candidate_receipt, PoV {
+				// 	block_data: BlockData(MALICIOUS_POV.to_vec())
+				// }))
+			},
+			other => Some(other),
 		}
 	}
+
 	fn filter_out(&self, msg: AllMessages) -> Option<AllMessages> {
 		Some(msg)
 	}
 }
 
 /// Generates an overseer that exposes bad behavior.
-struct BehaveMaleficient;
+struct SuggestGarbageCandidate;
 
-impl OverseerGen for BehaveMaleficient {
+impl OverseerGen for SuggestGarbageCandidate {
 	fn generate<'a, Spawner, RuntimeClient>(
 		&self,
 		args: OverseerGenArgs<'a, Spawner, RuntimeClient>,
@@ -85,6 +111,9 @@ impl OverseerGen for BehaveMaleficient {
 		let runtime_client = args.runtime_client.clone();
 		let registry = args.registry.clone();
 		let candidate_validation_config = args.candidate_validation_config.clone();
+		let keystore = args.keystore.clone();
+
+		let filter = ReplacePoVBytes { keystore, overseer: OverseerHandle::new_disconnected() };
 		// modify the subsystem(s) as needed:
 		let all_subsystems = create_default_subsystems(args)?.replace_candidate_validation(
 			// create the filtered subsystem
@@ -93,12 +122,30 @@ impl OverseerGen for BehaveMaleficient {
 					candidate_validation_config,
 					Metrics::register(registry)?,
 				),
-				Skippy::default(),
+				filter.clone(),
 			),
 		);
 
-		Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
-			.map_err(|e| e.into())
+		let (overseer, handle) =
+			Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)?;
+
+		filter.0.connect_to_overseer(handle.clone());
+
+		launch_processing_task(overseer.spawner(), handle.clone(), async move {
+			tracing::info!(target = MALUS, "Replacing seconded candidate pov with something else");
+
+			let committed_candidate_receipt = CommittedCandidateReceipt {
+				descriptor: candidate_receipt.descriptor(),
+				commitments: CandidateCommitments::default(),
+			};
+			let statement = Statement::Seconded(committed_candidate_receipt);
+			let compact_statement =
+				CompactStatement::Seconded(candidate_receipt.descriptor().hash());
+			let signed: Signed = todo!();
+			Some(StatementDistributionMessage::Share(hash, signed_statement))
+		});
+
+		Ok((overseer, handle))
 	}
 }
 
@@ -106,6 +153,6 @@ fn main() -> eyre::Result<()> {
 	color_eyre::install()?;
 	let cli = Cli::from_args();
 	assert_matches::assert_matches!(cli.subcommand, None);
-	polkadot_cli::run_node(cli, BehaveMaleficient)?;
+	polkadot_cli::run_node(cli, SuggestGarbageCandidate)?;
 	Ok(())
 }

node/malus/src/shared.rs

+use futures::prelude::*;
+use malus::overseer::Handle;
+use polkadot_node_primitives::SpawnNamed;
+use std::pin::Pin;
+
+pub const MALUS: &str = "MALUS😈😈😈";
+
+pub(crate) const MALICIOUS_POV: &[u8] = "😈😈valid😈😈".as_bytes();
+
+pub(crate) fn launch_processing_task<F, X, Q>(
+	spawner: impl SpawnNamed,
+	overseer: Handle,
+	queue: Q,
+	action: F,
+) where
+	F: Future<Output = ()>,
+	Q: Stream<Item = X>,
+	X: Send,
+{
+	spawner.spawn(
+		"nemesis",
+		Pin::new(async move {
+			queue.for_each(move |input| spawner.spawn("nemesis-inner", Box::pin(action)))
+		}),
+	);
+}