curl -X POST \
'https://developer.api.us.stg.walmart.com/api-proxy/service/display/api/v1/api/v1/snapshot/entity' \ -H "Authorization: Bearer <auth_token>" \ -H "WM_CONSUMER.ID: <consumer_id>" \ -H "WM_SEC.KEY_VERSION: <key_version>" \ -H "WM_SEC.AUTH_SIGNATURE: $SIGNATURE" \ -H "WM_CONSUMER.INTIMESTAMP: $TIMESTAMP" \ -H "Content-Type: application/json" \ -d '{"advertiserId":123456}'

package com.walmart.wpa; import java.io.*; import java.security.KeyRep; import java.security.PrivateKey; import java.security.Signature; import java.util.*; public class SignatureGenerator { public static void main(String args[]) throws IOException { // Consumer ID provided by Walmart String consumerId = "consumerId"; String privateKeyVersion = "1"; // Location to private key file for which public key file was shared with Walmart String privateKey = readFile("./agency_private_key.pem"); long inTimestamp = System.currentTimeMillis(); Map<String, String> map = new HashMap<>(); map.put("WM_CONSUMER.ID", consumerId); map.put("WM_CONSUMER.INTIMESTAMP", Long.toString(inTimestamp)); map.put("WM_SEC.KEY_VERSION", privateKeyVersion); String[] array = canonicalize(map); String auth_signature = null; try { auth_signature = generateSignature(privateKey, array[1]); // Auth_signature and Timestamp needed to call Display API // Valid for 300 seconds System.out.println("Signature: \t" + auth_signature); System.out.println("Intimestamp: \t" + inTimestamp); } catch (Exception e) { e.printStackTrace(); } } private static String readFile(String path) throws IOException { BufferedReader br = new BufferedReader(new FileReader(path)); try { StringBuilder sb = new StringBuilder(); String line = br.readLine(); while (line != null) { if (line.contains("BEGIN")) { line = br.readLine(); continue; } if (line.contains("END")) break; sb.append(line); line = br.readLine(); } return sb.toString(); } finally { br.close(); } } public static String generateSignature(String key, String stringToSign) throws Exception { Signature signatureInstance = Signature.getInstance("SHA256WithRSA"); ServiceKeyRep keyRep = new ServiceKeyRep(KeyRep.Type.PRIVATE, "RSA", "PKCS#8", Base64.getDecoder().decod e(key)); PrivateKey resolvedPrivateKey = (PrivateKey) keyRep.readResolve(); signatureInstance.initSign(resolvedPrivateKey); byte[] bytesToSign = stringToSign.getBytes("UTF-8"); signatureInstance.update(bytesToSign); byte[] signatureBytes = signatureInstance.sign(); String signatureString = Base64.getEncoder().encodeToString(signatureBytes); return signatureString; } protected static String[] canonicalize(Map<String, String> headersToSign) { StringBuffer canonicalizedStrBuffer = new StringBuffer(); StringBuffer parameterNamesBuffer = new StringBuffer(); Set<String> keySet = headersToSign.keySet(); // Create sorted key set to enforce order on the key names SortedSet<String> sortedKeySet = new TreeSet<String>(keySet); for (String key : sortedKeySet) { Object val = headersToSign.get(key); parameterNamesBuffer.append(key.trim()).append(";"); canonicalizedStrBuffer.append(val.toString().trim()).append("\n"); } return new String[] {parameterNamesBuffer.toString(), canonicalizedStrBuffer.toString()}; } static class ServiceKeyRep extends KeyRep { private static final long serialVersionUID = -7213340660431987616L; public ServiceKeyRep(Type type, String algorithm, String format, byte[] encoded) { super(type, algorithm, format, encoded); } protected Object readResolve() throws ObjectStreamException { return super.readResolve(); } } } 

(Note: this example requires the pycryptodome module to be installed – pip install pycryptodome) #!/usr/bin/env python import time import pycryptodome from Crypto.PublicKey import RSA from Crypto.Signature import PKCS1_v1_5 from Crypto.Hash import SHA256 from base64 import b64encode, b64decode key = open(r"{UPDATE_THIS}/agency_private_key.pem", "r").read() rsakey = RSA.importKey(key) signer = PKCS1_v1_5.new(rsakey) digest = SHA256.new() epoch_time = int(time.time()) * 1000 consumerId = '{CONSUMERID}' keyVersion = '1' data = consumerId + '\n' + str(epoch_time) + '\n' + keyVersion + '\n' digest.update(data.encode("utf8")) sign = signer.sign(digest) final = b64encode(sign) # Auth_signature and Timestamp needed to call Display API # Valid for 300 seconds print ('Timestamp:', epoch_time) print ('Signature:' , b64encode(sign).decode())