Verificar assinaturas de webhook

Digital-Signature: <base64(ECDSA P-256 SHA-256(rawBody))>

curl -s https://api.clearpago.com.br/public-key

-----BEGIN PUBLIC KEY-----
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE...
-----END PUBLIC KEY-----

app.use('/webhooks', (req, res, next) => {
  let data = '';
  req.on('data', chunk => { data += chunk; });
  req.on('end', () => {
    req.rawBody = data;
    next();
  });
});

@app.post("/webhooks/pix")
async def receive_webhook(request: Request):
    raw_body = await request.body()  # bytes brutos
    signature = request.headers.get("digital-signature")
    # ...

const crypto = require('crypto');

function verifyWebhookSignature(publicKeyPem, rawBody, signatureBase64) {
  const verify = crypto.createVerify('SHA256');
  verify.update(rawBody);
  return verify.verify(
    { key: publicKeyPem, dsaEncoding: 'ieee-p1363' },
    Buffer.from(signatureBase64, 'base64')
  );
}

// Uso no handler
app.post('/webhooks/pix', (req, res) => {
  const isValid = verifyWebhookSignature(
    cachedPublicKey,
    req.rawBody,
    req.headers['digital-signature']
  );

  if (!isValid) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  res.status(200).send();
  processEvent(req.body); // assíncrono
});

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.exceptions import InvalidSignature
import base64

def verify_webhook_signature(public_key_pem: bytes, raw_body: bytes, signature_b64: str) -> bool:
    public_key = serialization.load_pem_public_key(public_key_pem)
    signature = base64.b64decode(signature_b64)
    try:
        public_key.verify(signature, raw_body, ec.ECDSA(hashes.SHA256()))
        return True
    except InvalidSignature:
        return False

# Uso no handler
@app.post("/webhooks/pix")
async def receive_webhook(request: Request):
    raw_body = await request.body()
    signature = request.headers.get("digital-signature", "")

    if not verify_webhook_signature(cached_public_key, raw_body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    background_tasks.add_task(process_event, await request.json())
    return Response(status_code=200)

import (
    "crypto/ecdsa"
    "crypto/sha256"
    "encoding/base64"
    "encoding/asn1"
    "math/big"
)

func verifySignature(pubKey *ecdsa.PublicKey, rawBody []byte, signatureB64 string) bool {
    sigBytes, err := base64.StdEncoding.DecodeString(signatureB64)
    if err != nil {
        return false
    }
    hash := sha256.Sum256(rawBody)

    // Parse DER ASN.1 se necessário, ou IEEE P1363 conforme o formato
    var sig struct{ R, S *big.Int }
    if _, err := asn1.Unmarshal(sigBytes, &sig); err != nil {
        return false
    }
    return ecdsa.Verify(pubKey, hash[:], sig.R, sig.S)
}