Ebay Marketplace Account Deletion/Closure Notifications serverless (GCP) implementation written in python

main.py

import os
import json
import base64
import logging
import hashlib
import requests
from flask import request, Response
from OpenSSL import crypto

logger = logging.getLogger(__name__)

#Ebay config values
ENDPOINT = (
    "ENDPOINT_URL"
)
VERIFICATION_TOKEN = os.getenv("VERIFICATION_TOKEN")
# ^ NOTE: You can make this up
X_EBAY_SIGNATURE = "X-Ebay-Signature"
EBAY_BASE64_AUTHORIZATION_TOKEN = os.getenv("EBAY_BASE64_AUTHORIZATION_TOKEN")
# ^ NOTE: Here's how you can get your EBAY_BASE64_AUTHORIZATION_TOKEN:
# import base64
# base64.b64encode(b'{CLIENT_ID}:{CLIENT_SECRET}')


def get(self):
    if request.method == "GET":
        code = request.args.get("challenge_code")

        challenge_response = hashlib.sha256(
            code.encode("utf-8")
            + VERIFICATION_TOKEN.encode("utf-8")
            + ENDPOINT.encode("utf-8")
        )
        response_parameters = {"challengeResponse": challenge_response.hexdigest()}

        return response_parameters


def post(self):
    if request.method == "POST":
        x_ebay_signature = request.headers.get(X_EBAY_SIGNATURE)
        x_ebay_signature_decoded = json.loads(base64.b64decode(x_ebay_signature))
        kid = x_ebay_signature_decoded["kid"]
        signature = x_ebay_signature_decoded["signature"]

        public_key = None
        try:
            ebay_verification_url = (
                f"https://api.ebay.com/commerce/notification/v1/public_key/{kid}"
            )
            oauth_access_token = get_oauth_token()
            headers = {"Authorization": f"Bearer {oauth_access_token}"}
            public_key_request = requests.get(
                url=ebay_verification_url, headers=headers, data={}
            )
            if public_key_request.status_code == 200:
                public_key_response = public_key_request.json()
                public_key = public_key_response["key"]
                verify_key(public_key, signature, request)
                return Response(status=200)
        except Exception as e:
            message_title = "Ebay Marketplace Account Deletion: Error calling getPublicKey Notfication API."
            logger.error(f"{message_title} Error: {e}")
            return Response(status=500)


def verify_key(public_key, signature, request):
    pkey = crypto.load_publickey(
        crypto.FILETYPE_PEM, get_public_key_into_proper_format(public_key)
    )
    certification = crypto.X509()
    certification.set_pubkey(pkey)
    notification_payload = request.data
    signature_decoded = base64.b64decode(signature)
    try:
        crypto.verify(certification, signature_decoded, notification_payload, "sha1")
        return
    except crypto.Error as e:
        message_title = (
            f"Ebay Marketplace Account Deletion: Signature Invalid. "
            f"The signature is invalid or there is a problem verifying the signature. "
        )
        logger.warning(f"{message_title} Error: {e}")
        return Response(status=412)
    except Exception as e:
        message_title = f"Ebay Marketplace Account Deletion: Error performing cryptographic validation."
        logger.error(f"{message_title} Error: {e}")
        return Response(status=412)


def get_oauth_token():
    """
    Returns the OAuth Token from eBay which can be used for making other API requests such as getPublicKey
    """
    url = "https://api.ebay.com/identity/v1/oauth2/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": f"Basic {EBAY_BASE64_AUTHORIZATION_TOKEN}",
    }
    payload = {
        "grant_type": "client_credentials",
        "scope": "https://api.ebay.com/oauth/api_scope",
    }
    r = requests.post(url=url, headers=headers, data=payload)
    data = r.json()
    return data["access_token"]


def get_public_key_into_proper_format(public_key):
    """
    Public key needs to have \n in places to be properly assessed by crypto library.
    """
    return public_key[:26] + "\n" + public_key[26:-24] + "\n" + public_key[-24:]

requirements.txt

pyOpenSSL
requests