def _get_signing_public_key()

in emails/sns.py [0:0]

• 31 lines of code
• 8 McCabe index (conditional complexity)

def _get_signing_public_key(cert_url: str) -> rsa.RSAPublicKey:
    """
    Download the signing certificate and return the public key.

    Or, return the cached public key from a previous call.
    """
    cert_url_origin = f"https://sns.{settings.AWS_REGION}.amazonaws.com/"
    if not (cert_url.startswith(cert_url_origin)):
        raise SuspiciousOperation(
            f'SNS SigningCertURL "{cert_url}" did not start with "{cert_url_origin}"'
        )

    key_cache = caches[getattr(settings, "AWS_SNS_KEY_CACHE", "default")]
    cache_key = f"{cert_url}:public_key"
    public_pem = key_cache.get(cache_key)

    set_cache = False
    if public_pem:
        cert_pubkey = serialization.load_pem_public_key(public_pem)
    else:
        set_cache = True
        response = urlopen(cert_url)  # noqa: S310 (check for custom scheme)
        cert_pem = response.read()

        # Extract the first certificate in the file and confirm it's a valid
        # PEM certificate
        certs = x509.load_pem_x509_certificates(cert_pem)

        # A proper certificate file will contain 1 certificate
        if len(certs) != 1:
            raise VerificationFailed(
                f"SigningCertURL {cert_url} has {len(certs)} certificates."
            )
        cert_pubkey = certs[0].public_key()
        public_pem = cert_pubkey.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )

    if not isinstance(cert_pubkey, rsa.RSAPublicKey):
        raise VerificationFailed(f"SigningCertURL {cert_url} is not an RSA key")

    if set_cache:
        key_cache.set(cache_key, public_pem)
    return cert_pubkey