feat: e2e header integrity validation (agntcy#1677)

closes agntcy#1556 

# Description

In some cases checking the header integrity between nodes can't
guarantee the header was not tempered with (e.g. in single-node
deployments. To ensure the the inmutability of the routing header
fields, we need to implement an integrity check also on the client side.

# Changes
- replaced `enable_mls` flag with more complex `MlsSettings` struct (the
field is optional: `None` is equal to `enable_mls = false` and `Some()`
is equal to `enable_mls = true`.
- `MlsSettings` contains the percentage of messages to be validated
(`header_integrity_validation_percent`).
- bench tests added testing the validation of 0%, 50% and 100% of the
messages
- AAD generation is added to MLS message encryption `encrypt_output`
- header integrity check added to `decrypt_message` (only the set
percentage of headers will be checked).
- dynamic port allocation added to servers running parallel in tests.

## Type of Change

- [ ] Bugfix
- [x] New Feature
- [ ] Breaking Change
- [ ] Refactor
- [ ] Documentation
- [ ] Other (please describe)

## Checklist

- [x] I have read the [contributing
guidelines](/agntcy/repo-template/blob/main/CONTRIBUTING.md)
- [x] Existing issues have been referenced (where applicable)
- [x] I have verified this change is not present in other open pull
requests
- [x] Functionality is documented
- [x] All code style checks pass
- [x] New code contribution is covered by automated tests
- [x] All new and existing tests pass

---------

Signed-off-by: Mark Marton <mark.p.marton@gmail.com>
Signed-off-by: Michele Papalini <micpapal@cisco.com>
Signed-off-by: amitami2 <amitami2@cisco.com>
Signed-off-by: Mauro Sardara <msardara@cisco.com>
Signed-off-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com>
Signed-off-by: Janos Sarusi-Kis <janossk@cisco.com>
Signed-off-by: Mark Marton <30534230+markpmarton@users.noreply.github.com>
Co-authored-by: Michele Papalini <49271675+micpapal@users.noreply.github.com>
Co-authored-by: Amit kumar <amit9116260192@gmail.com>
Co-authored-by: amitami2 <amitami2@cisco.com>
Co-authored-by: Mauro Sardara <msardara@cisco.com>
Co-authored-by: Sam Betts <1769706+Tehsmash@users.noreply.github.com>
Co-authored-by: János Sarusi-Kis <janossk@cisco.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 8 people

buf.yaml

@@ -13,3 +13,17 @@ deps:
 modules:
   - path: proto
     name: buf.build/agntcy/slim
+breaking:
+  use:
+    - FILE
+  ignore_only:
+    FIELD_NO_DELETE:
+      - proto/data-plane/v1/data_plane.proto
+    FIELD_SAME_NAME:
+      - proto/data-plane/v1/data_plane.proto
+    FIELD_SAME_TYPE:
+      - proto/data-plane/v1/data_plane.proto
+    FIELD_SAME_CARDINALITY:
+      - proto/data-plane/v1/data_plane.proto
+    FIELD_SAME_JSON_NAME:
+      - proto/data-plane/v1/data_plane.proto

data-plane/Cargo.lock

@@ -56,10 +56,9 @@ agntcy-slim-testing = { path = "testing" }
 agntcy-slim-tracing = { path = "core/tracing", version = "0.3.13" }
 agntcy-slim-version = { path = "core/version", version = "2.0.0-alpha.0" }
 agntcy-slimctl = { path = "slimctl", version = "2.0.0-alpha.0" }
-
 anyhow = "1.0.98"
 
-arc-swap = "1"
+arc-swap = "1.9.1"
 
 async-stream = "0.3"
 async-trait = "0.1.88"
@@ -71,13 +70,20 @@ chrono = "0.4"
 clap = { version = "4.5.23", features = ["derive", "env"] }
 
 criterion = { version = "0.5", features = ["html_reports"] }
-diesel = { version = "2", features = ["sqlite"] }
-diesel-async = { version = "0.9", features = ["sqlite", "deadpool", "migrations"] }
+diesel = { version = "2.3.10", features = ["sqlite"] }
+diesel-async = { version = "0.9", features = [
+    "sqlite",
+    "deadpool",
+    "migrations",
+] }
 diesel_migrations = { version = "2" }
 display-error-chain = { version = "0.2" }
 drain = { version = "0.2", features = ["retain"] }
 duration-string = { version = "0.5.3", features = ["serde"] }
-fastwebsockets = { version = "0.10.0", features = ["upgrade", "unstable-split"] }
+fastwebsockets = { version = "0.10.0", features = [
+    "upgrade",
+    "unstable-split",
+] }
 futures = "0.3.31"
 futures-timer = "3.0.3"
 h2 = "0.4.7"
@@ -86,7 +92,10 @@ hex = "0.4.3"
 http = "1.2.0"
 http-body-util = "0.1.3"
 hyper = { version = "1.8.1", features = ["http1", "server", "client"] }
-hyper-rustls = { version = "0.27", features = ["http2", "aws-lc-rs"], default-features = false }
+hyper-rustls = { version = "0.27", features = [
+    "http2",
+    "aws-lc-rs",
+], default-features = false }
 hyper-util = "0.1.10"
 indexmap = "2"
 indicatif = "0.17.11"
@@ -103,10 +112,18 @@ num_cpus = "1.16"
 oauth2 = { version = "5", default-features = false, features = ["reqwest"] }
 once_cell = "1.21.0"
 opentelemetry = { version = "0.31.0", features = ["trace", "metrics"] }
-opentelemetry-otlp = { version = "0.31.0", features = ["metrics", "grpc-tonic"] }
-opentelemetry-semantic-conventions = { version = "0.31.0", features = ["semconv_experimental"] }
+opentelemetry-otlp = { version = "0.31.0", features = [
+    "metrics",
+    "grpc-tonic",
+] }
+opentelemetry-semantic-conventions = { version = "0.31.0", features = [
+    "semconv_experimental",
+] }
 opentelemetry-stdout = "0.31.0"
-opentelemetry_sdk = { version = "0.31.0", default-features = false, features = ["trace", "rt-tokio"] }
+opentelemetry_sdk = { version = "0.31.0", default-features = false, features = [
+    "trace",
+    "rt-tokio",
+] }
 parking_lot = "0.12.3"
 percent-encoding = "2.3"
 pin-project = "1.1.10"
@@ -121,7 +138,14 @@ pyo3-stub-gen = "0.14"
 rand = "0.9.0"
 regex = "1.11.1"
 
-reqwest = { version = "0.12", features = ["json", "rustls-tls-no-provider", "charset", "http2", "system-proxy", "blocking"], default-features = false }
+reqwest = { version = "0.12", features = [
+    "json",
+    "rustls-tls-no-provider",
+    "charset",
+    "http2",
+    "system-proxy",
+    "blocking",
+], default-features = false }
 rlimit = "0.11.0"
 rustls = { version = "0.23.31" }
 rustls-native-certs = "0.8"

data-plane/Cargo.toml

@@ -112,7 +112,7 @@ static async Task RunModerator(SlimApp app, ulong connId, string remote,
         var config = new SlimSessionConfig
         {
             SessionType = SlimSessionType.Group,
-            EnableMls = enableMls,
+            MlsSettings = enableMls ? new SlimMlsSettings() : null,
             MaxRetries = 5,
             RetryInterval = TimeSpan.FromSeconds(5),
             Metadata = new Dictionary<string, string>()

data-plane/bindings/dotnet/Slim.Examples.Group/Program.cs

@@ -105,7 +105,7 @@ static async Task RunSender(SlimApp app, ulong connId, string remote, string mes
         var config = new SlimSessionConfig
         {
             SessionType = SlimSessionType.PointToPoint,
-            EnableMls = enableMls
+            MlsSettings = enableMls ? new SlimMlsSettings() : null
         };
 
         Console.WriteLine($"[{instance}] Creating session to {remote}...");

data-plane/bindings/dotnet/Slim.Examples.PointToPoint/Program.cs

@@ -121,25 +121,23 @@ public void SessionConfig_DefaultValues()
     {
         var config = new SlimSessionConfig
         {
-            SessionType = SlimSessionType.PointToPoint,
-            EnableMls = false
+            SessionType = SlimSessionType.PointToPoint
         };
 
         Assert.Equal(SlimSessionType.PointToPoint, config.SessionType);
-        Assert.False(config.EnableMls);
+        Assert.Null(config.MlsSettings);
     }
 
     [Fact]
     public void SessionConfig_GroupType()
     {
         var config = new SlimSessionConfig
         {
-            SessionType = SlimSessionType.Group,
-            EnableMls = false
+            SessionType = SlimSessionType.Group
         };
 
         Assert.Equal(SlimSessionType.Group, config.SessionType);
-        Assert.False(config.EnableMls);
+        Assert.Null(config.MlsSettings);
     }
 
     [Fact]
@@ -148,10 +146,10 @@ public void SessionConfig_WithMls()
         var config = new SlimSessionConfig
         {
             SessionType = SlimSessionType.PointToPoint,
-            EnableMls = true
+            MlsSettings = new SlimMlsSettings()
         };
 
-        Assert.True(config.EnableMls);
+        Assert.NotNull(config.MlsSettings);
     }
 
     [Fact]

data-plane/bindings/dotnet/Slim.Tests/SlimTests.cs

@@ -212,6 +212,18 @@ public enum SlimSessionType
     Group
 }
 
+/// <summary>
+/// MLS-specific settings for a session.
+/// </summary>
+public sealed class SlimMlsSettings
+{
+    /// <summary>0 = disable header-integrity checks; 1–100 = percent of messages to verify after decrypt.</summary>
+    public uint HeaderIntegrityValidationPercent { get; init; } = 100;
+
+    internal Internal.MlsSettings ToInternal() =>
+        new(HeaderIntegrityValidationPercent);
+}
+
 /// <summary>
 /// Configuration for creating a SLIM session.
 /// </summary>
@@ -220,9 +232,6 @@ public sealed class SlimSessionConfig
     /// <summary>Session type (PointToPoint or Group).</summary>
     public SlimSessionType SessionType { get; init; } = SlimSessionType.PointToPoint;
 
-    /// <summary>Whether to enable MLS encryption.</summary>
-    public bool EnableMls { get; init; }
-
     /// <summary>Maximum retry attempts for session establishment.</summary>
     public uint MaxRetries { get; init; } = 5;
 
@@ -232,16 +241,19 @@ public sealed class SlimSessionConfig
     /// <summary>Optional metadata key-value pairs.</summary>
     public Dictionary<string, string>? Metadata { get; init; }
 
+    /// <summary>MLS options (None disables MLS).</summary>
+    public SlimMlsSettings? MlsSettings { get; init; }
+
     internal Internal.SessionConfig ToInternal()
     {
         return new Internal.SessionConfig(
             SessionType: SessionType == SlimSessionType.PointToPoint
                 ? Internal.SessionType.PointToPoint
                 : Internal.SessionType.Group,
-            EnableMls: EnableMls,
             MaxRetries: MaxRetries,
             Interval: RetryInterval,
-            Metadata: Metadata ?? new Dictionary<string, string>()
+            Metadata: Metadata ?? new Dictionary<string, string>(),
+            MlsSettings: MlsSettings?.ToInternal()
         );
     }
 }

data-plane/bindings/dotnet/Slim/Slim.cs

@@ -82,12 +82,17 @@ func runModerator(app *slim.App, connID uint64, remote string, invites []string,
 
 	// Create multicast session
 	interval := time.Second * 5
+	var mlsSettings *slim.MlsSettings
+	if enableMLS {
+		mlsSettings = &slim.MlsSettings{HeaderIntegrityValidationPercent: 100}
+	}
+
 	config := slim.SessionConfig{
 		SessionType: slim.SessionTypeGroup,
-		EnableMls:   enableMLS,
 		MaxRetries:  &[]uint32{5}[0], // 5 retries
 		Interval:    &interval,
 		Metadata:    make(map[string]string),
+		MlsSettings: mlsSettings,
 	}
 
 	session, err := app.CreateSessionAndWaitAsync(config, channelName)

data-plane/bindings/go/examples/group/main.go

@@ -64,9 +64,14 @@ func runSender(app *slim.App, connID uint64, remote, message string, iterations
 	}
 	fmt.Printf("[%s] 📍 Route set to %s via connection %d\n", instance, remote, connID)
 
+	var mlsSettings *slim.MlsSettings
+	if enableMLS {
+		mlsSettings = &slim.MlsSettings{HeaderIntegrityValidationPercent: 100}
+	}
+
 	config := slim.SessionConfig{
 		SessionType: slim.SessionTypePointToPoint,
-		EnableMls:   enableMLS,
+		MlsSettings: mlsSettings,
 	}
 
 	fmt.Printf("[%s] 🔍 Creating session to %s...\n", instance, remote)

data-plane/bindings/go/examples/point_to_point/main.go

@@ -399,7 +399,7 @@ func TestGroupSession(t *testing.T) {
 
 	sessionConfig := slim.SessionConfig{
 		SessionType: slim.SessionTypeGroup,
-		EnableMls:   false,
+		MlsSettings: nil,
 	}
 
 	destination := slim.NewName("org", "group", "v1")
@@ -420,7 +420,7 @@ func TestSessionInviteRemove(t *testing.T) {
 
 	sessionConfig := slim.SessionConfig{
 		SessionType: slim.SessionTypeGroup,
-		EnableMls:   false,
+		MlsSettings: nil,
 	}
 
 	destination := slim.NewName("org", "group", "v1")
@@ -515,7 +515,7 @@ func TestSessionCreationAutoWait(t *testing.T) {
 
 	sessionConfig := slim.SessionConfig{
 		SessionType: slim.SessionTypePointToPoint,
-		EnableMls:   false,
+		MlsSettings: nil,
 	}
 
 	// CreateSession should auto-wait for session establishment
@@ -540,7 +540,7 @@ func TestInviteAutoWait(t *testing.T) {
 
 	sessionConfig := slim.SessionConfig{
 		SessionType: slim.SessionTypeGroup,
-		EnableMls:   false,
+		MlsSettings: nil,
 	}
 
 	destination := slim.NewName("org", "group", "v1")
@@ -574,7 +574,7 @@ func TestRemoveAutoWait(t *testing.T) {
 
 	sessionConfig := slim.SessionConfig{
 		SessionType: slim.SessionTypeGroup,
-		EnableMls:   false,
+		MlsSettings: nil,
 	}
 
 	destination := slim.NewName("org", "group", "v1")
@@ -726,7 +726,7 @@ func BenchmarkPublishFireAndForget(b *testing.B) {
 	defer app.Destroy()
 
 	session, err := app.CreateSession(
-		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, EnableMls: false},
+		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, MlsSettings: nil},
 		slim.NewName("org", "receiver", "v1"),
 	)
 	if err != nil {
@@ -754,7 +754,7 @@ func BenchmarkPublishWithCompletion(b *testing.B) {
 	defer app.Destroy()
 
 	session, err := app.CreateSession(
-		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, EnableMls: false},
+		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, MlsSettings: nil},
 		slim.NewName("org", "receiver", "v1"),
 	)
 	if err != nil {
@@ -785,7 +785,7 @@ func BenchmarkPublishWithCompletionAndWait(b *testing.B) {
 	defer app.Destroy()
 
 	session, err := app.CreateSession(
-		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, EnableMls: false},
+		slim.SessionConfig{SessionType: slim.SessionTypePointToPoint, MlsSettings: nil},
 		slim.NewName("org", "receiver", "v1"),
 	)
 	if err != nil {