Harden webhook ingestion against outsiders

backend/src/CodeSpace.Api/Controllers/WebhooksController.cs

@@ -1,29 +1,49 @@
+using CodeSpace.Api.Extensions;
+using CodeSpace.Api.Http;
+using CodeSpace.Core.Services.Webhooks.Exceptions;
 using CodeSpace.Messages.Commands.Webhooks;
+using CodeSpace.Messages.Enums;
+using CodeSpace.Messages.Queries.Webhooks;
 using MediatR;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.RateLimiting;
 
 namespace CodeSpace.Api.Controllers;
 
+/// <summary>
+/// Provider deliveries. Anonymous — the per-hook signature is the proof — so everything here is bounded before it is
+/// authenticated: the rate limiter answers a flood before the action runs, an id naming no hook is answered before the
+/// body is read, and a body is read only up to <see cref="MaxDeliveryBodyBytes"/>.
+/// </summary>
 [ApiController]
 [Route("api/webhooks")]
 [AllowAnonymous]
+[EnableRateLimiting(WebhookIngressRateLimitExtension.PolicyName)]
 public class WebhooksController : ControllerBase
 {
+    /// <summary>GitHub caps a webhook payload at 25 MB and drops a larger event rather than send it; GitLab truncates its lists well below that. Nothing a provider sends is bigger.</summary>
+    public const long MaxDeliveryBodyBytes = 25L * 1024 * 1024;
+
     private readonly IMediator _mediator;
 
     public WebhooksController(IMediator mediator) { _mediator = mediator; }
 
     [HttpPost("{webhookId:guid}")]
+    [RequestSizeLimit(MaxDeliveryBodyBytes)]
     public async Task<IActionResult> Receive(Guid webhookId, CancellationToken cancellationToken)
     {
+        if (!await HookExistsAsync(ProviderWebhookScope.Repository, webhookId, cancellationToken).ConfigureAwait(false)) return NotFound();
+
         var delivery = await ReadDeliveryAsync(cancellationToken).ConfigureAwait(false);
 
+        if (delivery == null) return StatusCode(StatusCodes.Status413PayloadTooLarge);
+
         return await DispatchAsync(new ReceiveWebhookCommand
         {
             WebhookId = webhookId,
-            Body = delivery.Body,
-            Headers = delivery.Headers
+            Body = delivery.Value.Body,
+            Headers = delivery.Value.Headers
         }, cancellationToken).ConfigureAwait(false);
     }
 
@@ -34,22 +54,32 @@ public async Task<IActionResult> Receive(Guid webhookId, CancellationToken cance
     /// the wrong table and answered 404.
     /// </summary>
     [HttpPost("connection/{connectionWebhookId:guid}")]
+    [RequestSizeLimit(MaxDeliveryBodyBytes)]
     public async Task<IActionResult> ReceiveConnection(Guid connectionWebhookId, CancellationToken cancellationToken)
     {
+        if (!await HookExistsAsync(ProviderWebhookScope.Connection, connectionWebhookId, cancellationToken).ConfigureAwait(false)) return NotFound();
+
         var delivery = await ReadDeliveryAsync(cancellationToken).ConfigureAwait(false);
 
+        if (delivery == null) return StatusCode(StatusCodes.Status413PayloadTooLarge);
+
         return await DispatchAsync(new ReceiveConnectionWebhookCommand
         {
             ConnectionWebhookId = connectionWebhookId,
-            Body = delivery.Body,
-            Headers = delivery.Headers
+            Body = delivery.Value.Body,
+            Headers = delivery.Value.Headers
         }, cancellationToken).ConfigureAwait(false);
     }
 
-    private async Task<(string Body, Dictionary<string, string> Headers)> ReadDeliveryAsync(CancellationToken cancellationToken)
+    private async Task<bool> HookExistsAsync(ProviderWebhookScope scope, Guid webhookId, CancellationToken cancellationToken) =>
+        await _mediator.Send(new WebhookExistsQuery { WebhookId = webhookId, Scope = scope }, cancellationToken).ConfigureAwait(false);
+
+    /// <summary>Null when the body is larger than <see cref="MaxDeliveryBodyBytes"/>.</summary>
+    private async Task<(string Body, Dictionary<string, string> Headers)?> ReadDeliveryAsync(CancellationToken cancellationToken)
     {
-        using var reader = new StreamReader(Request.Body);
-        var body = await reader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
+        var body = await Request.ReadBoundedBodyAsync(MaxDeliveryBodyBytes, cancellationToken).ConfigureAwait(false);
+
+        if (body == null) return null;
 
         return (body, Request.Headers.ToDictionary(h => h.Key, h => h.Value.ToString()));
     }
@@ -58,7 +88,8 @@ public async Task<IActionResult> ReceiveConnection(Guid connectionWebhookId, Can
     /// The one status mapping, shared by both routes so a group delivery and a project delivery
     /// cannot answer differently for the same failure. 200 for anything the service handled — a
     /// provider reads 5xx as "retry" and GitLab disables a hook that keeps failing, which would take
-    /// out every repository the hook covers.
+    /// out every repository the hook covers. 400 for a signed delivery missing the id its provider
+    /// always sends — malformed for that provider, and never something the provider itself sends.
     /// </summary>
     private async Task<IActionResult> DispatchAsync(IRequest<Unit> command, CancellationToken cancellationToken)
     {
@@ -72,6 +103,10 @@ private async Task<IActionResult> DispatchAsync(IRequest<Unit> command, Cancella
         {
             return Unauthorized();
         }
+        catch (WebhookDeliveryUnidentifiedException)
+        {
+            return BadRequest();
+        }
         catch (InvalidOperationException)
         {
             return NotFound();

backend/src/CodeSpace.Api/Controllers/WorkflowCallbacksController.cs

@@ -1,30 +1,44 @@
+using CodeSpace.Api.Extensions;
+using CodeSpace.Api.Http;
 using CodeSpace.Messages.Commands.Workflows;
 using MediatR;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.RateLimiting;
 
 namespace CodeSpace.Api.Controllers;
 
 /// <summary>
 /// Unauthenticated callback surface: an external system resumes a run parked on
 /// <c>flow.wait_callback</c> by POSTing to the tokened URL the run-detail UI shows. The token is
 /// the bearer secret (high-entropy, single-use while the wait is pending) — there is no team /
-/// user context, mirroring <see cref="WebhooksController"/>.
+/// user context, mirroring <see cref="WebhooksController"/>, and bounded the same way: rate-limited
+/// per address and then per (token, address), and read only up to <see cref="MaxCallbackBodyBytes"/>.
 /// </summary>
 [ApiController]
 [Route("api/workflows/callbacks")]
 [AllowAnonymous]
+[EnableRateLimiting(WebhookIngressRateLimitExtension.PolicyName)]
 public class WorkflowCallbacksController : ControllerBase
 {
+    /// <summary>The body becomes the waiting node's output, stored in a jsonb row and read by every later step. A resume signal has no business being larger.</summary>
+    public const long MaxCallbackBodyBytes = 1024 * 1024;
+
     private readonly IMediator _mediator;
 
     public WorkflowCallbacksController(IMediator mediator) { _mediator = mediator; }
 
-    [HttpPost("{token}")]
+    /// <summary>
+    /// The route admits only the shape the engine mints — <c>Guid.NewGuid().ToString("N")</c>, 32 characters. A token no
+    /// wait can hold is answered 404 by routing, before the limiter keys a partition on it or anything looks it up.
+    /// </summary>
+    [HttpPost("{token:length(32)}")]
+    [RequestSizeLimit(MaxCallbackBodyBytes)]
     public async Task<IActionResult> Resume([FromRoute] string token, CancellationToken cancellationToken)
     {
-        using var reader = new StreamReader(Request.Body);
-        var body = await reader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
+        var body = await Request.ReadBoundedBodyAsync(MaxCallbackBodyBytes, cancellationToken).ConfigureAwait(false);
+
+        if (body == null) return StatusCode(StatusCodes.Status413PayloadTooLarge);
 
         var resumed = await _mediator.Send(new ResumeWorkflowCallbackCommand { Token = token, Body = body }, cancellationToken).ConfigureAwait(false);
 

backend/src/CodeSpace.Api/Extensions/WebhookIngressRateLimitExtension.cs

@@ -0,0 +1,76 @@
+using System.Threading.RateLimiting;
+using Microsoft.AspNetCore.RateLimiting;
+
+namespace CodeSpace.Api.Extensions;
+
+/// <summary>
+/// A ceiling on what one source can make the anonymous ingress endpoints do — provider webhooks and workflow callbacks.
+/// Both are reached before anything is authenticated, and both used to answer every request without a limit.
+///
+/// <para>Two limits, in order. First per remote address, across every ingress endpoint: the app's
+/// <c>GlobalLimiter</c>, which ASP.NET consults before any endpoint policy and which passes every other endpoint
+/// through untouched. Then per (target, address), where the target is the hook id or callback token in the URL. The
+/// order is the point. The target is a value the sender picks, and each distinct one gets a limiter of its own that
+/// outlives its window, so a sender changing the target on every request was never limited and left a limiter behind
+/// each time. Checked first, the address budget runs out, and it bounds how many per-target limiters one sender can
+/// create. The routes bound each target's length: a hook id is a GUID, a callback token the 32 characters the engine
+/// mints.</para>
+///
+/// <para>Sized for a provider, not a person. A group / organization hook subscribes to every event — pipeline and job
+/// events included — so a busy group sends hundreds a minute to one hook, and a self-managed instance sends every hook's
+/// traffic from one address. A provider reads a refusal as a failed delivery: GitLab disables a hook that keeps failing.
+/// So the per-hook ceiling sits well above one busy hook, and the per-address one above several of them at once.</para>
+///
+/// <para>The address is the connection's own; nothing here reads forwarded headers. Behind a proxy that hides the
+/// client address, every sender shares one address and so one budget.</para>
+/// </summary>
+public static class WebhookIngressRateLimitExtension
+{
+    public const string PolicyName = "webhook-ingress";
+
+    /// <summary>Per (target, address) per window.</summary>
+    public const int PermitsPerWindow = 2000;
+
+    /// <summary>Per address per window, across every ingress target — five hooks at their full budget.</summary>
+    public const int SourcePermitsPerWindow = 10_000;
+
+    public static readonly TimeSpan Window = TimeSpan.FromMinutes(1);
+
+    /// <summary>The route values that name what a request is addressed to, in the order the ingress routes declare them.</summary>
+    private static readonly string[] TargetRouteValues = { "webhookId", "connectionWebhookId", "token" };
+
+    /// <summary>The one partition every non-ingress request shares in the global limiter — no limit. Never an address.</summary>
+    private const string NotIngress = "not-ingress";
+
+    public static void AddWebhookIngressRateLimit(this IServiceCollection services)
+    {
+        services.AddRateLimiter(options =>
+        {
+            options.RejectionStatusCode = StatusCodes.Status429TooManyRequests;
+            options.GlobalLimiter = PartitionedRateLimiter.Create<HttpContext, string>(SourcePartition);
+            options.AddPolicy(PolicyName, context => FixedWindow(PartitionKey(context), PermitsPerWindow));
+        });
+    }
+
+    /// <summary>The address's budget for an ingress request; no limit for anything else the app serves.</summary>
+    private static RateLimitPartition<string> SourcePartition(HttpContext context) =>
+        IsIngress(context) ? FixedWindow(SourcePartitionKey(context), SourcePermitsPerWindow) : RateLimitPartition.GetNoLimiter(NotIngress);
+
+    /// <summary>An endpoint that opted into the ingress policy — the same marker the per-target limit is applied by.</summary>
+    private static bool IsIngress(HttpContext context) =>
+        context.GetEndpoint()?.Metadata.GetMetadata<EnableRateLimitingAttribute>()?.PolicyName == PolicyName;
+
+    private static RateLimitPartition<string> FixedWindow(string key, int permits) =>
+        RateLimitPartition.GetFixedWindowLimiter(key, _ => new FixedWindowRateLimiterOptions { PermitLimit = permits, Window = Window, QueueLimit = 0 });
+
+    /// <summary>Who sent the request, and nothing the sender chose.</summary>
+    internal static string SourcePartitionKey(HttpContext context) => context.Connection.RemoteIpAddress?.ToString() ?? "unknown";
+
+    /// <summary>"{target}|{address}". A request with no target route value shares one partition per address.</summary>
+    internal static string PartitionKey(HttpContext context)
+    {
+        var target = TargetRouteValues.Select(name => context.GetRouteValue(name)?.ToString()).FirstOrDefault(value => !string.IsNullOrEmpty(value)) ?? "-";
+
+        return $"{target}|{SourcePartitionKey(context)}";
+    }
+}

backend/src/CodeSpace.Api/Http/BoundedRequestBody.cs

@@ -0,0 +1,46 @@
+namespace CodeSpace.Api.Http;
+
+/// <summary>
+/// Reads an anonymous endpoint's body only up to a limit. <c>[RequestSizeLimit]</c> makes Kestrel stop at the same
+/// limit, but it does so by THROWING from the read, which the global exception filter would turn into a 500, and a host
+/// without the size feature does not stop at all. This answers "too large" the same way under either.
+/// </summary>
+public static class BoundedRequestBody
+{
+    /// <summary>The body as text, or null when it is longer than <paramref name="maxBytes"/>.</summary>
+    public static async Task<string?> ReadBoundedBodyAsync(this HttpRequest request, long maxBytes, CancellationToken cancellationToken)
+    {
+        if (request.ContentLength > maxBytes) return null;
+
+        try
+        {
+            using var buffer = await ReadUpToAsync(request.Body, maxBytes + 1, cancellationToken).ConfigureAwait(false);
+            if (buffer.Length > maxBytes) return null;
+
+            using var reader = new StreamReader(buffer);
+            return await reader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (BadHttpRequestException ex) when (ex.StatusCode == StatusCodes.Status413PayloadTooLarge)
+        {
+            return null;
+        }
+    }
+
+    /// <summary>At most <paramref name="limit"/> bytes of the body, rewound for reading.</summary>
+    private static async Task<MemoryStream> ReadUpToAsync(Stream body, long limit, CancellationToken cancellationToken)
+    {
+        var buffer = new MemoryStream();
+        var chunk = new byte[81920];
+
+        while (buffer.Length < limit)
+        {
+            var read = await body.ReadAsync(chunk.AsMemory(0, (int)Math.Min(chunk.Length, limit - buffer.Length)), cancellationToken).ConfigureAwait(false);
+            if (read == 0) break;
+
+            buffer.Write(chunk, 0, read);
+        }
+
+        buffer.Position = 0;
+        return buffer;
+    }
+}

backend/src/CodeSpace.Api/Startup.cs

@@ -95,6 +95,7 @@ public void ConfigureServices(IServiceCollection services)
         services.Configure<HostOptions>(o => o.ShutdownTimeout = ShutdownSettings.ResolveDrainTimeout());
 
         services.AddAnonymousRateLimit();
+        services.AddWebhookIngressRateLimit();
 
         services.AddCustomAuthentication(Configuration, Environment);
     }

backend/src/CodeSpace.Core/Handlers/CommandHandlers/Webhooks/PurgeExpiredWebhookClaimsCommandHandler.cs

@@ -0,0 +1,20 @@
+using CodeSpace.Core.Services.Webhooks;
+using CodeSpace.Messages.Commands.Webhooks;
+using MediatR;
+
+namespace CodeSpace.Core.Handlers.CommandHandlers.Webhooks;
+
+/// <summary>Rule 16 — thin handler. The expiry predicate and the batched delete live in <see cref="IWebhookClaimStore"/>.</summary>
+public sealed class PurgeExpiredWebhookClaimsCommandHandler : IRequestHandler<PurgeExpiredWebhookClaimsCommand, PurgeExpiredWebhookClaimsResponse>
+{
+    private readonly IWebhookClaimStore _claims;
+
+    public PurgeExpiredWebhookClaimsCommandHandler(IWebhookClaimStore claims) { _claims = claims; }
+
+    public async Task<PurgeExpiredWebhookClaimsResponse> Handle(PurgeExpiredWebhookClaimsCommand request, CancellationToken cancellationToken)
+    {
+        var deleted = await _claims.PurgeExpiredAsync(cancellationToken).ConfigureAwait(false);
+
+        return new PurgeExpiredWebhookClaimsResponse { Deleted = deleted };
+    }
+}

backend/src/CodeSpace.Core/Handlers/QueryHandlers/Webhooks/WebhookExistsQueryHandler.cs

@@ -0,0 +1,15 @@
+using CodeSpace.Core.Services.Webhooks;
+using CodeSpace.Messages.Queries.Webhooks;
+using MediatR;
+
+namespace CodeSpace.Core.Handlers.QueryHandlers.Webhooks;
+
+public sealed class WebhookExistsQueryHandler : IRequestHandler<WebhookExistsQuery, bool>
+{
+    private readonly IWebhookLookup _lookup;
+
+    public WebhookExistsQueryHandler(IWebhookLookup lookup) { _lookup = lookup; }
+
+    public async Task<bool> Handle(WebhookExistsQuery request, CancellationToken cancellationToken) =>
+        await _lookup.ExistsAsync(request.Scope, request.WebhookId, cancellationToken).ConfigureAwait(false);
+}

backend/src/CodeSpace.Core/Jobs/RecurringJobs/WebhookClaimPurgeRecurringJob.cs

@@ -0,0 +1,24 @@
+using CodeSpace.Messages.Commands.Webhooks;
+using MediatR;
+
+namespace CodeSpace.Core.Jobs.RecurringJobs;
+
+/// <summary>
+/// Every 5 minutes, dispatches <see cref="PurgeExpiredWebhookClaimsCommand"/> to clear expired <c>webhook_claim</c> rows.
+/// Thin Mediator dispatcher (Rule 14) — the delete lives in <see cref="Services.Webhooks.IWebhookClaimStore"/>.
+///
+/// <para>Replaces a bounded purge each claim used to run inside its own delivery's transaction. That held the purged
+/// rows' locks until the delivery committed, and two deliveries that each purged a batch and then claimed a key in the
+/// other's batch deadlocked — one of them failing with a 5xx the provider does not redeliver on its own.</para>
+/// </summary>
+public sealed class WebhookClaimPurgeRecurringJob : IRecurringJob
+{
+    private readonly IMediator _mediator;
+
+    public WebhookClaimPurgeRecurringJob(IMediator mediator) { _mediator = mediator; }
+
+    public string JobId => nameof(WebhookClaimPurgeRecurringJob);
+    public string CronExpression => "*/5 * * * *";   // every 5 minutes
+
+    public async Task Execute() => await _mediator.Send(new PurgeExpiredWebhookClaimsCommand()).ConfigureAwait(false);
+}

backend/src/CodeSpace.Core/Persistence/DbUpFiles/0244_webhook_claim.sql

@@ -0,0 +1,35 @@
+-- 0244_webhook_claim.sql
+--
+-- A key held for a window, so that "this already happened moments ago" is settled by the database instead of by a read
+-- that two deliveries landing together can both pass. Two webhook-ingress decisions ride it:
+--
+--   body:{hook}:{sha256}           A signed GitHub body this hook accepted, held by its X-GitHub-Delivery id. GitHub's
+--                                  signature covers the body only, so a captured body stays valid for as long as the
+--                                  secret does; posting it again under a fresh delivery id is a replay. The same id
+--                                  (a provider redelivery) re-claims it and is deduplicated further on, per activation.
+--   pr-debounce:{activation}:{repository}:{number}:{head}
+--                                  The pull request head commit that last started a run from this activation. A
+--                                  close/reopen loop re-sends a head that already ran, so it starts one run inside the
+--                                  window, not one per reopen; a push moves the head, so new commits always run.
+--
+-- A claim is taken inside the delivery's own transaction, so a delivery that rolls back releases it and the provider's
+-- retry is not refused, and it touches no row but its own key. An expired claim is taken over in place by the next claim
+-- on its key; the recurring WebhookClaimPurgeRecurringJob deletes the rest, outside any delivery. (A delivery that
+-- purged expired rows itself held their locks until it committed, and two such deliveries could deadlock.)
+--
+-- holder is TEXT, not a bounded VARCHAR: it is the delivery id from an unsigned header, and a value too long for a
+-- column would fail the delivery with a 500 instead of being refused or deduplicated.
+--
+-- New table, no backfill. Idempotent (IF NOT EXISTS). An older pod never reads or writes it.
+
+CREATE TABLE IF NOT EXISTS webhook_claim (
+    claim_key  TEXT         PRIMARY KEY,
+    holder     TEXT         NOT NULL,
+    claimed_at TIMESTAMPTZ  NOT NULL,
+    expires_at TIMESTAMPTZ  NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS ix_webhook_claim_expires_at ON webhook_claim (expires_at);
+
+COMMENT ON TABLE webhook_claim IS
+    'A key held until expires_at by one holder: a replayed webhook body (body:) or a debounced pull request head (pr-debounce:). Taken in the delivery transaction; expired rows are taken over in place by a claim on the same key and deleted by a recurring sweep.';