HMAC Authentication

import hashlib
import hmac
import time
import uuid
import requests
 
MARIE_KEY_ID = "msk_aBcDeFgHiJkLmNoP"
MARIE_SECRET = "your-secret-hex-string"
 
def sign_request(method: str, url: str, body: str = "") -> dict:
    """Generate HMAC signature headers for an M3 Forge API request."""
    timestamp = str(int(time.time()))
    nonce = str(uuid.uuid4())
 
    # Build the canonical message
    message = f"{timestamp}\n{nonce}\n{method.upper()}\n{url}\n{body}"
 
    # Compute HMAC-SHA256
    signature = hmac.new(
        MARIE_SECRET.encode("utf-8"),
        message.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
 
    return {
        "X-Marie-Timestamp": timestamp,
        "X-Marie-Nonce": nonce,
        "X-Marie-Signature": f"sha256={signature}",
        "X-Marie-Key-Id": MARIE_KEY_ID,
    }
 
 
# Example: List workflows
path = "/api/trpc/workflows.list?batch=1"
headers = sign_request("GET", path)
headers["Content-Type"] = "application/json"
 
response = requests.get(
    f"https://your-instance.example.com{path}",
    headers=headers,
)
print(response.json())

import crypto from 'node:crypto';
 
const MARIE_KEY_ID = 'msk_aBcDeFgHiJkLmNoP';
const MARIE_SECRET = 'your-secret-hex-string';
 
function signRequest(method: string, url: string, body = ''): Record<string, string> {
  const timestamp = Math.floor(Date.now() / 1000).toString();
  const nonce = crypto.randomUUID();
 
  // Build the canonical message
  const message = `${timestamp}\n${nonce}\n${method.toUpperCase()}\n${url}\n${body}`;
 
  // Compute HMAC-SHA256
  const signature = crypto
    .createHmac('sha256', MARIE_SECRET)
    .update(message)
    .digest('hex');
 
  return {
    'X-Marie-Timestamp': timestamp,
    'X-Marie-Nonce': nonce,
    'X-Marie-Signature': `sha256=${signature}`,
    'X-Marie-Key-Id': MARIE_KEY_ID,
  };
}
 
// Example: Create a workflow run
const path = '/api/trpc/runs.create';
const body = JSON.stringify({ workflowId: 'wf_abc123', input: {} });
const headers = signRequest('POST', path, body);
 
const response = await fetch(`https://your-instance.example.com${path}`, {
  method: 'POST',
  headers: { ...headers, 'Content-Type': 'application/json' },
  body,
});
console.log(await response.json());

#!/bin/bash
# Generate HMAC headers for a cURL request
 
KEY_ID="msk_aBcDeFgHiJkLmNoP"
SECRET="your-secret-hex-string"
METHOD="GET"
URL="/api/trpc/workflows.list?batch=1"
BODY=""
 
TIMESTAMP=$(date +%s)
NONCE=$(uuidgen)
 
MESSAGE="${TIMESTAMP}\n${NONCE}\n${METHOD}\n${URL}\n${BODY}"
SIGNATURE=$(printf '%b' "$MESSAGE" | openssl dgst -sha256 -hmac "$SECRET" | awk '{print $2}')
 
curl -s "https://your-instance.example.com${URL}" \
  -H "X-Marie-Timestamp: ${TIMESTAMP}" \
  -H "X-Marie-Nonce: ${NONCE}" \
  -H "X-Marie-Signature: sha256=${SIGNATURE}" \
  -H "X-Marie-Key-Id: ${KEY_ID}" \
  -H "Content-Type: application/json"

import hashlib
import hmac
import time
 
SHARED_SECRET = "your-secret-hex-string"
MAX_AGE_SECONDS = 60
 
def verify_webhook(headers: dict, method: str, url: str, body: str) -> bool:
    """Verify an HMAC-signed request from M3 Forge."""
    timestamp = headers.get("X-Marie-Timestamp", "")
    nonce = headers.get("X-Marie-Nonce", "")
    received_sig = headers.get("X-Marie-Signature", "")
 
    # Check timestamp freshness
    try:
        ts = int(timestamp)
    except ValueError:
        return False
 
    if abs(time.time() - ts) > MAX_AGE_SECONDS:
        return False
 
    # Strip the "sha256=" prefix
    if not received_sig.startswith("sha256="):
        return False
    received_hex = received_sig[7:]
 
    # Reconstruct and sign
    message = f"{timestamp}\n{nonce}\n{method.upper()}\n{url}\n{body}"
    expected_hex = hmac.new(
        SHARED_SECRET.encode("utf-8"),
        message.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
 
    # Timing-safe comparison
    return hmac.compare_digest(expected_hex, received_hex)

import crypto from 'node:crypto';
 
const SHARED_SECRET = 'your-secret-hex-string';
const MAX_AGE_SECONDS = 60;
 
function verifyWebhook(
  headers: Record<string, string>,
  method: string,
  url: string,
  body: string,
): boolean {
  const timestamp = headers['x-marie-timestamp'] ?? '';
  const nonce = headers['x-marie-nonce'] ?? '';
  const receivedSig = headers['x-marie-signature'] ?? '';
 
  // Check timestamp freshness
  const ts = parseInt(timestamp, 10);
  if (isNaN(ts) || Math.abs(Date.now() / 1000 - ts) > MAX_AGE_SECONDS) {
    return false;
  }
 
  // Strip the "sha256=" prefix
  if (!receivedSig.startsWith('sha256=')) return false;
  const receivedHex = receivedSig.slice(7);
 
  // Reconstruct and sign
  const message = `${timestamp}\n${nonce}\n${method.toUpperCase()}\n${url}\n${body}`;
  const expectedHex = crypto
    .createHmac('sha256', SHARED_SECRET)
    .update(message)
    .digest('hex');
 
  // Timing-safe comparison
  return crypto.timingSafeEqual(
    Buffer.from(expectedHex, 'hex'),
    Buffer.from(receivedHex, 'hex'),
  );
}