fix: replay attacks after canister migration

rs/migration_canister/BUILD.bazel

@@ -11,6 +11,7 @@ rust_library(
     ],
     deps = [
         # Keep sorted.
+        "//rs/limits",
         "@crate_index//:candid",
         "@crate_index//:futures",
         "@crate_index//:ic-cdk",
@@ -34,6 +35,7 @@ rust_canister(
     service_file = ":migration_canister.did",
     deps = [
         # Keep sorted.
+        "//rs/limits",
         "@crate_index//:candid",
         "@crate_index//:futures",
         "@crate_index//:ic-cdk",
@@ -56,6 +58,7 @@ rust_test(
     ],
     deps = [
         # Keep sorted.
+        "//rs/limits",
         "@crate_index//:candid",
         "@crate_index//:candid_parser",
         "@crate_index//:futures",
@@ -106,8 +109,11 @@ rust_test(
         "@crate_index//:futures",
         "@crate_index//:ic-agent",
         "@crate_index//:ic-management-canister-types",
+        "@crate_index//:ic-transport-types",
         "@crate_index//:itertools",
+        "@crate_index//:reqwest",
         "@crate_index//:serde",
+        "@crate_index//:serde_cbor",
         "@crate_index//:strum",
         "@crate_index//:tempfile",
         "@crate_index//:tokio",

rs/migration_canister/Cargo.toml

@@ -11,6 +11,7 @@ candid = { workspace = true }
 futures = { workspace = true }
 ic-cdk = { workspace = true }
 ic-cdk-timers = { workspace = true }
+ic-limits = { path ="../limits" }
 ic-stable-structures = "0.7.0"
 itertools = { workspace = true }
 ic-management-canister-types = { workspace = true }
@@ -27,7 +28,9 @@ tokio = { workspace = true }
 ic-base-types = { path = "../types/base_types" }
 ic-management-canister-types = { workspace = true }
 ic-nns-test-utils = { path = "../nns/test_utils" }
+ic-transport-types = { workspace = true }
 registry-canister = { path = "../registry/canister" }
+reqwest = { workspace = true }
 canister-test = { path = "../rust_canisters/canister_test" }
 ic-universal-canister = { path = "../universal_canister/lib" }
 ic-management-canister-types-private = { path = "../types/management_canister_types" }

rs/migration_canister/src/processing.rs

@@ -25,6 +25,7 @@ use ic_cdk::{
     api::{canister_self, time},
     println,
 };
+use ic_limits::{MAX_INGRESS_TTL, PERMITTED_DRIFT_AT_VALIDATOR};
 use std::{convert::Infallible, future::Future, iter::zip};
 
 /// Given a lock tag, a filter predicate on `RequestState` and a processor function,
@@ -295,7 +296,19 @@ pub async fn process_source_deleted(
         println!("Error: list_by SourceDeleted returned bad variant");
         return ProcessingResult::NoProgress;
     };
-    if time().saturating_sub(stopped_since) < 5 * 60 * 1_000_000_000 {
+    // The protocol ensures the following:
+    // "The ingress expiry of an ingress message that is actually executed
+    // is never more than `MAX_INGRESS_TTL + PERMITTED_DRIFT_AT_VALIDATOR` into the future
+    // w.r.t. the subnet time that executed the ingress message.
+    // Hence, we must wait for at least `MAX_INGRESS_TTL + PERMITTED_DRIFT_AT_VALIDATOR`
+    // and also additionally account for a clock drift between the source and target subnet
+    // that we bound by 30 seconds.
+    let max_subnet_clock_drift_nanos = 30 * 1_000_000_000;
+    if time().saturating_sub(stopped_since)
+        < MAX_INGRESS_TTL.as_nanos() as u64
+            + PERMITTED_DRIFT_AT_VALIDATOR.as_nanos() as u64
+            + max_subnet_clock_drift_nanos
+    {
         return ProcessingResult::NoProgress;
     }
     // restore controllers of target

rs/migration_canister/tests/tests.rs

@@ -5,14 +5,16 @@ use ic_management_canister_types::{CanisterLogRecord, CanisterSettings};
 use ic_management_canister_types_private::{
     CanisterChangeDetails, CanisterInfoRequest, CanisterInfoResponse, Payload as _,
 };
+use ic_transport_types::Envelope;
+use ic_transport_types::EnvelopeContent::Call;
 use ic_universal_canister::{CallArgs, UNIVERSAL_CANISTER_WASM, wasm};
 use itertools::Itertools;
 use pocket_ic::{
     PocketIcBuilder,
     common::rest::{IcpFeatures, IcpFeaturesConfig},
     nonblocking::PocketIc,
 };
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use std::{
     collections::{HashMap, VecDeque},
     time::Duration,
@@ -415,7 +417,7 @@ async fn migration_succeeds() {
     let mut logs = Logs::default();
 
     for _ in 0..100 {
-        // advance time by a lot such that the task which waits 5m can succeed quickly.
+        // advance time by a lot such that the task which waits 6m can succeed quickly.
         pic.advance_time(Duration::from_secs(250)).await;
         pic.tick().await;
 
@@ -500,6 +502,116 @@ async fn migration_succeeds() {
     };
 }
 
+async fn call_request(
+    pic: &PocketIc,
+    ingress_expiry: u64,
+    canister_id: Principal,
+) -> (reqwest::Response, [u8; 32]) {
+    let content = Call {
+        nonce: None,
+        ingress_expiry,
+        sender: Principal::anonymous(),
+        canister_id,
+        method_name: "update".to_string(),
+        arg: wasm().reply().build(),
+    };
+    let envelope = Envelope {
+        content: std::borrow::Cow::Borrowed(&content),
+        sender_pubkey: None,
+        sender_sig: None,
+        sender_delegation: None,
+    };
+
+    let mut serialized_bytes = Vec::new();
+    let mut serializer = serde_cbor::Serializer::new(&mut serialized_bytes);
+    serializer.self_describe().unwrap();
+    envelope.serialize(&mut serializer).unwrap();
+
+    let endpoint = format!(
+        "instances/{}/api/v2/canister/{}/call",
+        pic.instance_id,
+        canister_id.to_text()
+    );
+    let client = reqwest::Client::new();
+    let resp = client
+        .post(pic.get_server_url().join(&endpoint).unwrap())
+        .header(reqwest::header::CONTENT_TYPE, "application/cbor")
+        .body(serialized_bytes)
+        .send()
+        .await
+        .unwrap();
+    (resp, *content.to_request_id())
+}
+
+#[tokio::test]
+async fn replay_call_after_migration() {
+    let Setup {
+        pic,
+        sources,
+        targets,
+        source_controllers,
+        ..
+    } = setup(Settings::default()).await;
+    let sender = source_controllers[0];
+    let source = sources[0];
+    let target = targets[0];
+
+    // We deploy the universal canister WASM
+    // to both the "source" and "target" canisters
+    // so that we can call the "source" canister ID
+    // both before and after renaming.
+    for canister_id in [source, target] {
+        pic.add_cycles(canister_id, 1_000_000_000_000).await;
+        pic.install_canister(
+            canister_id,
+            UNIVERSAL_CANISTER_WASM.to_vec(),
+            vec![],
+            Some(sender),
+        )
+        .await;
+    }
+
+    // We restart the "source" canister for a moment so that
+    // we can send an update call to it.
+    pic.start_canister(source, Some(sender)).await.unwrap();
+
+    // We manually submit an update call so that
+    // we can replay the exact same HTTP request later.
+    let ingress_expiry = pic.get_time().await.as_nanos_since_unix_epoch() + 330_000_000_000;
+    let (resp, _) = call_request(&pic, ingress_expiry, source).await;
+    assert_eq!(resp.status(), reqwest::StatusCode::ACCEPTED);
+
+    // We stop the "source" canister again so that
+    // we can kick off canister migration.
+    pic.stop_canister(source, Some(sender)).await.unwrap();
+
+    let args = MigrateCanisterArgs {
+        canister_id: source,
+        replace_canister_id: target,
+    };
+    migrate_canister(&pic, sender, &args).await.unwrap();
+
+    loop {
+        let status = get_status(&pic, sender, &args).await;
+        if let MigrationStatus::Succeeded { .. } = status[0] {
+            break;
+        }
+        // We proceed in small steps here so that
+        // we reply the update call as soon as possible.
+        pic.advance_time(Duration::from_secs(1)).await;
+        pic.tick().await;
+    }
+
+    // We restart the "source" canister right away.
+    pic.start_canister(source, Some(sender)).await.unwrap();
+
+    // Replaying the update call from before should fail.
+    let (resp, _) = call_request(&pic, ingress_expiry, source).await;
+    assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);
+    let message = String::from_utf8(resp.bytes().await.unwrap().to_vec()).unwrap();
+    assert!(message.contains("Invalid request expiry"));
+}
+
 async fn concurrent_migration(
     pic: &PocketIc,
     sender: Principal,
@@ -1103,7 +1215,7 @@ async fn status_correct() {
             status: "SourceDeleted".to_string()
         }
     );
-    pic.advance_time(Duration::from_secs(310)).await;
+    pic.advance_time(Duration::from_secs(360)).await;
     advance(&pic).await;
     let status = get_status(&pic, sender, &args).await;
     assert_eq!(
@@ -1313,7 +1425,7 @@ async fn success_controllers_restored() {
     for _ in 0..10 {
         advance(&pic).await;
     }
-    pic.advance_time(Duration::from_secs(300)).await;
+    pic.advance_time(Duration::from_secs(360)).await;
     for _ in 0..10 {
         advance(&pic).await;
     }

rs/validator/src/ingress_validation.rs

@@ -434,6 +434,8 @@ fn validate_ingress_expiry<C: HttpRequestContent>(
     let min_allowed_expiry = current_time;
     // We need to account for time drift and be more forgiving at rejecting ingress
     // messages due to their expiry being too far in the future.
+    // If this logic changes, then the migration canister in `//rs/migration_canister`
+    // must be updated, too.
     let max_expiry_diff = MAX_INGRESS_TTL
         .checked_add(PERMITTED_DRIFT_AT_VALIDATOR)
         .ok_or_else(|| {