Verifying webhook signatures

import * as crypto from 'crypto'

const AUTH_HEADER_PREFIX = 'auth-'

export interface AuthVerificationResult {
  valid: boolean
  companyId?: string
  matchedHeader?: string
}

export function verifyWebhookAuth(
  headers: Record<string, string | string[] | undefined>,
  privateKeyPems: string[],
  expectedCompanyId: string
): AuthVerificationResult {
  for (const [name, rawValue] of Object.entries(headers)) {
    if (!name.toLowerCase().startsWith(AUTH_HEADER_PREFIX)) continue

    const value = Array.isArray(rawValue) ? rawValue[0] : rawValue
    if (!value) continue

    for (const pem of privateKeyPems) {
      const plaintext = tryDecrypt(value, pem)
      if (plaintext === null) continue

      const separator = plaintext.indexOf('||')
      if (separator < 0) continue

      const companyId = Buffer.from(plaintext.slice(0, separator), 'base64').toString('utf8')
      if (!safeEquals(companyId, expectedCompanyId)) continue
      if (!isAcceptableUtcDay(plaintext.slice(separator + 2))) continue

      return { valid: true, companyId, matchedHeader: name }
    }
  }

  return { valid: false }
}

function tryDecrypt(base64Ciphertext: string, privateKeyPem: string): string | null {
  try {
    const decrypted = crypto.privateDecrypt(
      { key: privateKeyPem, oaepHash: 'sha512' },
      Buffer.from(base64Ciphertext, 'base64')
    )
    return decrypted.toString('utf8')
  } catch {
    return null
  }
}

function safeEquals(a: string, b: string): boolean {
  const bufA = Buffer.from(a, 'utf8')
  const bufB = Buffer.from(b, 'utf8')
  if (bufA.length !== bufB.length) return false
  return crypto.timingSafeEqual(bufA, bufB)
}

function isAcceptableUtcDay(dayPart: string): boolean {
  const oneDayMs = 24 * 60 * 60 * 1000
  const now = Date.now()
  return [now - oneDayMs, now, now + oneDayMs]
    .map((ms) => new Date(ms).getUTCDate().toString())
    .includes(dayPart)
}

import base64
import hmac
from datetime import datetime, timedelta, timezone
from typing import Iterable, Mapping, NamedTuple, Optional

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

AUTH_HEADER_PREFIX = "auth-"

OAEP_SHA512 = padding.OAEP(
    mgf=padding.MGF1(algorithm=hashes.SHA512()),
    algorithm=hashes.SHA512(),
    label=None,
)

class AuthVerificationResult(NamedTuple):
    valid: bool
    company_id: Optional[str] = None
    matched_header: Optional[str] = None

def verify_webhook_auth(
    headers: Mapping[str, str],
    private_key_pems: Iterable[str],
    expected_company_id: str,
) -> AuthVerificationResult:
    keys = [
        serialization.load_pem_private_key(pem.encode("utf-8"), password=None)
        for pem in private_key_pems
    ]

    for name, value in headers.items():
        if not name.lower().startswith(AUTH_HEADER_PREFIX) or not value:
            continue

        try:
            ciphertext = base64.b64decode(value, validate=True)
        except Exception:
            continue

        for key in keys:
            plaintext = _try_decrypt(key, ciphertext)
            if plaintext is None:
                continue

            encoded_company_id, separator, day_part = plaintext.partition("||")
            if not separator:
                continue

            try:
                company_id = base64.b64decode(encoded_company_id).decode("utf-8")
            except Exception:
                continue

            if not hmac.compare_digest(company_id, expected_company_id):
                continue
            if not _is_acceptable_utc_day(day_part):
                continue

            return AuthVerificationResult(True, company_id, name)

    return AuthVerificationResult(False)

def _try_decrypt(key, ciphertext: bytes) -> Optional[str]:
    try:
        return key.decrypt(ciphertext, OAEP_SHA512).decode("utf-8")
    except Exception:
        return None

def _is_acceptable_utc_day(day_part: str) -> bool:
    now = datetime.now(timezone.utc)
    accepted = {str((now + timedelta(days=offset)).day) for offset in (-1, 0, 1)}
    return day_part in accepted

import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
import java.security.KeyFactory;
import java.security.MessageDigest;
import java.security.PrivateKey;
import java.security.spec.MGF1ParameterSpec;
import java.security.spec.PKCS8EncodedKeySpec;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import javax.crypto.Cipher;
import javax.crypto.spec.OAEPParameterSpec;
import javax.crypto.spec.PSource;

public final class WebhookAuthVerifier {

  private static final String AUTH_HEADER_PREFIX = "auth-";

  private static final OAEPParameterSpec OAEP_SHA512 = new OAEPParameterSpec(
      "SHA-512", "MGF1", MGF1ParameterSpec.SHA512, PSource.PSpecified.DEFAULT);

  public record Result(boolean valid, String companyId, String matchedHeader) {
    static Result invalid() {
      return new Result(false, null, null);
    }
  }

  private WebhookAuthVerifier() {}

  public static Result verify(
      Map<String, String> headers, List<String> privateKeyPems, String expectedCompanyId)
      throws GeneralSecurityException {

    List<PrivateKey> keys = new ArrayList<>();
    for (String pem : privateKeyPems) {
      keys.add(loadPkcs8PrivateKey(pem));
    }

    for (Map.Entry<String, String> header : headers.entrySet()) {
      if (!header.getKey().toLowerCase(Locale.ROOT).startsWith(AUTH_HEADER_PREFIX)
          || header.getValue() == null) {
        continue;
      }

      byte[] ciphertext;
      try {
        ciphertext = Base64.getDecoder().decode(header.getValue());
      } catch (IllegalArgumentException e) {
        continue;
      }

      for (PrivateKey key : keys) {
        String plaintext = tryDecrypt(key, ciphertext);
        if (plaintext == null) continue;

        int separator = plaintext.indexOf("||");
        if (separator < 0) continue;

        String companyId;
        try {
          companyId = new String(
              Base64.getDecoder().decode(plaintext.substring(0, separator)),
              StandardCharsets.UTF_8);
        } catch (IllegalArgumentException e) {
          continue;
        }

        if (!MessageDigest.isEqual(
            companyId.getBytes(StandardCharsets.UTF_8),
            expectedCompanyId.getBytes(StandardCharsets.UTF_8))) {
          continue;
        }
        if (!isAcceptableUtcDay(plaintext.substring(separator + 2))) continue;

        return new Result(true, companyId, header.getKey());
      }
    }

    return Result.invalid();
  }

  private static String tryDecrypt(PrivateKey key, byte[] ciphertext) {
    try {
      Cipher cipher = Cipher.getInstance("RSA/ECB/OAEPPadding");
      cipher.init(Cipher.DECRYPT_MODE, key, OAEP_SHA512);
      return new String(cipher.doFinal(ciphertext), StandardCharsets.UTF_8);
    } catch (GeneralSecurityException e) {
      return null;
    }
  }

  private static PrivateKey loadPkcs8PrivateKey(String pem) throws GeneralSecurityException {
    String body = pem
        .replace("-----BEGIN PRIVATE KEY-----", "")
        .replace("-----END PRIVATE KEY-----", "")
        .replaceAll("\\s", "");
    return KeyFactory.getInstance("RSA")
        .generatePrivate(new PKCS8EncodedKeySpec(Base64.getDecoder().decode(body)));
  }

  private static boolean isAcceptableUtcDay(String dayPart) {
    LocalDate today = LocalDate.now(ZoneOffset.UTC);
    for (long offset = -1; offset <= 1; offset++) {
      if (String.valueOf(today.plusDays(offset).getDayOfMonth()).equals(dayPart)) {
        return true;
      }
    }
    return false;
  }
}