mirror of
https://github.com/Stone-Red-Code/RemoteExec.git
synced 2026-09-04 23:42:03 +02:00
Move code to src folder
This commit is contained in:
Stone_Red
@@ -0,0 +1,23 @@
|
||||
namespace RemoteExec.Server.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// Represents an individual API key configuration.
|
||||
/// </summary>
|
||||
public class ApiKeyConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the API key value.
|
||||
/// </summary>
|
||||
public required string Key { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional description of the API key.
|
||||
/// </summary>
|
||||
public string? Description { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets whether this API key is enabled.
|
||||
/// Default is true.
|
||||
/// </summary>
|
||||
public bool Enabled { get; set; } = true;
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
namespace RemoteExec.Server.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// Configuration for API key authentication.
|
||||
/// </summary>
|
||||
public class AuthenticationConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the list of configured API keys.
|
||||
/// </summary>
|
||||
public List<ApiKeyConfiguration> ApiKeys { get; set; } = [];
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
namespace RemoteExec.Server.Configuration;
|
||||
|
||||
public class ExecutionConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Maximum number of concurrent tasks. Default is ProcessorCount * 2.
|
||||
/// </summary>
|
||||
public int? MaxConcurrentTasks { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Timeout for assembly loading requests in seconds. Default is 30 seconds.
|
||||
/// </summary>
|
||||
public int AssemblyLoadTimeoutSeconds { get; set; } = 30;
|
||||
|
||||
/// <summary>
|
||||
/// Type of execution environment to use. Default is "AssemblyLoadContext".
|
||||
/// Options: "AssemblyLoadContext", "DockerContainer"
|
||||
/// </summary>
|
||||
public string ExecutionEnvironment { get; set; } = "AssemblyLoadContext";
|
||||
}
|
||||
|
||||
public class DockerExecutionConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Docker host URL. Default is unix:///var/run/docker.sock (Linux) or npipe://./pipe/docker_engine (Windows).
|
||||
/// </summary>
|
||||
public string DockerHost { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>
|
||||
/// Docker worker image name. Default is remoteexec-worker:latest.
|
||||
/// </summary>
|
||||
public string WorkerImageName { get; set; } = "remoteexec-worker:latest";
|
||||
|
||||
/// <summary>
|
||||
/// Container timeout in seconds. Default is 300 (5 minutes).
|
||||
/// </summary>
|
||||
public int ContainerTimeoutSeconds { get; set; } = 300;
|
||||
|
||||
/// <summary>
|
||||
/// Memory limit per container in MB. Default is 512 MB.
|
||||
/// </summary>
|
||||
public long ContainerMemoryLimitMb { get; set; } = 512;
|
||||
|
||||
/// <summary>
|
||||
/// CPU shares per container. Default is 1024.
|
||||
/// </summary>
|
||||
public long ContainerCpuShares { get; set; } = 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Disable network access in containers. Default is true.
|
||||
/// </summary>
|
||||
public bool DisableNetwork { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Use read-only filesystem in containers. Default is true.
|
||||
/// </summary>
|
||||
public bool ReadOnlyFilesystem { get; set; } = true;
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
namespace RemoteExec.Server.Configuration;
|
||||
|
||||
public class MetricsConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Minimum CPU usage difference (in percentage points) to trigger a metrics broadcast. Default is 1.0.
|
||||
/// </summary>
|
||||
public double CpuDifferenceThreshold { get; set; } = 1.0;
|
||||
|
||||
/// <summary>
|
||||
/// Minimum memory usage difference (in bytes) to trigger a metrics broadcast. Default is 10 MB.
|
||||
/// </summary>
|
||||
public long MemoryDifferenceThreshold { get; set; } = 10 * 1024 * 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Broadcast interval in milliseconds. Default is 500 ms.
|
||||
/// </summary>
|
||||
public int BroadcastIntervalMs { get; set; } = 500;
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
using RemoteExec.Server.Hubs;
|
||||
|
||||
namespace RemoteExec.Server.Controllers;
|
||||
|
||||
/// <summary>
|
||||
/// Controller for handling assembly upload requests from clients.
|
||||
/// </summary>
|
||||
[ApiController]
|
||||
[Route("/")]
|
||||
public class AssemblyController(ILogger<AssemblyController> logger) : ControllerBase
|
||||
{
|
||||
/// <summary>
|
||||
/// Receives an assembly binary from a client in response to an assembly request.
|
||||
/// </summary>
|
||||
/// <param name="requestId">The unique identifier for the assembly request.</param>
|
||||
/// <returns>An action result indicating success or failure.</returns>
|
||||
[HttpPost("provide-assembly")]
|
||||
public async Task<IActionResult> ProvideAssembly([FromQuery] Guid requestId)
|
||||
{
|
||||
try
|
||||
{
|
||||
using MemoryStream ms = new MemoryStream();
|
||||
await Request.Body.CopyToAsync(ms);
|
||||
byte[] assemblyBytes = ms.ToArray();
|
||||
|
||||
logger.LogInformation("Received assembly for request {RequestId}, size: {Size} bytes", requestId, assemblyBytes.Length);
|
||||
|
||||
await RemoteExecutionHub.ProvideAssembly(requestId, assemblyBytes);
|
||||
|
||||
return Ok();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error providing assembly for request {RequestId}", requestId);
|
||||
return StatusCode(500, ex.Message);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
# See https://aka.ms/customizecontainer to learn how to customize your debug container and how Visual Studio uses this Dockerfile to build your images for faster debugging.
|
||||
|
||||
# This stage is used when running from VS in fast mode (Default for Debug configuration)
|
||||
FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS base
|
||||
USER $APP_UID
|
||||
WORKDIR /app
|
||||
EXPOSE 8080
|
||||
EXPOSE 8081
|
||||
|
||||
|
||||
# This stage is used to build the service project
|
||||
FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
|
||||
ARG BUILD_CONFIGURATION=Release
|
||||
WORKDIR /src
|
||||
COPY ["RemoteExec.Server/RemoteExec.Server.csproj", "RemoteExec.Server/"]
|
||||
RUN dotnet restore "./RemoteExec.Server/RemoteExec.Server.csproj"
|
||||
COPY . .
|
||||
WORKDIR "/src/RemoteExec.Server"
|
||||
RUN dotnet build "./RemoteExec.Server.csproj" -c $BUILD_CONFIGURATION -o /app/build
|
||||
|
||||
# This stage is used to publish the service project to be copied to the final stage
|
||||
FROM build AS publish
|
||||
ARG BUILD_CONFIGURATION=Release
|
||||
RUN dotnet publish "./RemoteExec.Server.csproj" -c $BUILD_CONFIGURATION -o /app/publish /p:UseAppHost=false
|
||||
|
||||
# This stage is used in production or when running from VS in regular mode (Default when not using the Debug configuration)
|
||||
FROM base AS final
|
||||
WORKDIR /app
|
||||
COPY --from=publish /app/publish .
|
||||
ENTRYPOINT ["dotnet", "RemoteExec.Server.dll"]
|
||||
@@ -0,0 +1,12 @@
|
||||
// This file is used by Code Analysis to maintain SuppressMessage
|
||||
// attributes that are applied to this project.
|
||||
// Project-level suppressions either have no target or are given
|
||||
// a specific target and scoped to a namespace, type, member, etc.
|
||||
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
|
||||
[assembly: SuppressMessage("Minor Code Smell", "S2325:Methods and properties that don't access instance data should be static", Justification = "<Pending>", Scope = "type", Target = "~T:RemoteExec.Server.Hubs.RemoteExecutionHub")]
|
||||
[assembly: SuppressMessage("Major Code Smell", "S3011:Reflection should not be used to increase accessibility of classes, methods, or fields", Justification = "<Pending>", Scope = "member", Target = "~M:RemoteExec.Server.Hubs.RemoteExecutionHub.Execute(RemoteExec.Shared.RemoteExecutionRequest)~System.Threading.Tasks.Task{RemoteExec.Shared.RemoteExecutionResult}")]
|
||||
[assembly: SuppressMessage("Major Code Smell", "S3010:Static fields should not be updated in constructors", Justification = "<Pending>")]
|
||||
[assembly: SuppressMessage("Performance", "CA1873:Avoid potentially expensive logging", Justification = "Annoying")]
|
||||
[assembly: SuppressMessage("Major Code Smell", "S3011:Reflection should not be used to increase accessibility of classes, methods, or fields", Justification = "<Pending>", Scope = "member", Target = "~M:RemoteExec.Server.Services.AssemblyLoadContextExecutionEnvironment.ExecuteTaskAsync(RemoteExec.Shared.Models.RemoteExecutionRequest)~System.Threading.Tasks.Task{RemoteExec.Shared.Models.RemoteExecutionResult}")]
|
||||
@@ -0,0 +1,309 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
using RemoteExec.Server.Configuration;
|
||||
using RemoteExec.Server.Services;
|
||||
using RemoteExec.Shared.Models;
|
||||
|
||||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics;
|
||||
|
||||
namespace RemoteExec.Server.Hubs;
|
||||
|
||||
/// <summary>
|
||||
/// SignalR hub that handles remote method execution requests from clients.
|
||||
/// </summary>
|
||||
public class RemoteExecutionHub : Hub
|
||||
{
|
||||
private static readonly ConcurrentDictionary<string, ExecutionEnvironment> connections = new();
|
||||
|
||||
private static readonly ConcurrentDictionary<Guid, TaskCompletionSource<byte[]>> pendingAssemblyRequests = new();
|
||||
|
||||
// Track pending assembly requests per connection to avoid duplicate requests
|
||||
private static readonly ConcurrentDictionary<string, ConcurrentDictionary<string, Lazy<Task<byte[]>>>> pendingAssemblyRequestsByConnection = new();
|
||||
|
||||
private static ServerMetrics? lastMetrics;
|
||||
private static DateTime lastMetricsTimestamp;
|
||||
private static TimeSpan lastTotalProcessorTime;
|
||||
|
||||
private static int activeTasks = 0;
|
||||
private static int maxConcurrentTasks;
|
||||
private static SemaphoreSlim taskSemaphore = null!;
|
||||
|
||||
private static string? executionEnvironmentName;
|
||||
private static int assemblyLoadTimeoutSeconds;
|
||||
private static double cpuDifferenceThreshold;
|
||||
private static long memoryDifferenceThreshold;
|
||||
|
||||
private readonly ILogger<RemoteExecutionHub> logger;
|
||||
private readonly IEnumerable<ExecutionEnvironment> executionEnvironments;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="RemoteExecutionHub"/> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">The logger instance.</param>
|
||||
/// <param name="executionOptions">The execution configuration options.</param>
|
||||
/// <param name="metricsOptions">The metrics configuration options.</param>
|
||||
/// <param name="executionEnvironments">The available execution environments.</param>
|
||||
public RemoteExecutionHub(ILogger<RemoteExecutionHub> logger, IOptions<ExecutionConfiguration> executionOptions, IOptions<MetricsConfiguration> metricsOptions, IEnumerable<ExecutionEnvironment> executionEnvironments)
|
||||
{
|
||||
this.logger = logger;
|
||||
this.executionEnvironments = executionEnvironments;
|
||||
|
||||
// Initialize static configuration values once
|
||||
if (taskSemaphore is null)
|
||||
{
|
||||
maxConcurrentTasks = executionOptions.Value.MaxConcurrentTasks ?? (Environment.ProcessorCount * 2);
|
||||
taskSemaphore = new SemaphoreSlim(maxConcurrentTasks, maxConcurrentTasks);
|
||||
executionEnvironmentName = executionOptions.Value.ExecutionEnvironment;
|
||||
assemblyLoadTimeoutSeconds = executionOptions.Value.AssemblyLoadTimeoutSeconds;
|
||||
cpuDifferenceThreshold = metricsOptions.Value.CpuDifferenceThreshold;
|
||||
memoryDifferenceThreshold = metricsOptions.Value.MemoryDifferenceThreshold;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override async Task OnConnectedAsync()
|
||||
{
|
||||
ExecutionEnvironment? executionEnvironment = executionEnvironments.FirstOrDefault(env => env.Name.Equals(executionEnvironmentName, StringComparison.OrdinalIgnoreCase));
|
||||
|
||||
if (executionEnvironment is null)
|
||||
{
|
||||
logger.LogError("Execution environment '{ExecutionEnvironment}' not found for connection {ConnectionId}", executionEnvironmentName, Context.ConnectionId);
|
||||
throw new InvalidOperationException($"Execution environment '{executionEnvironmentName}' not found");
|
||||
}
|
||||
|
||||
// Capture Context and Clients to avoid accessing disposed Hub instance
|
||||
HubCallerContext capturedContext = Context;
|
||||
IHubCallerClients capturedClients = Clients;
|
||||
|
||||
executionEnvironment.RequestAssembly += async (sender, e) =>
|
||||
{
|
||||
byte[] assemblyBytes = await RequestAssemblyBytesAsync(e.Value, capturedContext, capturedClients);
|
||||
e.SetCompleted(assemblyBytes);
|
||||
};
|
||||
|
||||
await executionEnvironment.PrepareEnvironmentAsync(Context.ConnectionAborted);
|
||||
|
||||
_ = connections.TryAdd(Context.ConnectionId, executionEnvironment);
|
||||
_ = pendingAssemblyRequestsByConnection.TryAdd(Context.ConnectionId, new());
|
||||
|
||||
logger.LogInformation("Connection {ConnectionId} established", Context.ConnectionId);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override async Task OnDisconnectedAsync(Exception? exception)
|
||||
{
|
||||
if (connections.TryRemove(Context.ConnectionId, out ExecutionEnvironment? executionEnvironment))
|
||||
{
|
||||
await executionEnvironment.CleanupEnvironmentAsync(CancellationToken.None);
|
||||
logger.LogInformation("Connection {ConnectionId} disconnected", Context.ConnectionId);
|
||||
}
|
||||
|
||||
_ = pendingAssemblyRequestsByConnection.TryRemove(Context.ConnectionId, out _);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts processing a stream of tasks from the client.
|
||||
/// </summary>
|
||||
/// <param name="taskStream">The async enumerable stream of tasks to execute.</param>
|
||||
public async Task StartTaskStream(IAsyncEnumerable<TaskItem> taskStream)
|
||||
{
|
||||
logger.LogInformation("Starting task stream for connection {ConnectionId}", Context.ConnectionId);
|
||||
|
||||
try
|
||||
{
|
||||
await foreach (TaskItem taskItem in taskStream.WithCancellation(Context.ConnectionAborted))
|
||||
{
|
||||
await taskSemaphore.WaitAsync(Context.ConnectionAborted);
|
||||
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
RemoteExecutionResult result = await ExecuteTask(taskItem.Request);
|
||||
await Clients.Caller.SendAsync("TaskResult", taskItem.TaskId, result);
|
||||
}
|
||||
catch (ObjectDisposedException ex)
|
||||
{
|
||||
logger.LogWarning(ex, "Connection {ConnectionId} disposed while processing task {TaskId}", Context.ConnectionId, taskItem.TaskId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error processing task {TaskId}", taskItem.TaskId);
|
||||
RemoteExecutionResult errorResult = new RemoteExecutionResult
|
||||
{
|
||||
Exception = ex.ToString()
|
||||
};
|
||||
await Clients.Caller.SendAsync("TaskResult", taskItem.TaskId, errorResult);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_ = taskSemaphore.Release();
|
||||
}
|
||||
}, Context.ConnectionAborted);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException ex)
|
||||
{
|
||||
logger.LogInformation(ex, "Task stream for connection {ConnectionId} was canceled", Context.ConnectionId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error in task stream for connection {ConnectionId}", Context.ConnectionId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<RemoteExecutionResult> ExecuteTask(RemoteExecutionRequest request)
|
||||
{
|
||||
_ = Interlocked.Increment(ref activeTasks);
|
||||
|
||||
try
|
||||
{
|
||||
if (!connections.TryGetValue(Context.ConnectionId, out ExecutionEnvironment? executionEnvironment))
|
||||
{
|
||||
logger.LogError("Connection {ConnectionId} not found for executing method {Method} in type {Type}", Context.ConnectionId, request.MethodName, request.TypeName);
|
||||
throw new InvalidOperationException("Connection not found");
|
||||
}
|
||||
|
||||
return await executionEnvironment.ExecuteTaskAsync(request);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error executing remote method {Method} in type {Type} for connection {ConnectionId}", request.MethodName, request.TypeName, Context.ConnectionId);
|
||||
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Exception = ex.ToString()
|
||||
};
|
||||
}
|
||||
finally
|
||||
{
|
||||
_ = Interlocked.Decrement(ref activeTasks);
|
||||
}
|
||||
}
|
||||
|
||||
internal static async Task ProvideAssembly(Guid requestId, byte[] assemblyBytes)
|
||||
{
|
||||
if (pendingAssemblyRequests.TryRemove(requestId, out TaskCompletionSource<byte[]>? tcs))
|
||||
{
|
||||
tcs.SetResult(assemblyBytes);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the current server metrics.
|
||||
/// </summary>
|
||||
/// <returns>The current server metrics.</returns>
|
||||
public async Task<ServerMetrics> GetMetrics()
|
||||
{
|
||||
return await GetServerMetrics();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Broadcasts server metrics to all connected clients if metrics have changed significantly.
|
||||
/// </summary>
|
||||
/// <param name="hubContext">The hub context for broadcasting.</param>
|
||||
public static async Task BroadcastMetricsAsync(IHubContext<RemoteExecutionHub> hubContext)
|
||||
{
|
||||
ServerMetrics metrics = await GetServerMetrics();
|
||||
|
||||
// Only broadcast if metrics have changed by a significant amount
|
||||
if (lastMetrics is not null)
|
||||
{
|
||||
double cpuDiff = Math.Abs(metrics.CpuUsage - lastMetrics.CpuUsage);
|
||||
long memoryDiff = Math.Abs(metrics.TotalMemoryUsage - lastMetrics.TotalMemoryUsage);
|
||||
int connectionsDiff = Math.Abs(metrics.ActiveConnections - lastMetrics.ActiveConnections);
|
||||
int tasksDiff = Math.Abs(metrics.ActiveTasks - lastMetrics.ActiveTasks);
|
||||
int maxTasksDiff = Math.Abs(metrics.MaxConcurrentTasks - lastMetrics.MaxConcurrentTasks);
|
||||
|
||||
if (cpuDiff < cpuDifferenceThreshold && memoryDiff < memoryDifferenceThreshold && connectionsDiff == 0 && tasksDiff == 0 && maxTasksDiff == 0)
|
||||
{
|
||||
return; // No significant change
|
||||
}
|
||||
}
|
||||
|
||||
lastMetrics = metrics;
|
||||
|
||||
await hubContext.Clients.All.SendAsync("MetricsUpdated", metrics);
|
||||
}
|
||||
|
||||
private static async Task<ServerMetrics> GetServerMetrics()
|
||||
{
|
||||
Process currentProcess = Process.GetCurrentProcess();
|
||||
|
||||
DateTime currentTime = DateTime.UtcNow;
|
||||
TimeSpan currentProcessorTime = currentProcess.TotalProcessorTime;
|
||||
|
||||
double elapsedMs = (currentTime - lastMetricsTimestamp).TotalMilliseconds;
|
||||
double cpuMsUsed = (currentProcessorTime - lastTotalProcessorTime).TotalMilliseconds;
|
||||
|
||||
double cpuUsagePercent = cpuMsUsed / elapsedMs / Environment.ProcessorCount * 100;
|
||||
|
||||
lastMetricsTimestamp = currentTime;
|
||||
lastTotalProcessorTime = currentProcessorTime;
|
||||
|
||||
return new ServerMetrics
|
||||
{
|
||||
ServerId = Environment.MachineName,
|
||||
ActiveConnections = connections.Count,
|
||||
ActiveTasks = activeTasks,
|
||||
MaxConcurrentTasks = maxConcurrentTasks,
|
||||
TotalMemoryUsage = currentProcess.WorkingSet64,
|
||||
CpuUsage = Math.Clamp(Math.Round(cpuUsagePercent, 2), 0, 100),
|
||||
Timestamp = currentTime
|
||||
};
|
||||
}
|
||||
|
||||
private async Task<byte[]> RequestAssemblyBytesAsync(string assemblyName, HubCallerContext context, IHubCallerClients hubCallerClients)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!pendingAssemblyRequestsByConnection.TryGetValue(context.ConnectionId, out ConcurrentDictionary<string, Lazy<Task<byte[]>>>? connectionPendingRequests))
|
||||
{
|
||||
throw new InvalidOperationException("Connection not found");
|
||||
}
|
||||
|
||||
// Use Lazy<Task<T>> pattern to ensure only one request is made
|
||||
// The Lazy.Value is only evaluated once, even if multiple threads access it simultaneously
|
||||
Lazy<Task<byte[]>> lazyTask = connectionPendingRequests.GetOrAdd(assemblyName, key =>
|
||||
{
|
||||
return new Lazy<Task<byte[]>>(() => Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
Guid guid = Guid.NewGuid();
|
||||
TaskCompletionSource<byte[]> tcs = new TaskCompletionSource<byte[]>();
|
||||
|
||||
_ = pendingAssemblyRequests.TryAdd(guid, tcs);
|
||||
|
||||
logger.LogInformation("Requesting assembly {Assembly} with RequestId {RequestId} for connection {ConnectionId}", key, guid, context.ConnectionId);
|
||||
|
||||
await hubCallerClients.Caller.SendAsync("RequestAssembly", key, guid);
|
||||
|
||||
return await tcs.Task.WaitAsync(TimeSpan.FromSeconds(assemblyLoadTimeoutSeconds));
|
||||
}
|
||||
finally
|
||||
{
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
// WORKAROUND: Delay to avoid race condition where the same assembly is requested again because the assembly hasn't been loaded yet
|
||||
// An alternative would be to only remove the request after the client disconnects or after a longer timeout
|
||||
await Task.Delay(1000);
|
||||
_ = connectionPendingRequests.TryRemove(key, out _);
|
||||
});
|
||||
}
|
||||
}));
|
||||
});
|
||||
|
||||
// All callers will await the same Task
|
||||
return await lazyTask.Value;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error loading assembly {Assembly}", assemblyName);
|
||||
}
|
||||
|
||||
return [];
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
using Microsoft.Extensions.Options;
|
||||
using Microsoft.Extensions.Primitives;
|
||||
|
||||
using RemoteExec.Server.Configuration;
|
||||
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace RemoteExec.Server.Middleware;
|
||||
|
||||
/// <summary>
|
||||
/// Middleware that authenticates requests using API keys in the X-API-Key header.
|
||||
/// </summary>
|
||||
public class ApiKeyAuthenticationMiddleware
|
||||
{
|
||||
private readonly RequestDelegate _next;
|
||||
private readonly ILogger<ApiKeyAuthenticationMiddleware> _logger;
|
||||
private readonly ConcurrentDictionary<string, ApiKeyConfiguration> apiKeys;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ApiKeyAuthenticationMiddleware"/> class.
|
||||
/// </summary>
|
||||
/// <param name="next">The next middleware in the pipeline.</param>
|
||||
/// <param name="authOptions">The authentication configuration options.</param>
|
||||
/// <param name="logger">The logger instance.</param>
|
||||
public ApiKeyAuthenticationMiddleware(RequestDelegate next, IOptionsMonitor<AuthenticationConfiguration> authOptions, ILogger<ApiKeyAuthenticationMiddleware> logger)
|
||||
{
|
||||
_next = next;
|
||||
_logger = logger;
|
||||
apiKeys = new ConcurrentDictionary<string, ApiKeyConfiguration>();
|
||||
|
||||
LoadApiKeys(authOptions.CurrentValue);
|
||||
|
||||
_ = authOptions.OnChange(LoadApiKeys);
|
||||
}
|
||||
|
||||
private void LoadApiKeys(AuthenticationConfiguration config)
|
||||
{
|
||||
apiKeys.Clear();
|
||||
|
||||
if (config.ApiKeys == null || config.ApiKeys.Count == 0)
|
||||
{
|
||||
_logger.LogWarning("No API keys configured. All requests will be rejected.");
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (ApiKeyConfiguration apiKey in config.ApiKeys.Where(k => k.Enabled))
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(apiKey.Key))
|
||||
{
|
||||
_logger.LogWarning("Skipping empty API key configuration");
|
||||
continue;
|
||||
}
|
||||
|
||||
if (apiKeys.TryAdd(apiKey.Key, apiKey))
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"Registered API key: {Description}",
|
||||
apiKey.Description ?? "No description");
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Duplicate API key found and skipped: {Description}",
|
||||
apiKey.Description ?? "No description");
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("Loaded {Count} active API keys", apiKeys.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invokes the middleware to authenticate the request.
|
||||
/// </summary>
|
||||
/// <param name="context">The HTTP context for the current request.</param>
|
||||
public async Task InvokeAsync(HttpContext context)
|
||||
{
|
||||
// Skip authentication for health checks
|
||||
if (context.Request.Path.StartsWithSegments("/health"))
|
||||
{
|
||||
await _next(context);
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if any API keys are configured
|
||||
if (apiKeys.IsEmpty)
|
||||
{
|
||||
_logger.LogError("No API keys configured. Rejecting request to {Path}", context.Request.Path);
|
||||
|
||||
context.Response.StatusCode = 503;
|
||||
|
||||
await context.Response.WriteAsync("Service is not properly configured");
|
||||
return;
|
||||
}
|
||||
|
||||
// Check for API key in header
|
||||
if (!context.Request.Headers.TryGetValue("X-API-Key", out StringValues extractedApiKey))
|
||||
{
|
||||
_logger.LogWarning("API Key missing from request to {Path} from {RemoteIp}", context.Request.Path, context.Connection.RemoteIpAddress);
|
||||
|
||||
context.Response.StatusCode = 401;
|
||||
|
||||
await context.Response.WriteAsync("API Key is missing");
|
||||
return;
|
||||
}
|
||||
|
||||
string providedKey = extractedApiKey.ToString();
|
||||
|
||||
// Validate API key
|
||||
if (!apiKeys.TryGetValue(providedKey, out ApiKeyConfiguration? apiKeyConfig))
|
||||
{
|
||||
_logger.LogWarning("Invalid API Key provided for request to {Path} from {RemoteIp}",
|
||||
context.Request.Path,
|
||||
context.Connection.RemoteIpAddress);
|
||||
context.Response.StatusCode = 401;
|
||||
await context.Response.WriteAsync("Invalid API Key");
|
||||
return;
|
||||
}
|
||||
|
||||
// Store API key info in HttpContext for potential use in controllers
|
||||
context.Items["ApiKeyDescription"] = apiKeyConfig.Description;
|
||||
context.Items["ApiKey"] = providedKey;
|
||||
|
||||
_logger.LogDebug("Authenticated request to {Path} using key: {Description}",
|
||||
context.Request.Path,
|
||||
apiKeyConfig.Description ?? "No description");
|
||||
|
||||
await _next(context);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
using RemoteExec.Server.Configuration;
|
||||
using RemoteExec.Server.Hubs;
|
||||
using RemoteExec.Server.Middleware;
|
||||
using RemoteExec.Server.Services;
|
||||
|
||||
WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
|
||||
|
||||
// Add services to the container.
|
||||
builder.Services.AddControllers();
|
||||
builder.Services.AddSignalR();
|
||||
builder.Services.AddOpenApi();
|
||||
builder.Services.AddHealthChecks();
|
||||
|
||||
builder.Services.Configure<AuthenticationConfiguration>(builder.Configuration.GetSection("Authentication"));
|
||||
builder.Services.Configure<ExecutionConfiguration>(builder.Configuration.GetSection("Execution"));
|
||||
builder.Services.Configure<DockerExecutionConfiguration>(builder.Configuration.GetSection("DockerExecution"));
|
||||
builder.Services.Configure<MetricsConfiguration>(builder.Configuration.GetSection("Metrics"));
|
||||
|
||||
builder.Services.AddScoped<ExecutionEnvironment, DockerContainerExecutionEnvironment>();
|
||||
builder.Services.AddScoped<ExecutionEnvironment, AssemblyLoadContextExecutionEnvironment>();
|
||||
|
||||
builder.Services.AddHostedService<MetricsBroadcastService>();
|
||||
|
||||
WebApplication app = builder.Build();
|
||||
|
||||
app.UseMiddleware<ApiKeyAuthenticationMiddleware>();
|
||||
|
||||
app.MapHub<RemoteExecutionHub>("/remote");
|
||||
|
||||
// Configure the HTTP request pipeline.
|
||||
if (app.Environment.IsDevelopment())
|
||||
{
|
||||
_ = app.MapOpenApi();
|
||||
}
|
||||
|
||||
app.UseHttpsRedirection();
|
||||
|
||||
app.UseAuthorization();
|
||||
|
||||
app.MapControllers();
|
||||
|
||||
app.MapHealthChecks("/health");
|
||||
|
||||
await app.RunAsync();
|
||||
@@ -0,0 +1,31 @@
|
||||
{
|
||||
"profiles": {
|
||||
"http": {
|
||||
"commandName": "Project",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
},
|
||||
"dotnetRunMessages": true,
|
||||
"applicationUrl": "http://localhost:5202"
|
||||
},
|
||||
"https": {
|
||||
"commandName": "Project",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
},
|
||||
"dotnetRunMessages": true,
|
||||
"applicationUrl": "https://localhost:7109;http://localhost:5202"
|
||||
},
|
||||
"Container (Dockerfile)": {
|
||||
"commandName": "Docker",
|
||||
"launchUrl": "{Scheme}://{ServiceHost}:{ServicePort}",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_HTTPS_PORTS": "8081",
|
||||
"ASPNETCORE_HTTP_PORTS": "8080"
|
||||
},
|
||||
"publishAllPorts": true,
|
||||
"useSSL": true
|
||||
}
|
||||
},
|
||||
"$schema": "https://json.schemastore.org/launchsettings.json"
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<UserSecretsId>288f55bd-4ce1-4c6a-867f-a12ef417c7cf</UserSecretsId>
|
||||
<DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Docker.DotNet" Version="3.125.15" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="10.0.0" />
|
||||
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.23.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\RemoteExec.Shared\RemoteExec.Shared.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,6 @@
|
||||
@RemoteExec.Server_HostAddress = http://localhost:5202
|
||||
|
||||
GET {{RemoteExec.Server_HostAddress}}/weatherforecast/
|
||||
Accept: application/json
|
||||
|
||||
###
|
||||
@@ -0,0 +1,10 @@
|
||||
using System.Runtime.Loader;
|
||||
|
||||
namespace RemoteExec.Server;
|
||||
|
||||
/// <summary>
|
||||
/// An isolated assembly load context for remote job execution, allowing assemblies to be unloaded.
|
||||
/// </summary>
|
||||
public class RemoteJobAssemblyLoadContext(string name) : AssemblyLoadContext(name, true)
|
||||
{
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
using System.Reflection;
|
||||
|
||||
namespace RemoteExec.Server;
|
||||
|
||||
/// <summary>
|
||||
/// Event arguments for assembly request events.
|
||||
/// </summary>
|
||||
public class RequestAssemblyEventArgs(AssemblyName assemblyName) : EventArgs
|
||||
{
|
||||
private readonly TaskCompletionSource taskCompletionSource = new TaskCompletionSource();
|
||||
|
||||
private Assembly? assembly = null;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the assembly name being requested.
|
||||
/// </summary>
|
||||
public AssemblyName Assembly { get; } = assemblyName;
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronously waits for the assembly to be provided.
|
||||
/// </summary>
|
||||
/// <returns>The assembly, or null if not provided.</returns>
|
||||
public async Task<Assembly?> GetAssemblyAsync()
|
||||
{
|
||||
await taskCompletionSource.Task;
|
||||
return assembly;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sets the assembly to fulfill the request.
|
||||
/// </summary>
|
||||
/// <param name="assembly">The assembly to provide.</param>
|
||||
public void SetAssembly(Assembly assembly)
|
||||
{
|
||||
this.assembly = assembly;
|
||||
taskCompletionSource.SetResult();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
using RemoteExec.Shared.Models;
|
||||
using RemoteExec.Shared.Utilities;
|
||||
|
||||
using System.Reflection;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace RemoteExec.Server.Services;
|
||||
|
||||
public class AssemblyLoadContextExecutionEnvironment : ExecutionEnvironment
|
||||
{
|
||||
public override string Name => "AssemblyLoadContext";
|
||||
|
||||
private RemoteJobAssemblyLoadContext? assemblyLoadContext;
|
||||
|
||||
public override async Task<RemoteExecutionResult> ExecuteTaskAsync(RemoteExecutionRequest request)
|
||||
{
|
||||
if (assemblyLoadContext is null)
|
||||
{
|
||||
throw new InvalidOperationException("The execution environment has not been prepared.");
|
||||
}
|
||||
|
||||
// Check if assembly is already loaded in the context
|
||||
Assembly? assembly = assemblyLoadContext.Assemblies.FirstOrDefault(a => a.GetName().FullName == request.AssemblyName);
|
||||
|
||||
// If not loaded, request and load it into the context
|
||||
assembly ??= assemblyLoadContext.LoadFromBytes(await RequestAssemblyAsync(request.AssemblyName));
|
||||
|
||||
Type type = assembly.GetType(request.TypeName, throwOnError: true)!;
|
||||
|
||||
Type[] argTypes = request.ArgumentTypes
|
||||
.Select(Type.GetType)
|
||||
.ToArray()!;
|
||||
|
||||
MethodInfo? method = type.GetMethod(request.MethodName, BindingFlags.Static | BindingFlags.Public | BindingFlags.NonPublic, binder: null, argTypes, modifiers: null) ?? throw new MissingMethodException(request.TypeName, request.MethodName);
|
||||
|
||||
// Pre-load all referenced assemblies to avoid triggering Resolving event during Invoke
|
||||
await AssemblyUtilities.PreLoadReferencedAssembliesAsync(assemblyLoadContext, assembly, RequestAssemblyAsync);
|
||||
|
||||
ParameterInfo[] parameters = method.GetParameters();
|
||||
|
||||
if (parameters.Length != request.Arguments.Length)
|
||||
{
|
||||
throw new ArgumentException($"Argument count mismatch: expected {parameters.Length}, received {request.Arguments.Length}");
|
||||
}
|
||||
|
||||
object?[] invokeArgs = new object?[request.Arguments.Length];
|
||||
|
||||
for (int i = 0; i < invokeArgs.Length; i++)
|
||||
{
|
||||
Type targetType = parameters[i].ParameterType;
|
||||
object arg = request.Arguments[i];
|
||||
|
||||
if (arg is JsonElement je)
|
||||
{
|
||||
invokeArgs[i] = JsonSerializer.Deserialize(je.GetRawText(), targetType);
|
||||
}
|
||||
else if (arg == null)
|
||||
{
|
||||
invokeArgs[i] = null;
|
||||
}
|
||||
else if (!targetType.IsInstanceOfType(arg))
|
||||
{
|
||||
invokeArgs[i] = Convert.ChangeType(arg, targetType);
|
||||
}
|
||||
else
|
||||
{
|
||||
invokeArgs[i] = arg;
|
||||
}
|
||||
}
|
||||
|
||||
object? result = method.Invoke(null, invokeArgs);
|
||||
|
||||
if (result is Task taskResult)
|
||||
{
|
||||
await taskResult.ConfigureAwait(false);
|
||||
|
||||
Type returnType = method.ReturnType;
|
||||
|
||||
if (returnType.IsGenericType && returnType.GetGenericTypeDefinition() == typeof(Task<>))
|
||||
{
|
||||
PropertyInfo resultProperty = returnType.GetProperty("Result")!;
|
||||
result = resultProperty.GetValue(taskResult);
|
||||
}
|
||||
else
|
||||
{
|
||||
result = null;
|
||||
}
|
||||
}
|
||||
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Result = result
|
||||
};
|
||||
}
|
||||
|
||||
public override Task PrepareEnvironmentAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
assemblyLoadContext = new RemoteJobAssemblyLoadContext($"RemoteJob_{Guid.NewGuid()}");
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override Task CleanupEnvironmentAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
assemblyLoadContext?.Unload();
|
||||
assemblyLoadContext = null;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,363 @@
|
||||
using Docker.DotNet;
|
||||
using Docker.DotNet.Models;
|
||||
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
using RemoteExec.Server.Configuration;
|
||||
using RemoteExec.Shared.Models;
|
||||
using RemoteExec.Shared.Models.Docker;
|
||||
|
||||
using System.Collections.Concurrent;
|
||||
using System.Formats.Tar;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace RemoteExec.Server.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Executes tasks in ephemeral Docker containers for maximum isolation.
|
||||
/// </summary>
|
||||
public class DockerContainerExecutionEnvironment : ExecutionEnvironment
|
||||
{
|
||||
public override string Name => "DockerContainer";
|
||||
|
||||
private readonly DockerClient dockerClient;
|
||||
private readonly ILogger<DockerContainerExecutionEnvironment> logger;
|
||||
private readonly string workerImageName;
|
||||
private readonly TimeSpan containerTimeout;
|
||||
private readonly long memoryLimit;
|
||||
private readonly long cpuLimit;
|
||||
private readonly bool networkDisabled;
|
||||
private readonly bool readOnlyFilesystem;
|
||||
|
||||
private readonly ConcurrentDictionary<string, byte[]> assemblyCache = [];
|
||||
private readonly ConcurrentDictionary<string, string> runningContainers = [];
|
||||
|
||||
public DockerContainerExecutionEnvironment(ILogger<DockerContainerExecutionEnvironment> logger, IOptions<DockerExecutionConfiguration> dockerConfig)
|
||||
{
|
||||
this.logger = logger;
|
||||
|
||||
DockerExecutionConfiguration config = dockerConfig.Value;
|
||||
|
||||
workerImageName = config.WorkerImageName;
|
||||
containerTimeout = TimeSpan.FromSeconds(config.ContainerTimeoutSeconds);
|
||||
memoryLimit = config.ContainerMemoryLimitMb * 1024 * 1024;
|
||||
cpuLimit = config.ContainerCpuShares;
|
||||
networkDisabled = config.DisableNetwork;
|
||||
readOnlyFilesystem = config.ReadOnlyFilesystem;
|
||||
|
||||
DockerClientConfiguration dockerClientConfig;
|
||||
|
||||
if (string.IsNullOrEmpty(config.DockerHost))
|
||||
{
|
||||
dockerClientConfig = new DockerClientConfiguration();
|
||||
}
|
||||
else
|
||||
{
|
||||
dockerClientConfig = new DockerClientConfiguration(new Uri(config.DockerHost));
|
||||
}
|
||||
|
||||
dockerClient = dockerClientConfig.CreateClient();
|
||||
}
|
||||
|
||||
public override Task PrepareEnvironmentAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override async Task<RemoteExecutionResult> ExecuteTaskAsync(RemoteExecutionRequest request)
|
||||
{
|
||||
string containerId = string.Empty;
|
||||
|
||||
try
|
||||
{
|
||||
if (!assemblyCache.TryGetValue(request.AssemblyName, out byte[]? assemblyBytes))
|
||||
{
|
||||
assemblyBytes = await RequestAssemblyAsync(request.AssemblyName);
|
||||
assemblyCache[request.AssemblyName] = assemblyBytes;
|
||||
}
|
||||
|
||||
ContainerExecutionRequest containerRequest = new()
|
||||
{
|
||||
AssemblyBytes = Convert.ToBase64String(assemblyBytes),
|
||||
TypeName = request.TypeName,
|
||||
MethodName = request.MethodName,
|
||||
ArgumentTypes = request.ArgumentTypes,
|
||||
Arguments = request.Arguments
|
||||
};
|
||||
|
||||
string requestJson = JsonSerializer.Serialize(containerRequest);
|
||||
|
||||
containerId = await CreateAndStartContainerAsync(requestJson, CancellationToken.None);
|
||||
|
||||
using CancellationTokenSource timeoutCts = new(containerTimeout);
|
||||
Task logMonitorTask = MonitorContainerLogsAsync(containerId, timeoutCts.Token);
|
||||
|
||||
ContainerWaitResponse waitResponse = await dockerClient.Containers.WaitContainerAsync(containerId, timeoutCts.Token);
|
||||
|
||||
await timeoutCts.CancelAsync();
|
||||
|
||||
try
|
||||
{
|
||||
await logMonitorTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected
|
||||
}
|
||||
|
||||
// Get container logs (stdout contains JSON result)
|
||||
string stdout = await GetContainerLogsAsync(containerId);
|
||||
|
||||
if (waitResponse.StatusCode != 0)
|
||||
{
|
||||
logger.LogError("Container {ContainerId} exited with code {ExitCode}", containerId, waitResponse.StatusCode);
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Exception = $"Container exited with code {waitResponse.StatusCode}\nOutput: {stdout}"
|
||||
};
|
||||
}
|
||||
|
||||
string[] lines = stdout.Split('\n', StringSplitOptions.RemoveEmptyEntries);
|
||||
string? resultLine = lines.LastOrDefault(l =>
|
||||
{
|
||||
string trimmed = l.TrimStart();
|
||||
return trimmed.StartsWith('{') && !trimmed.Contains("#REQUEST_ASSEMBLY") && !trimmed.Contains("#PROVIDE_ASSEMBLY");
|
||||
});
|
||||
|
||||
if (resultLine == null)
|
||||
{
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Exception = $"No valid JSON result found in output: {stdout}"
|
||||
};
|
||||
}
|
||||
|
||||
ContainerExecutionResponse? response = JsonSerializer.Deserialize<ContainerExecutionResponse>(resultLine);
|
||||
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Result = response?.Result,
|
||||
Exception = response?.Exception
|
||||
};
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error executing task in container {ContainerId}", containerId);
|
||||
return new RemoteExecutionResult
|
||||
{
|
||||
Exception = ex.ToString()
|
||||
};
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (!string.IsNullOrEmpty(containerId))
|
||||
{
|
||||
await CleanupContainerAsync(containerId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public override async Task CleanupEnvironmentAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (string containerId in runningContainers.Keys)
|
||||
{
|
||||
await CleanupContainerAsync(containerId);
|
||||
}
|
||||
|
||||
assemblyCache.Clear();
|
||||
|
||||
logger.LogInformation("Docker container execution environment cleaned up");
|
||||
}
|
||||
|
||||
private async Task MonitorContainerLogsAsync(string containerId, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
MultiplexedStream logStream = await dockerClient.Containers.GetContainerLogsAsync(containerId, false, new ContainerLogsParameters
|
||||
{
|
||||
ShowStdout = true,
|
||||
ShowStderr = false,
|
||||
Follow = true
|
||||
}, cancellationToken);
|
||||
|
||||
byte[] buffer = new byte[4096];
|
||||
StringBuilder lineBuffer = new();
|
||||
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
MultiplexedStream.ReadResult result = await logStream.ReadOutputAsync(buffer, 0, buffer.Length, cancellationToken);
|
||||
|
||||
if (result.Count == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
string text = Encoding.UTF8.GetString(buffer, 0, result.Count);
|
||||
_ = lineBuffer.Append(text);
|
||||
|
||||
string bufferContent = lineBuffer.ToString();
|
||||
int lastNewline = bufferContent.LastIndexOf('\n');
|
||||
|
||||
if (lastNewline == -1)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
string completeLines = bufferContent[..lastNewline];
|
||||
string remaining = bufferContent[(lastNewline + 1)..];
|
||||
|
||||
_ = lineBuffer.Clear();
|
||||
_ = lineBuffer.Append(remaining);
|
||||
|
||||
string[] lines = completeLines.Split('\n', StringSplitOptions.RemoveEmptyEntries);
|
||||
|
||||
foreach (string line in lines)
|
||||
{
|
||||
string trimmedLine = line.Trim();
|
||||
|
||||
if (trimmedLine.StartsWith("#REQUEST_ASSEMBLY ") && trimmedLine.EndsWith('#'))
|
||||
{
|
||||
string assemblyName = trimmedLine.Substring("#REQUEST_ASSEMBLY ".Length, trimmedLine.Length - "#REQUEST_ASSEMBLY ".Length - 1);
|
||||
_ = Task.Run(() => HandleAssemblyRequestAsync(containerId, assemblyName, cancellationToken), cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected when container completes or timeout occurs
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error monitoring container {ContainerId} logs", containerId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleAssemblyRequestAsync(string containerId, string assemblyName, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!assemblyCache.TryGetValue(assemblyName, out byte[]? assemblyBytes))
|
||||
{
|
||||
assemblyBytes = await RequestAssemblyAsync(assemblyName);
|
||||
assemblyCache[assemblyName] = assemblyBytes;
|
||||
}
|
||||
|
||||
using MemoryStream tarStream = new();
|
||||
using (TarWriter tarWriter = new(tarStream, TarEntryFormat.Ustar, leaveOpen: true))
|
||||
{
|
||||
string fileName = $"{assemblyName}.dll";
|
||||
|
||||
UstarTarEntry dllEntry = new(TarEntryType.RegularFile, fileName)
|
||||
{
|
||||
DataStream = new MemoryStream(assemblyBytes)
|
||||
};
|
||||
await tarWriter.WriteEntryAsync(dllEntry, cancellationToken);
|
||||
|
||||
UstarTarEntry sentinelEntry = new(TarEntryType.RegularFile, fileName + ".ready")
|
||||
{
|
||||
DataStream = new MemoryStream()
|
||||
};
|
||||
|
||||
await tarWriter.WriteEntryAsync(sentinelEntry, cancellationToken);
|
||||
}
|
||||
|
||||
tarStream.Position = 0;
|
||||
|
||||
ContainerPathStatParameters pathParams = new()
|
||||
{
|
||||
Path = "/tmp",
|
||||
AllowOverwriteDirWithFile = false,
|
||||
};
|
||||
|
||||
await dockerClient.Containers.ExtractArchiveToContainerAsync(containerId, pathParams, tarStream, cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error transferring assembly {AssemblyName} to container {ContainerId}", assemblyName, containerId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<string> CreateAndStartContainerAsync(string requestJson, CancellationToken cancellationToken)
|
||||
{
|
||||
CreateContainerParameters parameters = new()
|
||||
{
|
||||
Image = workerImageName,
|
||||
Name = $"remoteexec-task-{Guid.NewGuid()}",
|
||||
HostConfig = new HostConfig
|
||||
{
|
||||
Memory = memoryLimit,
|
||||
CPUShares = cpuLimit,
|
||||
NetworkMode = networkDisabled ? "none" : "bridge",
|
||||
ReadonlyRootfs = readOnlyFilesystem,
|
||||
AutoRemove = false,
|
||||
CapDrop = ["ALL"],
|
||||
SecurityOpt = ["no-new-privileges"],
|
||||
Tmpfs = new Dictionary<string, string>
|
||||
{
|
||||
["/tmp/assemblies"] = "rw,noexec,nosuid,size=100m"
|
||||
}
|
||||
},
|
||||
Env =
|
||||
[
|
||||
$"EXECUTION_REQUEST={Convert.ToBase64String(Encoding.UTF8.GetBytes(requestJson))}"
|
||||
],
|
||||
WorkingDir = "/app",
|
||||
AttachStdout = true,
|
||||
AttachStderr = true
|
||||
};
|
||||
|
||||
CreateContainerResponse container = await dockerClient.Containers.CreateContainerAsync(parameters, cancellationToken);
|
||||
|
||||
bool started = await dockerClient.Containers.StartContainerAsync(container.ID, new ContainerStartParameters(), cancellationToken);
|
||||
|
||||
if (!started)
|
||||
{
|
||||
throw new InvalidOperationException($"Failed to start container {container.ID}");
|
||||
}
|
||||
|
||||
runningContainers[container.ID] = string.Empty;
|
||||
|
||||
return container.ID;
|
||||
}
|
||||
|
||||
private async Task<string> GetContainerLogsAsync(string containerId)
|
||||
{
|
||||
MultiplexedStream logStream = await dockerClient.Containers.GetContainerLogsAsync(containerId, false, new ContainerLogsParameters
|
||||
{
|
||||
ShowStdout = true,
|
||||
ShowStderr = true
|
||||
});
|
||||
|
||||
StringBuilder output = new();
|
||||
byte[] buffer = new byte[4096];
|
||||
|
||||
MultiplexedStream.ReadResult result = await logStream.ReadOutputAsync(buffer, 0, buffer.Length, CancellationToken.None);
|
||||
|
||||
while (result.Count > 0)
|
||||
{
|
||||
logger.LogDebug("Read {ByteCount} bytes from container {ContainerId} logs", result.Count, containerId);
|
||||
|
||||
_ = output.Append(Encoding.UTF8.GetString(buffer, 0, result.Count));
|
||||
result = await logStream.ReadOutputAsync(buffer, 0, buffer.Length, CancellationToken.None);
|
||||
}
|
||||
|
||||
return output.ToString();
|
||||
}
|
||||
|
||||
private async Task CleanupContainerAsync(string containerId)
|
||||
{
|
||||
try
|
||||
{
|
||||
_ = await dockerClient.Containers.StopContainerAsync(containerId, new ContainerStopParameters { WaitBeforeKillSeconds = 5 });
|
||||
await dockerClient.Containers.RemoveContainerAsync(containerId, new ContainerRemoveParameters { Force = true, RemoveVolumes = true });
|
||||
|
||||
_ = runningContainers.TryRemove(containerId, out _);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "Failed to cleanup container {ContainerId}", containerId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
using RemoteExec.Server.Utilities;
|
||||
using RemoteExec.Shared.Models;
|
||||
|
||||
namespace RemoteExec.Server.Services;
|
||||
|
||||
public abstract class ExecutionEnvironment
|
||||
{
|
||||
public event EventHandler<CompletableEventArgs<string, byte[]>>? RequestAssembly;
|
||||
|
||||
public abstract string Name { get; }
|
||||
|
||||
public abstract Task PrepareEnvironmentAsync(CancellationToken cancellationToken);
|
||||
|
||||
public abstract Task<RemoteExecutionResult> ExecuteTaskAsync(RemoteExecutionRequest request);
|
||||
|
||||
public abstract Task CleanupEnvironmentAsync(CancellationToken cancellationToken);
|
||||
|
||||
protected async Task<byte[]> RequestAssemblyAsync(string assemblyName)
|
||||
{
|
||||
CompletableEventArgs<string, byte[]> args = new CompletableEventArgs<string, byte[]>(assemblyName);
|
||||
RequestAssembly?.Invoke(this, args);
|
||||
return await args.WaitAsync();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
using RemoteExec.Server.Configuration;
|
||||
using RemoteExec.Server.Hubs;
|
||||
|
||||
namespace RemoteExec.Server.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Background service that periodically broadcasts server metrics to all connected clients.
|
||||
/// </summary>
|
||||
public class MetricsBroadcastService(IHubContext<RemoteExecutionHub> hubContext, ILogger<MetricsBroadcastService> logger, IOptions<MetricsConfiguration> metricsOptions) : BackgroundService
|
||||
{
|
||||
private readonly TimeSpan broadcastInterval = TimeSpan.FromMilliseconds(metricsOptions.Value.BroadcastIntervalMs);
|
||||
|
||||
/// <inheritdoc/>
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
logger.LogInformation("Metrics broadcast service started with interval {Interval}ms", broadcastInterval.TotalMilliseconds);
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(broadcastInterval, stoppingToken);
|
||||
|
||||
await RemoteExecutionHub.BroadcastMetricsAsync(hubContext);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected when service is stopping
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Error broadcasting metrics");
|
||||
}
|
||||
}
|
||||
|
||||
logger.LogInformation("Metrics broadcast service stopped");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
using System.Reflection;
|
||||
|
||||
namespace RemoteExec.Server.Utilities;
|
||||
|
||||
public static class AssemblyUtilities
|
||||
{
|
||||
public static async Task PreLoadReferencedAssembliesAsync(RemoteJobAssemblyLoadContext assemblyLoadContext, Assembly assembly, Func<string, Task<byte[]>> requestAssembly)
|
||||
{
|
||||
AssemblyName[] referencedAssemblies = assembly.GetReferencedAssemblies();
|
||||
|
||||
foreach (AssemblyName referencedAssembly in referencedAssemblies)
|
||||
{
|
||||
// Try to load from the assembly load context first
|
||||
Assembly? loadedAssembly = assemblyLoadContext.Assemblies.FirstOrDefault(a => a.GetName().FullName == referencedAssembly.FullName);
|
||||
|
||||
if (loadedAssembly != null)
|
||||
{
|
||||
continue; // Already loaded in the context
|
||||
}
|
||||
|
||||
// Try to load from default context (BCL assemblies)
|
||||
try
|
||||
{
|
||||
_ = assemblyLoadContext.LoadFromAssemblyName(referencedAssembly);
|
||||
continue; // Successfully loaded from default context
|
||||
}
|
||||
catch
|
||||
{
|
||||
byte[] assemblyBytes = await requestAssembly(referencedAssembly.FullName!);
|
||||
_ = assemblyLoadContext.LoadFromBytes(assemblyBytes);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static Assembly LoadFromBytes(this RemoteJobAssemblyLoadContext assemblyLoadContext, byte[] assemblyBytes)
|
||||
{
|
||||
using MemoryStream ms = new(assemblyBytes);
|
||||
return assemblyLoadContext.LoadFromStream(ms);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
namespace RemoteExec.Server.Utilities;
|
||||
|
||||
public class CompletableEventArgs : EventArgs
|
||||
{
|
||||
private readonly TaskCompletionSource<bool> tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
public void SetCompleted()
|
||||
{
|
||||
_ = tcs.TrySetResult(true);
|
||||
}
|
||||
|
||||
public Task WaitAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (cancellationToken.CanBeCanceled)
|
||||
{
|
||||
_ = cancellationToken.Register(() => tcs.TrySetCanceled(cancellationToken));
|
||||
}
|
||||
return tcs.Task;
|
||||
}
|
||||
}
|
||||
|
||||
public class CompletableEventArgs<T> : EventArgs
|
||||
{
|
||||
private readonly TaskCompletionSource<T> tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
public void SetCompleted(T result)
|
||||
{
|
||||
_ = tcs.TrySetResult(result);
|
||||
}
|
||||
public Task<T> WaitAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (cancellationToken.CanBeCanceled)
|
||||
{
|
||||
_ = cancellationToken.Register(() => tcs.TrySetCanceled(cancellationToken));
|
||||
}
|
||||
return tcs.Task;
|
||||
}
|
||||
}
|
||||
|
||||
public class CompletableEventArgs<TValue, TResult>(TValue value)
|
||||
{
|
||||
public TValue Value { get; } = value;
|
||||
|
||||
private readonly TaskCompletionSource<TResult> tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
public void SetCompleted(TResult result)
|
||||
{
|
||||
_ = tcs.TrySetResult(result);
|
||||
}
|
||||
public Task<TResult> WaitAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (cancellationToken.CanBeCanceled)
|
||||
{
|
||||
_ = cancellationToken.Register(() => tcs.TrySetCanceled(cancellationToken));
|
||||
}
|
||||
return tcs.Task;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
{
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
},
|
||||
"AllowedHosts": "*",
|
||||
|
||||
// Authentication configuration
|
||||
"Authentication": {
|
||||
// List of API keys for authenticating clients
|
||||
// Each entry should contain key identification and validation information
|
||||
"ApiKeys": [
|
||||
// Example:
|
||||
// {
|
||||
// "Key": "your-secret-api-key-here",
|
||||
// "Name": "Client1",
|
||||
// "Enabled": true
|
||||
// }
|
||||
]
|
||||
},
|
||||
|
||||
// Remote execution configuration
|
||||
"Execution": {
|
||||
// Maximum number of tasks that can execute concurrently on the server
|
||||
// null = uses Environment.ProcessorCount * 2 (default)
|
||||
// Increase for CPU-light tasks, decrease for resource-intensive operations
|
||||
"MaxConcurrentTasks": null,
|
||||
|
||||
// Timeout in seconds for loading assemblies from clients
|
||||
// If a client doesn't respond with assembly bytes within this time, the request fails
|
||||
// Default: 30 seconds
|
||||
"AssemblyLoadTimeoutSeconds": 30,
|
||||
|
||||
// Type of execution environment to use
|
||||
// Options: "AssemblyLoadContext", "DockerContainer"
|
||||
// - AssemblyLoadContext: Lighter weight, per-connection assembly isolation
|
||||
// - DockerContainer: Maximum isolation, ephemeral containers per task
|
||||
// Default: "AssemblyLoadContext"
|
||||
"ExecutionEnvironment": "AssemblyLoadContext"
|
||||
},
|
||||
|
||||
// Docker execution configuration (only used when ExecutionEnvironment = "DockerContainer")
|
||||
"DockerExecution": {
|
||||
// Docker host URL
|
||||
// Linux: "unix:///var/run/docker.sock"
|
||||
// Windows: "npipe://./pipe/docker_engine"
|
||||
"DockerHost": "unix:///var/run/docker.sock",
|
||||
|
||||
// Docker worker image name to use for task execution
|
||||
// Must be built and available on the Docker host
|
||||
// Default: "remoteexec-worker:latest"
|
||||
"WorkerImageName": "remoteexec-worker:latest",
|
||||
|
||||
// Maximum execution time per container in seconds
|
||||
// Containers exceeding this time will be forcefully terminated
|
||||
// Default: 300 (5 minutes)
|
||||
"ContainerTimeoutSeconds": 300,
|
||||
|
||||
// Memory limit per container in MB
|
||||
// Prevents containers from consuming excessive memory
|
||||
// Default: 512 MB
|
||||
"ContainerMemoryLimitMb": 512,
|
||||
|
||||
// CPU shares allocated to each container (relative weight)
|
||||
// Higher values = more CPU priority
|
||||
// Default: 1024
|
||||
"ContainerCpuShares": 1024,
|
||||
|
||||
// Disable network access in containers for security
|
||||
// Set to false if tasks require network connectivity
|
||||
// Default: true
|
||||
"DisableNetwork": true,
|
||||
|
||||
// Make container filesystem read-only
|
||||
// Enhances security by preventing file modifications
|
||||
// Default: true
|
||||
"ReadOnlyFilesystem": true
|
||||
},
|
||||
|
||||
// Server metrics broadcasting configuration
|
||||
"Metrics": {
|
||||
// Minimum CPU usage difference (in percentage points) required to trigger a metrics broadcast
|
||||
// Lower values = more frequent updates, higher values = less network traffic
|
||||
// Default: 1.0 (1%)
|
||||
"CpuDifferenceThreshold": 1.0,
|
||||
|
||||
// Minimum memory usage difference (in bytes) required to trigger a metrics broadcast
|
||||
// Default: 10485760 (10 MB)
|
||||
"MemoryDifferenceThreshold": 10485760,
|
||||
|
||||
// Interval in milliseconds between metrics broadcasts to all connected clients
|
||||
// Lower values = more real-time updates, higher values = less server load
|
||||
// Default: 5000 (5 seconds)
|
||||
"BroadcastIntervalMs": 5000
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user