verify.py

import base64
import requests

from Crypto.Signature import PKCS1_v1_5
from Crypto.Hash import SHA
from Crypto.PublicKey import RSA
from Crypto.Util.asn1 import DerSequence


def import_pubkey_from_x509(pem):
    b64der = ''.join(pem.split('\n')[1:][:-2])
    cert = DerSequence()
    cert.decode(base64.b64decode(b64der))
    tbs_certificate = DerSequence()
    tbs_certificate.decode(cert[0])
    subject_public_key_info = tbs_certificate[6]
    return RSA.importKey(subject_public_key_info)


def verify_pycrypto(pub, dig, b64signature):
    verifier = PKCS1_v1_5.new(pub)
    sig = base64.standard_b64decode(b64signature)
    return verifier.verify(dig, sig)


def get_public_key(url):
    r = requests.get(url, timeout=3.0)
    if r.status_code != 200:
        raise RuntimeError("%d %s" % (r.status_code, r.reason))

    return import_pubkey_from_x509(r.content)


def string_to_sign(payload):
    signing_input = "".join(
        ("%s\n%s\n" % (key, value)
         for (key, value) in sorted(payload.items())
         if key not in (
            "Signature",
            "SignatureVersion",
            "SigningCertURL",
            "UnsubscribeURL",
         ))
    )

    if isinstance(signing_input, unicode):
        signing_input = signing_input.encode('utf8')

    return SHA.new(signing_input)


def verify_sns_signature(payload):
    if payload.get('SignatureVersion') != '1':
        return False

    pub = get_public_key(payload['SigningCertURL'])
    if pub is None:
        return False

    dig = string_to_sign(payload)
    return verify_pycrypto(pub, dig, payload['Signature'])