<commit_before>import base64
import binascii
import hashlib
import logging
import re
from datetime import datetime

import pytz
import requests
import ecdsa.util
import fastecdsa.ecdsa
from fastecdsa.encoding.pem import PEMEncoder
from hashlib import sha256
from pyasn1.codec.der.decoder import decode as der_decode
from pyasn1.codec.der.encoder import encode as der_encode
from pyasn1_modules import rfc5280
from requests_hawk import HawkAuth

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.utils import timezone
from django.utils.functional import cached_property


INFO_RECEIVED_SIGNATURES = "normandy.autograph.I001"


logger = logging.getLogger(__name__)


class Autographer(object):
    """
    Interacts with an Autograph service.

    If Autograph signing is not configured using `settings.AUTOGRAPH`,
    raises `ImproperlyConfigured`. If the Autograph server can't be reached
    or returns an HTTP error, an error will be raised by `requests`.
    """

    def __init__(self):
        self.check_config()

    @cached_property
    def session(self):
        session = requests.Session()
        session.auth = HawkAuth(
            id=str(settings.AUTOGRAPH_HAWK_ID), key=str(settings.AUTOGRAPH_HAWK_SECRET_KEY)
        )
        return session

    def check_config(self):
        required_keys = ["URL", "HAWK_ID", "HAWK_SECRET_KEY"]
        for key in required_keys:
            if getattr(settings, "AUTOGRAPH_" + key) is None:
                msg = "set settings.AUTOGRAPH_{} to use action signatures".format(key)
                raise ImproperlyConfigured(msg)

    def sign_data(self, content_list):
        """
        Fetches Signatures objects from Autograph for each item in `content_list`.

        The items in `content_list` must be bytes objects.
        """
        ts = timezone.now()
        url = "{}sign/data".format(settings.AUTOGRAPH_URL)
        signing_request = []
        for item in content_list:
            # base64 works in bytes. requests work in UTF-8.
            # Convert to bytes, and then back.
            encoded_implementation = base64.b64encode(item).decode("utf8")
            signing_request.append({"input": encoded_implementation})

        res = self.session.post(url, json=signing_request)
        res.raise_for_status()
        signing_responses = res.json()

        logger.info(
            f"Got {len(signing_responses)} signatures from Autograph",
            extra={"code": INFO_RECEIVED_SIGNATURES},
        )

        signatures = []
        for res in signing_responses:
            logger.info(f"Autograph response: {res['ref']}")

            assert res["signature"], "Response from Autograph did not contain signature"
            assert res["x5u"], "Response from Autograph did not contain x5u"

            signatures.append({"timestamp": ts, "signature": res["signature"], "x5u": res["x5u"]})
        return signatures


BASE64_WRONG_LENGTH_RE = re.compile(
    r"Invalid base64-encoded string: number of data characters \(\d+\) cannot "
    r"be [123] more than a multiple of 4"
)


def verify_signature_x5u(data, signature, x5u):
    """
    Verify a signature, given the x5u of the public key.

    If the signature is valid, returns True. If the signature is invalid, raise
    an exception explaining why.
    """
    cert = verify_x5u(x5u)
    encoded = der_encode(cert)
    cert_b64 = base64.b64encode(encoded).decode()
    return verify_signature_pubkey(data, signature, cert_b64)


def verify_signature_pubkey(data, signature, pubkey):
    """
    Verify a signature.

    If the signature is valid, returns True. If the signature is invalid, raise
    an exception explaining why.
    """
    # Data must be encoded as bytes
    if isinstance(data, str):
        data = data.encode()

    # Content signature implicitly adds a prefix to signed data
    data = b"Content-Signature:\x00" + data

    # fastecdsa expects ASCII armored keys, but ours is unarmored. Add the
    # armor before passing the key to the library.
    EC_PUBLIC_HEADER = "-----BEGIN PUBLIC KEY-----"
    EC_PUBLIC_FOOTER = "-----END PUBLIC KEY-----"
    verifying_pubkey = PEMEncoder.decode_public_key(
        "\n".join([EC_PUBLIC_HEADER, pubkey, EC_PUBLIC_FOOTER])
    )

    try:
        signature = base64.urlsafe_b64decode(signature)
        signature = ecdsa.util.sigdecode_string(signature, order=ecdsa.curves.NIST384p.order)
    except binascii.Error as e:
        if BASE64_WRONG_LENGTH_RE.match(e.args[0]):
            raise WrongSignatureSize("Base64 encoded signature was not a multiple of 4")
        else:
            raise
    except AssertionError as e:
        # The signature decoder has a clause like
        #     assert len(signature) == 2*l, (len(signature), 2*l)
        # If the AssertionError is consistent with that signature, translate it
        # to a nicer error. Otherwise re-raise.
        if (
            len(e.args) == 1
            and isinstance(e.args[0], tuple)
            and len(e.args[0]) == 2
            and isinstance(e.args[0][0], int)
            and isinstance(e.args[0][1], int)
        ):
            raise WrongSignatureSize()<commit_msg>Update to new ecdsa exception type<commit_after>import base64
import binascii
import hashlib
import logging
import re
from datetime import datetime

import pytz
import requests
import ecdsa.util
import fastecdsa.ecdsa
from fastecdsa.encoding.pem import PEMEncoder
from hashlib import sha256
from pyasn1.codec.der.decoder import decode as der_decode
from pyasn1.codec.der.encoder import encode as der_encode
from pyasn1_modules import rfc5280
from requests_hawk import HawkAuth

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.utils import timezone
from django.utils.functional import cached_property


INFO_RECEIVED_SIGNATURES = "normandy.autograph.I001"


logger = logging.getLogger(__name__)


class Autographer(object):
    """
    Interacts with an Autograph service.

    If Autograph signing is not configured using `settings.AUTOGRAPH`,
    raises `ImproperlyConfigured`. If the Autograph server can't be reached
    or returns an HTTP error, an error will be raised by `requests`.
    """

    def __init__(self):
        self.check_config()

    @cached_property
    def session(self):
        session = requests.Session()
        session.auth = HawkAuth(
            id=str(settings.AUTOGRAPH_HAWK_ID), key=str(settings.AUTOGRAPH_HAWK_SECRET_KEY)
        )
        return session

    def check_config(self):
        required_keys = ["URL", "HAWK_ID", "HAWK_SECRET_KEY"]
        for key in required_keys:
            if getattr(settings, "AUTOGRAPH_" + key) is None:
                msg = "set settings.AUTOGRAPH_{} to use action signatures".format(key)
                raise ImproperlyConfigured(msg)

    def sign_data(self, content_list):
        """
        Fetches Signatures objects from Autograph for each item in `content_list`.

        The items in `content_list` must be bytes objects.
        """
        ts = timezone.now()
        url = "{}sign/data".format(settings.AUTOGRAPH_URL)
        signing_request = []
        for item in content_list:
            # base64 works in bytes. requests work in UTF-8.
            # Convert to bytes, and then back.
            encoded_implementation = base64.b64encode(item).decode("utf8")
            signing_request.append({"input": encoded_implementation})

        res = self.session.post(url, json=signing_request)
        res.raise_for_status()
        signing_responses = res.json()

        logger.info(
            f"Got {len(signing_responses)} signatures from Autograph",
            extra={"code": INFO_RECEIVED_SIGNATURES},
        )

        signatures = []
        for res in signing_responses:
            logger.info(f"Autograph response: {res['ref']}")

            assert res["signature"], "Response from Autograph did not contain signature"
            assert res["x5u"], "Response from Autograph did not contain x5u"

            signatures.append({"timestamp": ts, "signature": res["signature"], "x5u": res["x5u"]})
        return signatures


BASE64_WRONG_LENGTH_RE = re.compile(
    r"Invalid base64-encoded string: number of data characters \(\d+\) cannot "
    r"be [123] more than a multiple of 4"
)


ECDSA_WRONG_LENGTH_RE = re.compile(
    r"Invalid length of signature, expected \d+ bytes long, provided string is \d+ bytes long"
)


def verify_signature_x5u(data, signature, x5u):
    """
    Verify a signature, given the x5u of the public key.

    If the signature is valid, returns True. If the signature is invalid, raise
    an exception explaining why.
    """
    cert = verify_x5u(x5u)
    encoded = der_encode(cert)
    cert_b64 = base64.b64encode(encoded).decode()
    return verify_signature_pubkey(data, signature, cert_b64)


def verify_signature_pubkey(data, signature, pubkey):
    """
    Verify a signature.

    If the signature is valid, returns True. If the signature is invalid, raise
    an exception explaining why.
    """
    # Data must be encoded as bytes
    if isinstance(data, str):
        data = data.encode()

    # Content signature implicitly adds a prefix to signed data
    data = b"Content-Signature:\x00" + data

    # fastecdsa expects ASCII armored keys, but ours is unarmored. Add the
    # armor before passing the key to the library.
    EC_PUBLIC_HEADER = "-----BEGIN PUBLIC KEY-----"
    EC_PUBLIC_FOOTER = "-----END PUBLIC KEY-----"
    verifying_pubkey = PEMEncoder.decode_public_key(
        "\n".join([EC_PUBLIC_HEADER, pubkey, EC_PUBLIC_FOOTER])
    )

    try:
        signature = base64.urlsafe_b64decode(signature)
        signature = ecdsa.util.sigdecode_string(signature, order=ecdsa.curves.NIST384p.order)
    except binascii.Error as e:
        if BASE64_WRONG_LENGTH_RE.match(e.args[0]):
            raise WrongSignatureSize("Base64 encoded signature was not a multiple of 4")
        else:
            raise
    except ecdsa.util.MalformedSignature as e:
        if ECDSA_WRONG_LENGTH_RE.match(e.args[0]):
            raise WrongSignatureSize()
{'fraction_non_alphanumeric': 0.08945812807881773, 'fraction_numerical': 0.013497536945812807, 'mean_word_length': 2.905732974220854, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1047428', 'n_tokens_mistral': 3123, 'n_tokens_neox': 2963, 'n_words': 887}