<commit_before># This file is part of fedmsg.
# Copyright (C) 2012 - 2014 Red Hat, Inc.
#
# fedmsg is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# fedmsg is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with fedmsg; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
# Authors:  <NAME> <<EMAIL>>
#
""" ``fedmsg.crypto.x509`` - X.509 backend for :mod:`fedmsg.crypto`.  """


import os
import requests
import time

import fedmsg.crypto
import fedmsg.encoding

import logging
log = logging.getLogger(__name__)

import cryptography.x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import asymmetric, hashes, serialization

import six

if six.PY3:
    long = int


def sign(message, ssldir=None, certname=None, **config):
    """ Insert two new fields into the message dict and return it.

    Those fields are:

        - 'signature' - the computed RSA message digest of the JSON repr.
        - 'certificate' - the base64 X509 certificate of the sending host.
    """

    if ssldir is None or certname is None:
        error = "You must set the ssldir and certname keyword arguments."
        raise ValueError(error)

    message['crypto'] = 'x509'

    with open("%s/%s.crt" % (ssldir, certname), "rb") as f:
        data = f.read()
        certificate = cryptography.x509.load_pem_x509_certificate(
            data,
            default_backend()
        )
        # TODO: it seems that there's no way in cryptography to
        #  get back the PEM block of loaded certificate
        cert_pem = data[data.decode('utf-8').find('-----BEGIN'):]

    # Opening this file requires elevated privileges in stg/prod.
    with open("%s/%s.key" % (ssldir, certname), "rb") as f:
        rsa_private = serialization.load_pem_private_key(
            data=f.read(),
            password=None,
            backend=default_backend()
        )

    signer = rsa_private.signer(
        asymmetric.padding.PKCS1v15(),
        hashes.SHA1()
    )
    signer.update(fedmsg.encoding.dumps(message).encode('utf-8'))
    signature = signer.finalize()

    # Return a new dict containing the pairs in the original message as well
    # as the new authn fields.
    return dict(message.items() + [
        ('signature', signature.encode('base64')),
        ('certificate', cert_pem.encode('base64')),
    ])


def validate(message, ssldir=None, **config):
    """ Return true or false if the message is signed appropriately.

    Four things must be true:

      1) The X509 cert must be signed by our CA
      2) The cert must not be in our CRL.
      3) We must be able to verify the signature using the RSA public key
         contained in the X509 cert.
      4) The topic of the message and the CN on the cert must appear in the
         :term:`routing_policy` dict.<commit_msg>Get the encoding story straight here.<commit_after># This file is part of fedmsg.
# Copyright (C) 2012 - 2014 Red Hat, Inc.
#
# fedmsg is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# fedmsg is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with fedmsg; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
# Authors:  <NAME> <<EMAIL>>
#
""" ``fedmsg.crypto.x509`` - X.509 backend for :mod:`fedmsg.crypto`.  """


import base64
import os
import requests
import time

import fedmsg.crypto
import fedmsg.encoding

import logging
log = logging.getLogger(__name__)

import cryptography.x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import asymmetric, hashes, serialization

import six

if six.PY3:
    long = int


def sign(message, ssldir=None, certname=None, **config):
    """ Insert two new fields into the message dict and return it.

    Those fields are:

        - 'signature' - the computed RSA message digest of the JSON repr.
        - 'certificate' - the base64 X509 certificate of the sending host.
    """

    if ssldir is None or certname is None:
        error = "You must set the ssldir and certname keyword arguments."
        raise ValueError(error)

    message['crypto'] = 'x509'

    with open("%s/%s.crt" % (ssldir, certname), "rb") as f:
        data = f.read()
        certificate = cryptography.x509.load_pem_x509_certificate(
            data,
            default_backend()
        )
        # TODO: it seems that there's no way in cryptography to
        #  get back the PEM block of loaded certificate
        cert_pem = data[data.decode('utf-8').find('-----BEGIN'):]

    # Opening this file requires elevated privileges in stg/prod.
    with open("%s/%s.key" % (ssldir, certname), "rb") as f:
        rsa_private = serialization.load_pem_private_key(
            data=f.read(),
            password=<PASSWORD>,
            backend=default_backend()
        )

    signer = rsa_private.signer(
        asymmetric.padding.PKCS1v15(),
        hashes.SHA1()
    )
    signer.update(fedmsg.encoding.dumps(message).encode('utf-8'))
    signature = signer.finalize()

    # Return a new dict containing the pairs in the original message as well
    # as the new authn fields.
    return dict(list(message.items()) + [
        ('signature', base64.b64encode(signature).decode('utf-8')),
        ('certificate', base64.b64encode(cert_pem).decode('utf-8')),
    ])


def validate(message, ssldir=None, **config):
    """ Return true or false if the message is signed appropriately.

    Four things must be true:

      1) The X509 cert must be signed by our CA
      2) The cert must not be in our CRL.
      3) We must be able to verify the signature using the RSA public key
         contained in the X509 cert.
      4) The topic of the message and the CN on the cert must appear in the
         :term:`routing_policy` dict.
{'fraction_non_alphanumeric': 0.09202363367799114, 'fraction_numerical': 0.020384047267355983, 'mean_word_length': 3.008880994671403, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 10, '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': '6114969', 'n_tokens_mistral': 2038, 'n_tokens_neox': 1910, 'n_words': 778}