Validate external link requests from Maxsight

Python code example:

The following Python code can be used to validate an incoming signature on a Flask API.

This code creates a new class called URLSignatureAuth and validates a request object coming in from flask after calling the validate_url_signature method.

import base64
import hashlib
import hmac
from urllib.parse import urlparse, parse_qs
from datetime import datetime
import pytz
import re
from functools import wraps

from flask import request
from werkzeug.exceptions import NotFound, BadRequest


class URLSignatureAuth:
    def __init__(self):
        self.MANDATORY_PARAMS = ("signature", "valid_until", "version", "auditee_id",)

    def validate_url_signature(self):
        url = request.url
        parsed_url = urlparse(url)
        query = parsed_url.query
        query_dict = parse_qs(query)

        if any([param not in query_dict for param in self.MANDATORY_PARAMS]):
            raise BadRequest(f"URL missing mandatory parameters. Mandatory parameters required are `{self.MANDATORY_PARAMS}`")

        valid_until = int(query_dict["valid_until"][0])

        if not self._request_time_valid(valid_until):
            raise NotFound("URL valid until time expired.")

        signature = query_dict["signature"][0]
        query_dict.pop("signature")
        query = re.sub(r'&signature=.*', '', request.url)
        secret_key = base64.b64decode(request.arc.secret_key)

        if not self._request_signature_valid(query, signature, secret_key):
            raise NotFound("Invalid URL signature.")

    def _request_signature_valid(self, query:str, signature:str, secret_key:bytes):

        expected_signature = self._generate_signature(query, secret_key)
        request_signature = base64.urlsafe_b64decode(signature)
        # this can fail because of the http/https difference caused by ngrok
        if expected_signature != request_signature:
            # reattempt using https
            expected_signature = self._generate_signature(query.replace('http://', 'https://'), secret_key)

        return request_signature == expected_signature
    
    def _request_time_valid(self, valid_until:int):
        time_mins = datetime.now(pytz.utc)
        max_valid_until = int(time_mins.timestamp())
        return valid_until > max_valid_until

    def _generate_signature(self, query:str, secret_key:bytes):
        signature = hmac.new(secret_key, query.encode(), hashlib.sha256).digest()
        return signature

Javascript code example:

import axios from "axios";
import * as crypto from "crypto";


const passfortAxios = axios.create({
  headers: {
    "Content-Type": "application/json",
  },
});

// Add a request interceptor
passfortAxios.interceptors.request.use(
  (config) => {
    const decoded_secret_key = Buffer.from(
      process.env.INTEGRATION_SECRET_KEY,
      "base64"
    );
    const secret_key_id = process.env.INTEGRATION_SECRET_KEY.substring(0, 8);

    // the date will be set using GMT and the format below
    // as an example this presents as: 'Fri, 06 Oct 2023 12:27:32 GMT'
    const DATE = new Date().toUTCString();

    // ensure your request payload is set to bytes before being signed
    // do not overwrite the request_payload raw though as that is sent in the request
    const payloadStr = JSON.stringify(config.data);
    const payloadBytes = Buffer.from(payloadStr, "utf8");

    // generate the digest from the request payload - note we use the bytes form stored in the payloadBytes variable
    const DIGEST = crypto
      .createHash("sha256")
      .update(payloadBytes)
      .digest("base64");

    // Extract the path from the URL
    const url = new URL(config.url);
    const path = url.pathname;
    // const path = '/integrations/v1/callbacks';

    // NOTE THE NEW LINE CHARACTERS AND SPACES,
    const headers_str = `(request-target): post ${path}\ndate: ${DATE}\ndigest: SHA-256=${DIGEST}`;
    // encode the string into bytes
    const headers_str_as_bytes = Buffer.from(headers_str);

    // this will be how the signature in the Authorization header will be generated
    // create the hmac, note, use the decoded secret key, and a supported sha256 digest
    const new_hmac = crypto
      .createHmac("sha256", decoded_secret_key)
      .update(headers_str_as_bytes)
      .digest();
    // encode the hmac, note for usual requests to your integration,
    // url_safe is NOT used for standard requests
    const signature = Buffer.from(new_hmac).toString("base64");
    const authorization = `Signature keyId="${secret_key_id}",algorithm="hmac-sha256",signature="${signature}",headers="(request-target) date digest"`;

    // set the request headers
    config.headers = {
      ...(config.headers as any),
      Date: DATE,
      Authorization: authorization,
      Digest: `SHA-256=${DIGEST}`,
    };

    return config;
  },
  (error) => {
    // Do something with request error

    return Promise.reject(error);
  }
);

passfortAxios.post(url, data).catch((e) => {
  if (!e) return;
  // console.error(e);
  console.error("Status code", e.response?.status, e.response?.statusText);
  console.error("Failed to send callback", e.response?.data);

  // If you want to retry the callback, you can call this function again here.
  // Delaying the retry is recommended to avoid rate limiting.
});