Add tests for DMAPI interop communication limits

src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeResponse.cs

@@ -3,7 +3,7 @@
 	/// <summary>
 	/// A response to a bridge request.
 	/// </summary>
-	public sealed class BridgeResponse : DMApiResponse
+	public class BridgeResponse : DMApiResponse
 	{
 		/// <summary>
 		/// The new port for <see cref="BridgeCommandType.Reboot"/> requests.

tests/DMAPI/LongRunning/Config.dm

@@ -6,6 +6,6 @@
 #define TGS_WORLD_ANNOUNCE(message) world << ##message
 #define TGS_INFO_LOG(message) world.log << "Info: [##message]"
 #define TGS_WARNING_LOG(message) world.log << "Warn: [##message]"
-#define TGS_ERROR_LOG(message) world.log << "Err: [##message]"
+#define TGS_ERROR_LOG(message) TgsError(##message)
 #define TGS_NOTIFY_ADMINS(event)
 #define TGS_CLIENT_COUNT 0

tests/DMAPI/LongRunning/Test.dm

@@ -49,11 +49,18 @@
 	world.TgsChatBroadcast(response)
 	world.TgsInitializationComplete()
 
+	startup_complete = TRUE
+	if(run_bridge_test)
+		CheckBridgeLimits()
+
 /world/Topic(T, Addr, Master, Keys)
 	log << "Topic: [T]"
 	. =  HandleTopic(T)
 	log << "Response: [.]"
 
+var/startup_complete
+var/run_bridge_test
+
 /world/proc/HandleTopic(T)
 	TGS_TOPIC
 
@@ -63,6 +70,40 @@
 		RebootAsync()
 		return "ack"
 
+	var/tactics2 = data["tgs_integration_test_tactics2"]
+	if(tactics2)
+		if(startup_complete)
+			CheckBridgeLimits()
+		else
+			run_bridge_test = TRUE
+		return "ack2"
+
+	// Topic limit tests
+	// Receive
+	var/tactics3 = data["tgs_integration_test_tactics3"]
+	if(tactics3)
+		var/list/json = json_decode(tactics3)
+		if(!json || !istext(json["payload"]) || !istext(json["size"]))
+			return "fail"
+
+		var/size = text2num(json["size"])
+		var/payload = json["payload"]
+		if(length(payload) != size)
+			return "fail"
+
+		return "pass"
+
+	// Send
+	var/tactics4 = data["tgs_integration_test_tactics4"]
+	if(tactics4)
+		var/size = isnum(tactics4) ? tactics4 : text2num(tactics4)
+		if(!isnum(size))
+			text2file("tgs_integration_test_tactics4 wasn't a number!", "test_fail_reason.txt")
+			del(world)
+
+		var/payload = create_payload(size)
+		return payload
+
 	TgsChatBroadcast("Recieved non-tgs topic: [T]")
 
 	return "feck"
@@ -77,7 +118,8 @@
 	world.TgsChatBroadcast("Recieved event: `[json_encode(args)]`")
 
 /world/Export(url)
-	log << "Export: [url]"
+	if(length(url) < 1000)
+		log << "Export: [url]"
 	return ..()
 
 /proc/RebootAsync()
@@ -113,3 +155,60 @@
 	response.embed.fields += field
 	response.embed.footer = new /datum/tgs_chat_embed/footer("Footer text")
 	return response
+
+var/lastTgsError
+
+/proc/TgsError(message)
+	world.log << "Err: [message]"
+	lastTgsError = message
+
+/proc/create_payload(size)
+	var/builder = list()
+	for(var/j = 0; j < size; ++j)
+		builder += "a"
+	var/payload = jointext(builder, "")
+	return payload
+	
+/proc/CheckBridgeLimits()
+	set waitfor = FALSE
+	CheckBridgeLimitsImpl()
+
+/proc/CheckBridgeLimitsImpl()
+	sleep(30)
+
+	// Evil custom bridge command hacking here
+	var/datum/tgs_api/v5/api = TGS_READ_GLOBAL(tgs)
+	var/old_ai = api.access_identifier
+	api.access_identifier = "tgs_integration_test"
+
+	// Always send chat messages because they can have extremely large payloads with the text
+
+	// bisecting request test
+	var/base = 1
+	var/nextPow = 0
+	var/lastI = 0
+	var/i
+	lastTgsError = null
+	for(i = 1; ; i = base + (2 ** nextPow))
+		var/payload = create_payload(i)
+		
+		var/list/result = api.Bridge(0, list("chatMessage" = list("text" = "payload:[payload]")))
+
+		if(!result || lastTgsError || result["integrationHack"] != "ok")
+			if(i == lastI + 1)
+				break
+			lastTgsError = null
+			i = lastI
+			base = lastI
+			nextPow = 0
+			continue
+
+		lastI = i
+		++nextPow
+
+	var/finalResult = api.Bridge(0, list("chatMessage" = list("text" = "done")))
+	if(!finalResult || lastTgsError || finalResult["integrationHack"] != "ok")
+		text2file("Failed to end bridge limit test!", "test_fail_reason.txt")
+		del(world)
+
+	api.access_identifier = old_ai

tests/Tgstation.Server.Tests/Instance/InstanceTest.cs

@@ -12,13 +12,15 @@ namespace Tgstation.Server.Tests.Instance
 	{
 		readonly IInstanceClient instanceClient;
 		readonly IInstanceManagerClient instanceManagerClient;
-		readonly IInstanceManager instanceManager;
+		readonly InstanceManager instanceManager;
+		readonly ushort serverPort;
 
-		public InstanceTest(IInstanceClient instanceClient, IInstanceManagerClient instanceManagerClient, IInstanceManager instanceManager)
+		public InstanceTest(IInstanceClient instanceClient, IInstanceManagerClient instanceManagerClient, InstanceManager instanceManager, ushort serverPort)
 		{
 			this.instanceClient = instanceClient ?? throw new ArgumentNullException(nameof(instanceClient));
 			this.instanceManagerClient = instanceManagerClient ?? throw new ArgumentNullException(nameof(instanceManagerClient));
 			this.instanceManager = instanceManager ?? throw new ArgumentNullException(nameof(instanceManager));
+			this.serverPort = serverPort;
 		}
 
 		public async Task RunTests(CancellationToken cancellationToken)
@@ -38,7 +40,7 @@ namespace Tgstation.Server.Tests.Instance
 			await configTest.Run(cancellationToken);
 			await chatTests;
 			await repoTests;
-			await new WatchdogTest(instanceClient, instanceManager).Run(cancellationToken);
+			await new WatchdogTest(instanceClient, instanceManager, serverPort).Run(cancellationToken);
 		}
 	}
 }

tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs

@@ -4,6 +4,8 @@ using Microsoft.Extensions.Logging;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using Moq;
 
+using Newtonsoft.Json;
+
 using System;
 using System.Globalization;
 using System.IO;
@@ -13,6 +15,7 @@ using System.Net.Sockets;
 using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Web;
 
 using Tgstation.Server.Api;
 using Tgstation.Server.Api.Models;
@@ -22,25 +25,29 @@ using Tgstation.Server.Client;
 using Tgstation.Server.Client.Components;
 using Tgstation.Server.Host.Components;
 using Tgstation.Server.Host.Components.Interop;
+using Tgstation.Server.Host.Components.Interop.Bridge;
 using Tgstation.Server.Host.Components.Watchdog;
+using Tgstation.Server.Host.Controllers;
 using Tgstation.Server.Host.Extensions;
 using Tgstation.Server.Host.IO;
 using Tgstation.Server.Host.System;
 
 namespace Tgstation.Server.Tests.Instance
 {
-	sealed class WatchdogTest : JobsRequiredTest
+	sealed class WatchdogTest : JobsRequiredTest, IBridgeHandler
 	{
 		readonly IInstanceClient instanceClient;
-		readonly IInstanceManager instanceManager;
+		readonly InstanceManager instanceManager;
+		readonly ushort serverPort;
 
 		bool ranTimeoutTest = false;
 
-		public WatchdogTest(IInstanceClient instanceClient, IInstanceManager instanceManager)
+		public WatchdogTest(IInstanceClient instanceClient, InstanceManager instanceManager, ushort serverPort)
 			: base(instanceClient.Jobs)
 		{
 			this.instanceClient = instanceClient ?? throw new ArgumentNullException(nameof(instanceClient));
 			this.instanceManager = instanceManager ?? throw new ArgumentNullException(nameof(instanceManager));
+			this.serverPort = serverPort;
 		}
 
 		public async Task Run(CancellationToken cancellationToken)
@@ -75,6 +82,7 @@ namespace Tgstation.Server.Tests.Instance
 			await RunLongRunningTestThenUpdate(cancellationToken);
 
 			await GhettoChatCommandTest(cancellationToken);
+			await GhettoValidateDMApiLimits(cancellationToken);
 
 			await RunLongRunningTestThenUpdateWithNewDme(cancellationToken);
 			await RunLongRunningTestThenUpdateWithByondVersionSwitch(cancellationToken);
@@ -242,7 +250,7 @@ namespace Tgstation.Server.Tests.Instance
 			using var ddProc = ddProcs.Single();
 			IProcessExecutor executor = null;
 			executor = new ProcessExecutor(
-				RuntimeInformation.IsOSPlatform(OSPlatform.Windows)
+                System.Runtime.InteropServices.RuntimeInformation.IsOSPlatform(OSPlatform.Windows)
 					? new WindowsProcessFeatures(Mock.Of<ILogger<WindowsProcessFeatures>>())
 					: new PosixProcessFeatures(new Lazy<IProcessExecutor>(() => executor), Mock.Of<IIOManager>(), Mock.Of<ILogger<PosixProcessFeatures>>()),
 				Mock.Of<IIOManager>(),
@@ -328,6 +336,198 @@ namespace Tgstation.Server.Tests.Instance
 			return await instanceClient.DreamDaemon.Start(cancellationToken);
 		}
 
+		class DMApiParametersImpl : DMApiParameters { }
+		public DMApiParameters DMApiParameters => new DMApiParametersImpl
+		{
+			AccessIdentifier = "tgs_integration_test"
+		};
+
+		TaskCompletionSource bridgeTestsTcs;
+
+		public class BridgeResponseHack : BridgeResponse
+		{
+			public string IntegrationHack { get; set; }
+		}
+
+		public class ResponseTestData : TestData
+		{
+			public bool Continue { get; set; }
+			public string PayloadId { get; set; }
+		}
+
+		bool bridgeStageResponse;
+		long lastBridgeRequestSize = 0;
+
+		uint lastPayloadId = 0;
+
+		public Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
+		{
+			try
+			{
+				Assert.AreEqual(DMApiParameters.AccessIdentifier, parameters.AccessIdentifier);
+				Assert.AreEqual((BridgeCommandType)0, parameters.CommandType);
+				Assert.IsNotNull(parameters.ChatMessage?.Text);
+				var splits = parameters.ChatMessage.Text.Split(':', StringSplitOptions.RemoveEmptyEntries);
+				var coreMessage = splits[0];
+				Assert.IsFalse(String.IsNullOrWhiteSpace(coreMessage));
+				if (!bridgeStageResponse)
+					if (coreMessage == "done")
+					{
+						Assert.AreEqual(DMApiConstants.MaximumBridgeRequestLength, lastBridgeRequestSize);
+
+						bridgeTestsTcs.SetResult();
+						return Task.FromResult<BridgeResponse>(
+							new BridgeResponseHack
+							{
+								IntegrationHack = "ok"
+							});
+					}
+
+				Assert.AreEqual("payload", coreMessage);
+				lastBridgeRequestSize = $"http://127.0.0.1:{serverPort}/Bridge?data=".Length + HttpUtility.UrlEncode(
+					JsonConvert.SerializeObject(parameters, DMApiConstants.SerializerSettings)).Length;
+				return Task.FromResult<BridgeResponse>(
+					new BridgeResponseHack
+					{
+						IntegrationHack = "ok"
+					});
+			}
+			catch (Exception ex)
+			{
+				bridgeTestsTcs.SetException(ex);
+				return Task.FromResult<BridgeResponse>(null);
+			}
+		}
+
+		public class TestData
+		{
+			public string Size { get; set; }
+			public string Payload { get; set; }
+		}
+
+		async Task GhettoValidateDMApiLimits(CancellationToken cancellationToken)
+		{
+			var startJob = await StartDD(cancellationToken);
+
+			await WaitForJob(startJob, 40, false, null, cancellationToken);
+
+			// first check the bridge limits
+			bridgeTestsTcs = new TaskCompletionSource();
+			BridgeController.LogContent = false;
+			using (var bridgeRegistration = instanceManager.RegisterHandler(this))
+			{
+				System.Console.WriteLine("TEST: Sending Bridge tests topic...");
+				var bridgeTestTopicResult = await topicClient.SendTopic(IPAddress.Loopback, "tgs_integration_test_tactics2=1", IntegrationTest.DDPort, cancellationToken);
+				Assert.AreEqual("ack2", bridgeTestTopicResult.StringData);
+
+				using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+				// this test doesn't take long
+				cts.CancelAfter(TimeSpan.FromMinutes(10));
+				await bridgeTestsTcs.Task.WithToken(cts.Token);
+			}
+
+			BridgeController.LogContent = true;
+
+			// Time for DD to revert the bridge access identifier change
+			await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken);
+
+			// Time for topic tests
+			// Request
+
+			System.Console.WriteLine("TEST: Sending Topic tests topics...");
+
+			var nextPow = 0;
+			var lastSize = 0;
+
+			var baseTopic = new TestData
+			{
+				Size = 0.ToString().PadLeft(6, '0'),
+				Payload = "",
+			};
+
+			var json = JsonConvert.SerializeObject(baseTopic, DMApiConstants.SerializerSettings);
+			var topicString = $"tgs_integration_test_tactics3={topicClient.SanitizeString(json)}";
+
+			var baseSize = topicString.Length;
+			var wrappingSize = baseSize;
+
+			while (!cancellationToken.IsCancellationRequested)
+			{
+				var currentSize = baseSize + (int)Math.Pow(2, nextPow);
+				var payloadSize = currentSize - wrappingSize;
+
+				var topic = new TestData
+				{
+					Size = payloadSize.ToString().PadLeft(6, '0'),
+					Payload = new string('a', payloadSize),
+				};
+
+				TopicResponse topicRequestResult = null;
+				try
+				{
+					topicRequestResult = await topicClient.SendTopic(
+						IPAddress.Loopback,
+						$"tgs_integration_test_tactics3={topicClient.SanitizeString(JsonConvert.SerializeObject(topic, DMApiConstants.SerializerSettings))}",
+						IntegrationTest.DDPort,
+						cancellationToken);
+				}
+				catch (ArgumentOutOfRangeException)
+				{
+				}
+
+				if (topicRequestResult == null
+					|| topicRequestResult.ResponseType != TopicResponseType.StringResponse
+					|| topicRequestResult.StringData != "pass")
+				{
+					if (topicRequestResult != null)
+						Assert.AreEqual("fail", topicRequestResult.StringData);
+					if (currentSize == lastSize + 1)
+						break;
+					baseSize = lastSize;
+					nextPow = 0;
+					continue;
+				}
+
+				lastSize = currentSize;
+				++nextPow;
+			}
+
+			cancellationToken.ThrowIfCancellationRequested();
+
+			Assert.AreEqual(DMApiConstants.MaximumTopicRequestLength, (uint)lastSize);
+
+			// Receive
+			baseSize = 1;
+			nextPow = 0;
+			lastSize = 0;
+			while (!cancellationToken.IsCancellationRequested)
+			{
+				var currentSize = baseSize + (int)Math.Pow(2, nextPow);
+				var topicRequestResult = await topicClient.SendTopic(
+					IPAddress.Loopback,
+					$"tgs_integration_test_tactics4={topicClient.SanitizeString(currentSize.ToString())}",
+					IntegrationTest.DDPort,
+					cancellationToken);
+
+				if (topicRequestResult.ResponseType != TopicResponseType.StringResponse
+					|| new string('a', currentSize) != topicRequestResult.StringData)
+				{
+					if (currentSize == lastSize + 1)
+						break;
+					baseSize = lastSize;
+					nextPow = 0;
+					continue;
+				}
+
+				lastSize = currentSize;
+				++nextPow;
+			}
+
+			cancellationToken.ThrowIfCancellationRequested();
+			Assert.AreEqual(DMApiConstants.MaximumTopicResponseLength, (uint)lastSize);
+			await instanceClient.DreamDaemon.Shutdown(cancellationToken);
+		}
+
 		async Task GhettoChatCommandTest(CancellationToken cancellationToken)
 		{
 			var startJob = await StartDD(cancellationToken);
@@ -581,23 +781,23 @@ namespace Tgstation.Server.Tests.Instance
 			return ddProc != null;
 		}
 
+		TopicClient topicClient = new (new SocketParameters
+		{
+			SendTimeout = TimeSpan.FromSeconds(30),
+			ReceiveTimeout = TimeSpan.FromSeconds(30),
+			ConnectTimeout = TimeSpan.FromSeconds(30),
+			DisconnectTimeout = TimeSpan.FromSeconds(30)
+		});
+
 		public async Task<DreamDaemonResponse> TellWorldToReboot(CancellationToken cancellationToken)
 		{
 			var daemonStatus = await instanceClient.DreamDaemon.Read(cancellationToken);
 			var initialCompileJob = daemonStatus.ActiveCompileJob;
 
-			var bts = new TopicClient(new SocketParameters
-			{
-				SendTimeout = TimeSpan.FromSeconds(30),
-				ReceiveTimeout = TimeSpan.FromSeconds(30),
-				ConnectTimeout = TimeSpan.FromSeconds(30),
-				DisconnectTimeout = TimeSpan.FromSeconds(30)
-			});
-
 			try
 			{
 				System.Console.WriteLine("TEST: Sending world reboot topic...");
-				var result = await bts.SendTopic(IPAddress.Loopback, "tgs_integration_test_special_tactics=1", IntegrationTest.DDPort, cancellationToken);
+				var result = await topicClient.SendTopic(IPAddress.Loopback, "tgs_integration_test_special_tactics=1", IntegrationTest.DDPort, cancellationToken);
 				Assert.AreEqual("ack", result.StringData);
 
 				using (var tempCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))

tests/Tgstation.Server.Tests/IntegrationTest.cs

@@ -829,7 +829,7 @@ namespace Tgstation.Server.Tests
 
 			TerminateAllDDs();
 
-			IInstanceManager GetInstanceManager() => ((Host.Server)server.RealServer).Host.Services.GetRequiredService<IInstanceManager>();
+			InstanceManager GetInstanceManager() => ((Host.Server)server.RealServer).Host.Services.GetRequiredService<InstanceManager>();
 
 			// main run
 			var serverTask = server.Run(cancellationToken);
@@ -878,7 +878,7 @@ namespace Tgstation.Server.Tests
 
 					Assert.IsTrue(Directory.Exists(instanceClient.Metadata.Path));
 
-					var instanceTests = FailFast(new InstanceTest(instanceClient, adminClient.Instances, GetInstanceManager()).RunTests(cancellationToken));
+					var instanceTests = FailFast(new InstanceTest(instanceClient, adminClient.Instances, GetInstanceManager(), (ushort)server.Url.Port).RunTests(cancellationToken));
 
 					await Task.WhenAll(rootTest, adminTest, instanceTests, usersTest);
 
@@ -998,7 +998,7 @@ namespace Tgstation.Server.Tests
 					Assert.AreEqual(WatchdogStatus.Online, dd.Status.Value);
 
 					var compileJob = await instanceClient.DreamMaker.Compile(cancellationToken);
-					var wdt = new WatchdogTest(instanceClient, GetInstanceManager());
+					var wdt = new WatchdogTest(instanceClient, GetInstanceManager(), (ushort)server.Url.Port);
 					await wdt.WaitForJob(compileJob, 30, false, null, cancellationToken);
 
 					dd = await instanceClient.DreamDaemon.Read(cancellationToken);
@@ -1045,7 +1045,7 @@ namespace Tgstation.Server.Tests
 					Assert.AreEqual(WatchdogStatus.Online, currentDD.Status);
 					Assert.AreEqual(expectedStaged, currentDD.StagedCompileJob.Job.Id.Value);
 
-					var wdt = new WatchdogTest(instanceClient, GetInstanceManager());
+					var wdt = new WatchdogTest(instanceClient, GetInstanceManager(), (ushort)server.Url.Port);
 					currentDD = await wdt.TellWorldToReboot(cancellationToken);
 					Assert.AreEqual(expectedStaged, currentDD.ActiveCompileJob.Job.Id.Value);
 					Assert.IsNull(currentDD.StagedCompileJob);