Verify GitLab webhook signature headers, replay defense, rotation

// Minimal GitLab token header verification (Express)
// npm i express
import crypto from "crypto";
import express from "express";

const app = express();
app.use(express.json({ limit: "2mb" }));

app.post("/webhook", (req, res) => {
  const secret = process.env.WEBHOOK_SECRET ?? "replace_me";
  const token = req.header("x-gitlab-token") ?? "";
  if (!token) return res.status(401).send("missing x-gitlab-token");
  if (!timingSafeEqualString(token, secret)) return res.status(401).send("bad token");
  return res.sendStatus(202);
});

app.listen(3000, () => console.log("listening on :3000"));


function timingSafeEqualString(a, b) {
  const aBuf = Buffer.from(a, "utf8");
  const bBuf = Buffer.from(b, "utf8");
  if (aBuf.length !== bBuf.length) return false;
  return crypto.timingSafeEqual(aBuf, bBuf);
}

# Minimal GitLab token header verification (FastAPI)
# pip install fastapi uvicorn
import hmac
import os
from fastapi import FastAPI, Request, Response

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request) -> Response:
    secret = os.getenv("WEBHOOK_SECRET") or "replace_me"
    token = request.headers.get("x-gitlab-token", "")
    if not token:
        return Response("missing x-gitlab-token", status_code=401)
    if not hmac.compare_digest(token, secret):
        return Response("bad token", status_code=401)
    return Response(status_code=202)

// Replay protection: timestamp max-age (when available) + dedupe window (in-memory example)
import crypto from "crypto";

const seen = new Map(); // key -> expiresAtMs

function purgeSeen() {
  const now = Date.now();
  for (const [key, expiresAt] of seen.entries()) if (expiresAt <= now) seen.delete(key);
}

// Inside your verified webhook handler:
purgeSeen();
const sig = req.header("x-gitlab-token") ?? "";

// Prefer a provider delivery/event ID when available. Otherwise hash signature + raw body.
const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");
const key = crypto.createHash("sha256").update(sig).update(".").update(rawBody).digest("hex");
const ttlMs = 10 * 60 * 1000;
if (seen.has(key)) return res.status(409).send("replay detected");
seen.set(key, Date.now() + ttlMs);

# Replay protection: timestamp max-age (when available) + dedupe window (in-memory example)
import hashlib
import time

seen: dict[str, float] = {}  # key -> expires_at_epoch_seconds

def purge_seen() -> None:
    now = time.time()
    for k, exp in list(seen.items()):
        if exp <= now:
            del seen[k]

# Inside your verified webhook handler:
purge_seen()
sig = request.headers.get("x-gitlab-token", "")

raw_body = await request.body()
key = hashlib.sha256((sig + ".").encode("utf-8") + raw_body).hexdigest()
ttl_seconds = 10 * 60
if key in seen:
    return Response("replay detected", status_code=409)
seen[key] = time.time() + ttl_seconds

// Secret rotation for token headers: accept {current, previous} during overlap
import crypto from "crypto";

function verifyTokenWithSecrets(headers, secrets) {
  const token = headers["x-gitlab-token"] ?? "";
  if (!token) return false;
  for (const secret of secrets) {
    if (timingSafeEqualString(token, secret)) return true;
  }
  return false;
}

const secrets = [process.env.WEBHOOK_SECRET_CURRENT, process.env.WEBHOOK_SECRET_PREVIOUS].filter(Boolean);

function timingSafeEqualString(a, b) {
  const aBuf = Buffer.from(a, "utf8");
  const bBuf = Buffer.from(b, "utf8");
  if (aBuf.length !== bBuf.length) return false;
  return crypto.timingSafeEqual(aBuf, bBuf);
}

# Secret rotation for token headers: accept {current, previous} during overlap
import hmac
import os

def verify_token_with_secrets(headers: dict, secrets: list[str]) -> bool:
    token = headers.get("x-gitlab-token", "")
    if not token:
        return False
    for secret in secrets:
        if hmac.compare_digest(token, secret):
            return True
    return False

secrets = [os.getenv("WEBHOOK_SECRET_CURRENT"), os.getenv("WEBHOOK_SECRET_PREVIOUS")]
secrets = [s for s in secrets if s]