<commit_before>from __future__ import unicode_literals

import re
import json
import datetime
from moto.core import BaseBackend, BaseModel
from moto.ec2 import ec2_backends

from .utils import make_arn_for_certificate

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


DEFAULT_ACCOUNT_ID = 123456789012
GOOGLE_ROOT_CA = b"""-----<KEY>
-----END CERTIFICATE-----"""
# Added google root CA as AWS returns chain you gave it + root CA (provided or not)
# so for now a cheap response is just give any old root CA


def datetime_to_epoch(date):
    # As only Py3 has datetime.timestamp()
    return int((date - datetime.datetime(1970, 1, 1)).total_seconds())


class AWSError(Exception):
    TYPE = None
    STATUS = 400

    def __init__(self, message):
        self.message = message

    def response(self):
        resp = {'__type': self.TYPE, 'message': self.message}
        return json.dumps(resp), dict(status=self.STATUS)


class AWSValidationException(AWSError):
    TYPE = 'ValidationException'


class AWSResourceNotFoundException(AWSError):
    TYPE = 'ResourceNotFoundException'


class CertBundle(BaseModel):
    def __init__(self, certificate, private_key, chain=None, region='us-east-1', arn=None, cert_type='IMPORTED', cert_status='ISSUED'):
        self.created_at = datetime.datetime.now()
        self.cert = certificate
        self._cert = None
        self.common_name = None
        self.key = private_key
        self._key = None
        self.chain = chain
        self.tags = {}
        self._chain = None
        self.type = cert_type  # Should really be an enum
        self.status = cert_status  # Should really be an enum

        # AWS always returns your chain + root CA
        if self.chain is None:
            self.chain = GOOGLE_ROOT_CA
        else:
            self.chain += b'\n' + GOOGLE_ROOT_CA

        # Takes care of PEM checking
        self.validate_pk()
        self.validate_certificate()
        if chain is not None:
            self.validate_chain()

        # TODO check cert is valid, or if self-signed then a chain is provided, otherwise
        # raise AWSValidationException('Provided certificate is not a valid self signed. Please provide either a valid self-signed certificate or certificate chain.')

        # Used for when one wants to overwrite an arn
        if arn is None:
            self.arn = make_arn_for_certificate(DEFAULT_ACCOUNT_ID, region)
        else:
            self.arn = arn

    @classmethod
    def generate_cert(cls, domain_name, sans=None):
        if sans is None:
            sans = set()
        else:
            sans = set(sans)

        sans.add(domain_name)
        sans = [cryptography.x509.DNSName(item) for item in sans]

        key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
        subject = cryptography.x509.Name([
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COUNTRY_NAME, u"US"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME, u"CA"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.LOCALITY_NAME, u"San Francisco"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATION_NAME, u"My Company"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COMMON_NAME, domain_name),
        ])
        issuer = cryptography.x509.Name([  # C = US, O = Amazon, OU = Server CA 1B, CN = Amazon
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COUNTRY_NAME, u"US"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATION_NAME, u"Amazon"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATIONAL_UNIT_NAME, u"Server CA 1B"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COMMON_NAME, u"Amazon"),
        ])
        cert = cryptography.x509.CertificateBuilder().subject_name(
            subject
        ).issuer_name(
            issuer
        ).public_key(
            key.public_key()
        ).serial_number(
            cryptography.x509.random_serial_number()
        ).not_valid_before(
            datetime.datetime.utcnow()
        ).not_valid_after(
            datetime.datetime.utcnow() + datetime.timedelta(days=365)
        ).add_extension(
            cryptography.x509.SubjectAlternativeName(sans),
            critical=False,
        ).sign(key, hashes.SHA512(), default_backend())

        cert_armored = cert.public_bytes(serialization.Encoding.PEM)
        private_key = key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        )

        return cls(cert_armored, private_key, cert_type='AMAZON_ISSUED', cert_status='PENDING_VALIDATION')

    def validate_pk(self):
        try:
            self._key = serialization.load_pem_private_key(self.key, password=None, backend=default_backend())

            if self._key.key_size > 2048:
                AWSValidationException('The private key length is not supported. Only 1024-bit and 2048-bit are allowed.')

        except Exception as err:
            if isinstance(err, AWSValidationException):
                raise
            raise AWSValidationException('The private key is not PEM-encoded or is not valid.')

    def validate_certificate(self):
        try:
            self._cert = cryptography.x509.load_pem_x509_certificate(self.cert, default_backend())

            now = datetime.datetime.now()
            if self._cert.not_valid_after < now:
                raise AWSValidationException('The certificate has expired, is not valid.')

            if self._cert.not_valid_before > now:
                raise AWSValidationException('The certificate is not in effect yet, is not valid.')

            # Extracting some common fields for ease of use
            # Have to search through cert.subject for OIDs
            self.common_name = self._cert.subject.get_attributes_for_oid(cryptography.x509.OID_COMMON_NAME)[0].value

        except Exception as err:
            if isinstance(err, AWSValidationException):
                raise
            raise AWSValidationException('The certificate is not PEM-encoded or is not valid.')
<commit_msg>Fix timezone mismatch in acm certificate created date and now date check<commit_after>from __future__ import unicode_literals

import re
import json
import datetime
from moto.core import BaseBackend, BaseModel
from moto.ec2 import ec2_backends

from .utils import make_arn_for_certificate

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


DEFAULT_ACCOUNT_ID = 123456789012
GOOGLE_ROOT_CA = b"""-----BEGIN CERTIFICATE-----
<KEY>
-----END CERTIFICATE-----"""
# Added google root CA as AWS returns chain you gave it + root CA (provided or not)
# so for now a cheap response is just give any old root CA


def datetime_to_epoch(date):
    # As only Py3 has datetime.timestamp()
    return int((date - datetime.datetime(1970, 1, 1)).total_seconds())


class AWSError(Exception):
    TYPE = None
    STATUS = 400

    def __init__(self, message):
        self.message = message

    def response(self):
        resp = {'__type': self.TYPE, 'message': self.message}
        return json.dumps(resp), dict(status=self.STATUS)


class AWSValidationException(AWSError):
    TYPE = 'ValidationException'


class AWSResourceNotFoundException(AWSError):
    TYPE = 'ResourceNotFoundException'


class CertBundle(BaseModel):
    def __init__(self, certificate, private_key, chain=None, region='us-east-1', arn=None, cert_type='IMPORTED', cert_status='ISSUED'):
        self.created_at = datetime.datetime.now()
        self.cert = certificate
        self._cert = None
        self.common_name = None
        self.key = private_key
        self._key = None
        self.chain = chain
        self.tags = {}
        self._chain = None
        self.type = cert_type  # Should really be an enum
        self.status = cert_status  # Should really be an enum

        # AWS always returns your chain + root CA
        if self.chain is None:
            self.chain = GOOGLE_ROOT_CA
        else:
            self.chain += b'\n' + GOOGLE_ROOT_CA

        # Takes care of PEM checking
        self.validate_pk()
        self.validate_certificate()
        if chain is not None:
            self.validate_chain()

        # TODO check cert is valid, or if self-signed then a chain is provided, otherwise
        # raise AWSValidationException('Provided certificate is not a valid self signed. Please provide either a valid self-signed certificate or certificate chain.')

        # Used for when one wants to overwrite an arn
        if arn is None:
            self.arn = make_arn_for_certificate(DEFAULT_ACCOUNT_ID, region)
        else:
            self.arn = arn

    @classmethod
    def generate_cert(cls, domain_name, sans=None):
        if sans is None:
            sans = set()
        else:
            sans = set(sans)

        sans.add(domain_name)
        sans = [cryptography.x509.DNSName(item) for item in sans]

        key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
        subject = cryptography.x509.Name([
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COUNTRY_NAME, u"US"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME, u"CA"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.LOCALITY_NAME, u"San Francisco"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATION_NAME, u"My Company"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COMMON_NAME, domain_name),
        ])
        issuer = cryptography.x509.Name([  # C = US, O = Amazon, OU = Server CA 1B, CN = Amazon
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COUNTRY_NAME, u"US"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATION_NAME, u"Amazon"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.ORGANIZATIONAL_UNIT_NAME, u"Server CA 1B"),
            cryptography.x509.NameAttribute(cryptography.x509.NameOID.COMMON_NAME, u"Amazon"),
        ])
        cert = cryptography.x509.CertificateBuilder().subject_name(
            subject
        ).issuer_name(
            issuer
        ).public_key(
            key.public_key()
        ).serial_number(
            cryptography.x509.random_serial_number()
        ).not_valid_before(
            datetime.datetime.utcnow()
        ).not_valid_after(
            datetime.datetime.utcnow() + datetime.timedelta(days=365)
        ).add_extension(
            cryptography.x509.SubjectAlternativeName(sans),
            critical=False,
        ).sign(key, hashes.SHA512(), default_backend())

        cert_armored = cert.public_bytes(serialization.Encoding.PEM)
        private_key = key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        )

        return cls(cert_armored, private_key, cert_type='AMAZON_ISSUED', cert_status='PENDING_VALIDATION')

    def validate_pk(self):
        try:
            self._key = serialization.load_pem_private_key(self.key, password=None, backend=default_backend())

            if self._key.key_size > 2048:
                AWSValidationException('The private key length is not supported. Only 1024-bit and 2048-bit are allowed.')

        except Exception as err:
            if isinstance(err, AWSValidationException):
                raise
            raise AWSValidationException('The private key is not PEM-encoded or is not valid.')

    def validate_certificate(self):
        try:
            self._cert = cryptography.x509.load_pem_x509_certificate(self.cert, default_backend())

            now = datetime.datetime.utcnow()
            if self._cert.not_valid_after < now:
                raise AWSValidationException('The certificate has expired, is not valid.')

            if self._cert.not_valid_before > now:
                raise AWSValidationException('The certificate is not in effect yet, is not valid.')

            # Extracting some common fields for ease of use
            # Have to search through cert.subject for OIDs
            self.common_name = self._cert.subject.get_attributes_for_oid(cryptography.x509.OID_COMMON_NAME)[0].value

        except Exception as err:
            if isinstance(err, AWSValidationException):
                raise
            raise AWSValidationException('The certificate is not PEM-encoded or is not valid.')

{'fraction_non_alphanumeric': 0.08777262709932365, 'fraction_numerical': 0.021126225397066645, 'mean_word_length': 2.750356226845255, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '6018276', 'n_tokens_mistral': 4019, 'n_tokens_neox': 3678, 'n_words': 891}