harn-vm 0.10.162

use harn_kernel::pure::sha256_hex;
use std::collections::BTreeMap;

use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::Engine;
use hmac::{Hmac, KeyInit, Mac};
use serde_json::json;
use sha2::Sha256;
use subtle::ConstantTimeEq;
use time::{Duration, OffsetDateTime};

use crate::event_log::{EventLog, LogEvent, Topic};
use crate::triggers::ProviderId;

use super::shared::header_value;
use super::ConnectorError;

pub const SIGNATURE_VERIFY_AUDIT_TOPIC: &str = "audit.signature_verify";
pub const DEFAULT_GITHUB_SIGNATURE_HEADER: &str = "x-hub-signature-256";
pub const DEFAULT_LINEAR_SIGNATURE_HEADER: &str = "linear-signature";
pub const DEFAULT_SLACK_SIGNATURE_HEADER: &str = "x-slack-signature";
pub const DEFAULT_SLACK_TIMESTAMP_HEADER: &str = "x-slack-request-timestamp";
pub const DEFAULT_STRIPE_SIGNATURE_HEADER: &str = "stripe-signature";
pub const DEFAULT_STANDARD_WEBHOOKS_ID_HEADER: &str = "webhook-id";
pub const DEFAULT_STANDARD_WEBHOOKS_SIGNATURE_HEADER: &str = "webhook-signature";
pub const DEFAULT_STANDARD_WEBHOOKS_TIMESTAMP_HEADER: &str = "webhook-timestamp";
pub const DEFAULT_CANONICAL_AUTHORIZATION_HEADER: &str = "authorization";
pub const DEFAULT_CANONICAL_HMAC_SCHEME: &str = "HMAC-SHA256";

/// Supported HMAC signature header conventions for inbound webhook providers.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HmacSignatureStyle<'a> {
    GitHub {
        signature_header: &'a str,
        prefix: &'a str,
    },
    Linear {
        signature_header: &'a str,
    },
    Slack {
        signature_header: &'a str,
        timestamp_header: &'a str,
        version: &'a str,
    },
    Stripe {
        signature_header: &'a str,
        version: &'a str,
    },
    StandardWebhooks {
        id_header: &'a str,
        signature_header: &'a str,
        timestamp_header: &'a str,
        version: &'a str,
    },
    CanonicalRequest {
        authorization_header: &'a str,
        scheme: &'a str,
    },
}

impl<'a> HmacSignatureStyle<'a> {
    pub fn github() -> Self {
        Self::GitHub {
            signature_header: DEFAULT_GITHUB_SIGNATURE_HEADER,
            prefix: "sha256=",
        }
    }

    pub fn stripe() -> Self {
        Self::Stripe {
            signature_header: DEFAULT_STRIPE_SIGNATURE_HEADER,
            version: "v1",
        }
    }

    pub fn linear() -> Self {
        Self::Linear {
            signature_header: DEFAULT_LINEAR_SIGNATURE_HEADER,
        }
    }

    pub fn slack() -> Self {
        Self::Slack {
            signature_header: DEFAULT_SLACK_SIGNATURE_HEADER,
            timestamp_header: DEFAULT_SLACK_TIMESTAMP_HEADER,
            version: "v0",
        }
    }

    pub fn standard_webhooks() -> Self {
        Self::StandardWebhooks {
            id_header: DEFAULT_STANDARD_WEBHOOKS_ID_HEADER,
            signature_header: DEFAULT_STANDARD_WEBHOOKS_SIGNATURE_HEADER,
            timestamp_header: DEFAULT_STANDARD_WEBHOOKS_TIMESTAMP_HEADER,
            version: "v1",
        }
    }

    pub fn canonical_request() -> Self {
        Self::CanonicalRequest {
            authorization_header: DEFAULT_CANONICAL_AUTHORIZATION_HEADER,
            scheme: DEFAULT_CANONICAL_HMAC_SCHEME,
        }
    }

    fn label(self) -> &'static str {
        match self {
            Self::GitHub { .. } => "github",
            Self::Linear { .. } => "linear",
            Self::Slack { .. } => "slack",
            Self::Stripe { .. } => "stripe",
            Self::StandardWebhooks { .. } => "standard_webhooks",
            Self::CanonicalRequest { .. } => "canonical_request",
        }
    }
}

/// Verify an HMAC-signed raw request body against one of the supported webhook
/// header conventions.
pub async fn verify_hmac_signed<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Option<Duration>,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    match style {
        HmacSignatureStyle::GitHub {
            signature_header,
            prefix,
        } => {
            verify_github(
                event_log,
                provider,
                style,
                signature_header,
                prefix,
                body,
                headers,
                secret,
                now,
            )
            .await
        }
        HmacSignatureStyle::Linear { signature_header } => {
            verify_linear(
                event_log,
                provider,
                style,
                signature_header,
                body,
                headers,
                secret,
                timestamp_window,
                now,
            )
            .await
        }
        HmacSignatureStyle::Slack {
            signature_header,
            timestamp_header,
            version,
        } => {
            verify_slack(
                event_log,
                provider,
                style,
                signature_header,
                timestamp_header,
                version,
                body,
                headers,
                secret,
                timestamp_window,
                now,
            )
            .await
        }
        HmacSignatureStyle::Stripe {
            signature_header,
            version,
        } => {
            verify_stripe(
                event_log,
                provider,
                style,
                signature_header,
                version,
                body,
                headers,
                secret,
                timestamp_window,
                now,
            )
            .await
        }
        HmacSignatureStyle::StandardWebhooks {
            id_header,
            signature_header,
            timestamp_header,
            version,
        } => {
            verify_standard_webhooks(
                event_log,
                provider,
                style,
                id_header,
                signature_header,
                timestamp_header,
                version,
                body,
                headers,
                secret,
                timestamp_window,
                now,
            )
            .await
        }
        HmacSignatureStyle::CanonicalRequest { .. } => {
            let error = ConnectorError::Unsupported(
                "canonical-request verification requires method + path context".to_string(),
            );
            reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                timestamp_window,
            )
            .await
        }
    }
}

/// Verify an `Authorization: HMAC-SHA256 ...` header against a canonical
/// request string built from method, path, timestamp, and a SHA-256 body hash.
#[allow(clippy::too_many_arguments)]
pub async fn verify_hmac_authorization<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    method: &str,
    path: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Duration,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let style = HmacSignatureStyle::canonical_request();
    let HmacSignatureStyle::CanonicalRequest {
        authorization_header,
        scheme,
    } = style
    else {
        unreachable!("canonical_request constructor must return CanonicalRequest");
    };
    verify_canonical_request(
        event_log,
        provider,
        style,
        authorization_header,
        scheme,
        method,
        path,
        body,
        headers,
        secret,
        timestamp_window,
        now,
    )
    .await
}

#[allow(clippy::too_many_arguments)]
async fn verify_github<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    signature_header: &str,
    prefix: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let header = required_header(headers, signature_header).map_err(|error| {
        ConnectorError::MissingHeader(match error {
            ConnectorError::MissingHeader(name) => name,
            other => other.to_string(),
        })
    });
    let header = match header {
        Ok(value) => value,
        Err(error) => return reject(event_log, provider, style, &error, now, None, None).await,
    };

    let encoded = match header.strip_prefix(prefix) {
        Some(value) => value,
        None => {
            let error = ConnectorError::InvalidHeader {
                name: signature_header.to_string(),
                detail: format!("expected `{prefix}` prefix"),
            };
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };
    let provided = match hex::decode(encoded) {
        Ok(value) => value,
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: signature_header.to_string(),
                detail: error.to_string(),
            };
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };

    let expected = hmac_sha256(secret.as_bytes(), body);
    if secure_eq(&expected, &provided) {
        Ok(())
    } else {
        let error =
            ConnectorError::invalid_signature("signature did not match the raw request body");
        reject(event_log, provider, style, &error, now, None, None).await
    }
}

#[allow(clippy::too_many_arguments)]
async fn verify_linear<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    signature_header: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Option<Duration>,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let window = match timestamp_window {
        Some(window) => window,
        None => {
            let error = ConnectorError::Unsupported(
                "linear-style signature verification requires a timestamp window".to_string(),
            );
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };

    let payload: serde_json::Value = match serde_json::from_slice(body) {
        Ok(payload) => payload,
        Err(error) => {
            let error = ConnectorError::Json(error.to_string());
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    let timestamp_ms = match payload.get("webhookTimestamp").and_then(parse_json_i64ish) {
        Some(timestamp_ms) => timestamp_ms,
        None => {
            let error = ConnectorError::InvalidHeader {
                name: "webhookTimestamp".to_string(),
                detail: "missing from payload body".to_string(),
            };
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    let timestamp = match offset_from_unix_millis(timestamp_ms) {
        Ok(timestamp) => timestamp,
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: "webhookTimestamp".to_string(),
                detail: error,
            };
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    ensure_timestamp_within_window(event_log, provider, style, timestamp, window, now).await?;

    let header = match required_header(headers, signature_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await;
        }
    };
    let provided = match hex::decode(header.trim()) {
        Ok(value) => value,
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: signature_header.to_string(),
                detail: error.to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await;
        }
    };

    let expected = hmac_sha256(secret.as_bytes(), body);
    if secure_eq(&expected, &provided) {
        Ok(())
    } else {
        let error =
            ConnectorError::invalid_signature("signature did not match the raw request body");
        reject(
            event_log,
            provider,
            style,
            &error,
            now,
            Some(timestamp),
            Some(window),
        )
        .await
    }
}

#[allow(clippy::too_many_arguments)]
async fn verify_slack<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    signature_header: &str,
    timestamp_header: &str,
    version: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Option<Duration>,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let window = match timestamp_window {
        Some(window) => window,
        None => {
            let error = ConnectorError::Unsupported(
                "slack-style signature verification requires a timestamp window".to_string(),
            );
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };

    let timestamp_raw = match required_header(headers, timestamp_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    let timestamp = match timestamp_raw.parse::<i64>() {
        Ok(raw) => match OffsetDateTime::from_unix_timestamp(raw) {
            Ok(timestamp) => timestamp,
            Err(error) => {
                let error = ConnectorError::InvalidHeader {
                    name: timestamp_header.to_string(),
                    detail: error.to_string(),
                };
                return reject(event_log, provider, style, &error, now, None, Some(window)).await;
            }
        },
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: timestamp_header.to_string(),
                detail: error.to_string(),
            };
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    ensure_timestamp_within_window(event_log, provider, style, timestamp, window, now).await?;

    let header = match required_header(headers, signature_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await;
        }
    };
    let prefix = format!("{version}=");
    let encoded = match header.strip_prefix(prefix.as_str()) {
        Some(value) => value,
        None => {
            let error = ConnectorError::InvalidHeader {
                name: signature_header.to_string(),
                detail: format!("expected `{prefix}` prefix"),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await;
        }
    };
    let provided = match hex::decode(encoded) {
        Ok(value) => value,
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: signature_header.to_string(),
                detail: error.to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await;
        }
    };

    let mut signed = Vec::with_capacity(version.len() + timestamp_raw.len() + body.len() + 2);
    signed.extend_from_slice(version.as_bytes());
    signed.push(b':');
    signed.extend_from_slice(timestamp_raw.as_bytes());
    signed.push(b':');
    signed.extend_from_slice(body);
    let expected = hmac_sha256(secret.as_bytes(), &signed);
    if secure_eq(&expected, &provided) {
        Ok(())
    } else {
        let error = ConnectorError::invalid_signature("slack signature mismatch");
        reject(
            event_log,
            provider,
            style,
            &error,
            now,
            Some(timestamp),
            Some(window),
        )
        .await
    }
}

#[allow(clippy::too_many_arguments)]
async fn verify_stripe<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    signature_header: &str,
    version: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Option<Duration>,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let window = match timestamp_window {
        Some(window) => window,
        None => {
            let error = ConnectorError::Unsupported(
                "stripe-style signature verification requires a timestamp window".to_string(),
            );
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };

    if let Err(error) = super::stripe::verify_stripe_signature_with(
        body,
        headers,
        secret,
        window,
        now,
        signature_header,
        version,
    ) {
        let timestamp = match &error {
            ConnectorError::TimestampOutOfWindow { timestamp, .. } => Some(*timestamp),
            _ => None,
        };
        return reject(
            event_log,
            provider,
            style,
            &error,
            now,
            timestamp,
            Some(window),
        )
        .await;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
async fn verify_standard_webhooks<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    id_header: &str,
    signature_header: &str,
    timestamp_header: &str,
    version: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Option<Duration>,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let window = match timestamp_window {
        Some(window) => window,
        None => {
            let error = ConnectorError::Unsupported(
                "standard-webhooks verification requires a timestamp window".to_string(),
            );
            return reject(event_log, provider, style, &error, now, None, None).await;
        }
    };

    let message_id = match required_header(headers, id_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(event_log, provider, style, &error, now, None, Some(window)).await
        }
    };
    let signature_header_value = match required_header(headers, signature_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(event_log, provider, style, &error, now, None, Some(window)).await
        }
    };
    let timestamp_raw = match required_header(headers, timestamp_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(event_log, provider, style, &error, now, None, Some(window)).await
        }
    };

    let timestamp = match timestamp_raw.parse::<i64>() {
        Ok(raw) => match OffsetDateTime::from_unix_timestamp(raw) {
            Ok(timestamp) => timestamp,
            Err(error) => {
                let error = ConnectorError::InvalidHeader {
                    name: timestamp_header.to_string(),
                    detail: error.to_string(),
                };
                return reject(event_log, provider, style, &error, now, None, Some(window)).await;
            }
        },
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: timestamp_header.to_string(),
                detail: error.to_string(),
            };
            return reject(event_log, provider, style, &error, now, None, Some(window)).await;
        }
    };
    ensure_timestamp_within_window(event_log, provider, style, timestamp, window, now).await?;

    let signing_key = match decode_standard_webhooks_secret(secret) {
        Ok(secret) => secret,
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(window),
            )
            .await
        }
    };

    let mut signed = message_id.as_bytes().to_vec();
    signed.push(b'.');
    signed.extend_from_slice(timestamp_raw.as_bytes());
    signed.push(b'.');
    signed.extend_from_slice(body);
    let expected = hmac_sha256(&signing_key, &signed);

    let mut any_v1 = false;
    for versioned in signature_header_value.split_ascii_whitespace() {
        let Some((current_version, encoded_signature)) = versioned.split_once(',') else {
            continue;
        };
        if current_version != version {
            continue;
        }
        any_v1 = true;
        let provided = match BASE64_STANDARD.decode(encoded_signature) {
            Ok(signature) => signature,
            Err(error) => {
                let error = ConnectorError::InvalidHeader {
                    name: signature_header.to_string(),
                    detail: error.to_string(),
                };
                return reject(
                    event_log,
                    provider,
                    style,
                    &error,
                    now,
                    Some(timestamp),
                    Some(window),
                )
                .await;
            }
        };
        if secure_eq(&expected, &provided) {
            return Ok(());
        }
    }

    let error = if any_v1 {
        ConnectorError::invalid_signature(
            "no standard-webhooks signature matched the raw request body",
        )
    } else {
        ConnectorError::InvalidHeader {
            name: signature_header.to_string(),
            detail: format!("missing `{version},` signature entry"),
        }
    };
    reject(
        event_log,
        provider,
        style,
        &error,
        now,
        Some(timestamp),
        Some(window),
    )
    .await
}

#[allow(clippy::too_many_arguments)]
async fn verify_canonical_request<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    authorization_header: &str,
    scheme: &str,
    method: &str,
    path: &str,
    body: &[u8],
    headers: &BTreeMap<String, String>,
    secret: &str,
    timestamp_window: Duration,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    let authorization = match required_header(headers, authorization_header) {
        Ok(value) => value,
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                Some(timestamp_window),
            )
            .await
        }
    };

    let params = authorization
        .strip_prefix(scheme)
        .map(str::trim_start)
        .ok_or_else(|| ConnectorError::InvalidHeader {
            name: authorization_header.to_string(),
            detail: format!("expected `{scheme}` authorization scheme"),
        });
    let params = match params {
        Ok(value) if !value.is_empty() => value,
        Ok(_) => {
            let error = ConnectorError::InvalidHeader {
                name: authorization_header.to_string(),
                detail: "missing signature parameters".to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                Some(timestamp_window),
            )
            .await;
        }
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                Some(timestamp_window),
            )
            .await
        }
    };

    let mut timestamp_raw = None;
    let mut signature = None;
    for part in params.split(',') {
        let Some((key, value)) = part.trim().split_once('=') else {
            continue;
        };
        match key.trim() {
            "timestamp" => timestamp_raw = Some(value.trim()),
            "signature" => signature = Some(value.trim()),
            "key-id" => {}
            _ => {}
        }
    }

    let timestamp_raw = match timestamp_raw {
        Some(value) if !value.is_empty() => value,
        _ => {
            let error = ConnectorError::InvalidHeader {
                name: authorization_header.to_string(),
                detail: "missing `timestamp=` parameter".to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                Some(timestamp_window),
            )
            .await;
        }
    };
    let timestamp = match timestamp_raw.parse::<i64>() {
        Ok(raw) => match OffsetDateTime::from_unix_timestamp(raw) {
            Ok(parsed) => parsed,
            Err(error) => {
                let error = ConnectorError::InvalidHeader {
                    name: authorization_header.to_string(),
                    detail: error.to_string(),
                };
                return reject(
                    event_log,
                    provider,
                    style,
                    &error,
                    now,
                    None,
                    Some(timestamp_window),
                )
                .await;
            }
        },
        Err(error) => {
            let error = ConnectorError::InvalidHeader {
                name: authorization_header.to_string(),
                detail: error.to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                None,
                Some(timestamp_window),
            )
            .await;
        }
    };
    ensure_timestamp_within_window(event_log, provider, style, timestamp, timestamp_window, now)
        .await?;

    let signature = match signature {
        Some(value) if !value.is_empty() => value,
        _ => {
            let error = ConnectorError::InvalidHeader {
                name: authorization_header.to_string(),
                detail: "missing `signature=` parameter".to_string(),
            };
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(timestamp_window),
            )
            .await;
        }
    };
    let provided = match decode_base64(signature, authorization_header) {
        Ok(bytes) => bytes,
        Err(error) => {
            return reject(
                event_log,
                provider,
                style,
                &error,
                now,
                Some(timestamp),
                Some(timestamp_window),
            )
            .await
        }
    };

    let signed = canonical_request_message(method, path, timestamp_raw, body);
    let expected = hmac_sha256(secret.as_bytes(), signed.as_bytes());
    if secure_eq(&expected, &provided) {
        Ok(())
    } else {
        let error =
            ConnectorError::invalid_signature("signature did not match the canonical request");
        reject(
            event_log,
            provider,
            style,
            &error,
            now,
            Some(timestamp),
            Some(timestamp_window),
        )
        .await
    }
}

async fn ensure_timestamp_within_window<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    timestamp: OffsetDateTime,
    window: Duration,
    now: OffsetDateTime,
) -> Result<(), ConnectorError> {
    if now - timestamp > window || timestamp - now > window {
        let error = ConnectorError::TimestampOutOfWindow {
            timestamp,
            now,
            window,
        };
        return reject(
            event_log,
            provider,
            style,
            &error,
            now,
            Some(timestamp),
            Some(window),
        )
        .await;
    }
    Ok(())
}

pub(super) fn required_header<'a>(
    headers: &'a BTreeMap<String, String>,
    name: &str,
) -> Result<&'a str, ConnectorError> {
    header_value(headers, name).ok_or_else(|| ConnectorError::MissingHeader(name.to_string()))
}

fn parse_json_i64ish(value: &serde_json::Value) -> Option<i64> {
    value
        .as_i64()
        .or_else(|| value.as_u64().and_then(|raw| i64::try_from(raw).ok()))
        .or_else(|| value.as_str().and_then(|raw| raw.parse::<i64>().ok()))
}

fn offset_from_unix_millis(raw: i64) -> Result<OffsetDateTime, String> {
    let seconds = raw.div_euclid(1_000);
    let millis = raw.rem_euclid(1_000);
    let timestamp =
        OffsetDateTime::from_unix_timestamp(seconds).map_err(|error| error.to_string())?;
    timestamp
        .checked_add(Duration::milliseconds(millis))
        .ok_or_else(|| "timestamp overflow".to_string())
}

fn decode_standard_webhooks_secret(secret: &str) -> Result<Vec<u8>, ConnectorError> {
    let normalized = secret.strip_prefix("whsec_").unwrap_or(secret);
    let mut padded = normalized.to_string();
    let remainder = padded.len() % 4;
    if remainder != 0 {
        padded.push_str(&"=".repeat(4 - remainder));
    }
    BASE64_STANDARD
        .decode(padded)
        .map_err(|error| ConnectorError::InvalidHeader {
            name: "webhook-secret".to_string(),
            detail: error.to_string(),
        })
}

fn decode_base64(value: &str, header_name: &str) -> Result<Vec<u8>, ConnectorError> {
    let mut padded = value.trim().to_string();
    let remainder = padded.len() % 4;
    if remainder != 0 {
        padded.push_str(&"=".repeat(4 - remainder));
    }
    BASE64_STANDARD
        .decode(padded)
        .map_err(|error| ConnectorError::InvalidHeader {
            name: header_name.to_string(),
            detail: error.to_string(),
        })
}

fn canonical_request_message(method: &str, path: &str, timestamp: &str, body: &[u8]) -> String {
    format!(
        "{}\n{}\n{}\n{}",
        method.trim().to_ascii_uppercase(),
        path.trim(),
        timestamp.trim(),
        sha256_hex(body)
    )
}

pub(crate) fn hmac_sha256(secret: &[u8], data: &[u8]) -> Vec<u8> {
    let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("HMAC accepts any key length");
    mac.update(data);
    mac.finalize().into_bytes().to_vec()
}

pub(crate) fn hmac_sha1(secret: &[u8], data: &[u8]) -> Vec<u8> {
    let mut mac = Hmac::<sha1::Sha1>::new_from_slice(secret).expect("HMAC accepts any key length");
    mac.update(data);
    mac.finalize().into_bytes().to_vec()
}

pub(crate) fn secure_eq(expected: &[u8], provided: &[u8]) -> bool {
    expected.ct_eq(provided).into()
}

async fn reject<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    error: &ConnectorError,
    now: OffsetDateTime,
    signed_at: Option<OffsetDateTime>,
    window: Option<Duration>,
) -> Result<(), ConnectorError> {
    audit_rejection(event_log, provider, style, error, now, signed_at, window).await;
    Err(match error {
        ConnectorError::DuplicateProvider(value) => {
            ConnectorError::DuplicateProvider(value.clone())
        }
        ConnectorError::DuplicateDelivery(value) => {
            ConnectorError::DuplicateDelivery(value.clone())
        }
        ConnectorError::UnknownProvider(value) => ConnectorError::UnknownProvider(value.clone()),
        ConnectorError::MissingHeader(value) => ConnectorError::MissingHeader(value.clone()),
        ConnectorError::InvalidHeader { name, detail } => ConnectorError::InvalidHeader {
            name: name.clone(),
            detail: detail.clone(),
        },
        ConnectorError::InvalidSignature(value) => ConnectorError::InvalidSignature(value.clone()),
        ConnectorError::TimestampOutOfWindow {
            timestamp,
            now,
            window,
        } => ConnectorError::TimestampOutOfWindow {
            timestamp: *timestamp,
            now: *now,
            window: *window,
        },
        ConnectorError::Json(value) => ConnectorError::Json(value.clone()),
        ConnectorError::Secret(value) => ConnectorError::Secret(value.clone()),
        ConnectorError::EventLog(value) => ConnectorError::EventLog(value.clone()),
        ConnectorError::HarnRuntime(value) => ConnectorError::HarnRuntime(value.clone()),
        ConnectorError::Client(value) => ConnectorError::Client(value.clone()),
        ConnectorError::Unsupported(value) => ConnectorError::Unsupported(value.clone()),
        ConnectorError::Activation(value) => ConnectorError::Activation(value.clone()),
    })
}

async fn audit_rejection<L: EventLog + ?Sized>(
    event_log: &L,
    provider: &ProviderId,
    style: HmacSignatureStyle<'_>,
    error: &ConnectorError,
    now: OffsetDateTime,
    signed_at: Option<OffsetDateTime>,
    window: Option<Duration>,
) {
    let payload = json!({
        "provider": provider.as_str(),
        "style": style.label(),
        "reason": error.to_string(),
        "observed_at": now.format(&time::format_description::well_known::Rfc3339).ok(),
        "signed_at": signed_at.and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok()),
        "window_seconds": window.map(|value| value.whole_seconds()),
    });
    let topic = Topic::new(SIGNATURE_VERIFY_AUDIT_TOPIC).expect("audit topic is valid");
    if let Err(error) = event_log
        .append(&topic, LogEvent::new("signature_rejected", payload))
        .await
    {
        crate::events::log_warn(
            "connectors.signature_verify.audit",
            &format!("failed to append signature verification audit event: {error}"),
        );
    }
}

#[cfg(test)]
mod tests;