diff --git a/.gitignore b/.gitignore index 7505ec58..a7b00969 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +/.cache .coverage .tox .DS_Store diff --git a/.travis.yml b/.travis.yml index d8aa6116..4caacda2 100644 --- a/.travis.yml +++ b/.travis.yml @@ -9,10 +9,8 @@ matrix: include: - python: "2.7" env: TOXENV=py27 - - python: "3.3" - env: TOXENV=py33 - - python: "3.4" - env: TOXENV=py34 + - python: "3.5" + env: TOXENV=py35 cache: directories: diff --git a/bower.json b/bower.json index b4dba71d..45e54528 100644 --- a/bower.json +++ b/bower.json @@ -29,7 +29,7 @@ "angular-ui-switch": "~0.1.0", "angular-chart.js": "~0.7.1", "satellizer": "~0.9.4", - "angularjs-toaster": "~0.4.14", + "angularjs-toaster": "~1.0.0", "ngletteravatar": "~3.0.1", "angular-ui-router": "~0.2.15", "angular-clipboard": "~1.1.1", diff --git a/lemur/authorities/models.py b/lemur/authorities/models.py index 0dc5420b..54b13348 100644 --- a/lemur/authorities/models.py +++ b/lemur/authorities/models.py @@ -42,13 +42,15 @@ class Authority(db.Model): self.body = body self.chain = chain self.owner = owner + self.description = description self.plugin_name = plugin_name - cert = x509.load_pem_x509_certificate(str(body), default_backend()) + cert = x509.load_pem_x509_certificate(bytes(body), default_backend()) self.cn = get_cn(cert) self.not_before = get_not_before(cert) self.not_after = get_not_after(cert) - self.roles = roles - self.description = description + + if roles: + self.roles = roles def as_dict(self): return {c.name: getattr(self, c.name) for c in self.__table__.columns} diff --git a/lemur/authorities/schemas.py b/lemur/authorities/schemas.py new file mode 100644 index 00000000..e69de29b diff --git a/lemur/certificates/models.py b/lemur/certificates/models.py index d629d327..ce6b4d57 100644 --- a/lemur/certificates/models.py +++ b/lemur/certificates/models.py @@ -228,7 +228,7 @@ class Certificate(db.Model): cn = Column(String(128)) description = Column(String(1024)) active = Column(Boolean, default=True) - san = Column(String(1024)) + san = Column(String(1024)) # TODO this should be migrated to boolean not_before = Column(DateTime) not_after = Column(DateTime) date_created = Column(DateTime, PassiveDefault(func.now()), nullable=False) @@ -250,7 +250,7 @@ class Certificate(db.Model): # We encrypt the private_key on creation self.private_key = private_key self.chain = chain - cert = x509.load_pem_x509_certificate(str(self.body), default_backend()) + cert = x509.load_pem_x509_certificate(bytes(self.body), default_backend()) self.signing_algorithm = get_signing_algorithm(cert) self.bits = get_bitstrength(cert) self.issuer = get_issuer(cert) diff --git a/lemur/certificates/schemas.py b/lemur/certificates/schemas.py new file mode 100644 index 00000000..b83612d2 --- /dev/null +++ b/lemur/certificates/schemas.py @@ -0,0 +1,302 @@ +""" +.. module: lemur.certificates.schemas + :platform: unix + :copyright: (c) 2015 by Netflix Inc., see AUTHORS for more + :license: Apache, see LICENSE for more details. +.. moduleauthor:: Kevin Glisson +""" +from flask import current_app + +import arrow + +from marshmallow import fields, validates_schema, pre_load, post_dump +from marshmallow.exceptions import ValidationError + +from cryptography import x509 +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization + +from lemur.auth.permissions import SensitiveDomainPermission +from lemur.schemas import AssociatedAuthoritySchema, AssociatedDestinationSchema, AssociatedCertificateSchema, \ + AssociatedNotificationSchema, PluginSchema +from lemur.common.schema import LemurInputSchema, LemurOutputSchema, LemurSchema + +from lemur.domains import service as domain_service + + +def validate_public_certificate(body): + """ + Determines if specified string is valid public certificate. + + :param body: + :return: + """ + try: + x509.load_pem_x509_certificate(bytes(body), default_backend()) + except Exception: + raise ValidationError('Public certificate presented is not valid.') + + +def validate_private_key(key): + """ + User to validate that a given string is a RSA private key + + :param key: + :return: :raise ValueError: + """ + try: + serialization.load_pem_private_key(bytes(key), None, backend=default_backend()) + except Exception: + raise ValidationError('Private key presented is not valid.') + + +def validate_domain(domain): + """ + Determines if domain has been marked as sensitive. + :param domain: + :return: + """ + domains = domain_service.get_by_name(domain) + for domain in domains: + # we only care about non-admins + if not SensitiveDomainPermission().can(): + if domain.sensitive: + raise ValidationError( + 'Domain {0} has been marked as sensitive, contact and administrator \ + to issue the certificate.'.format(domain)) + + +def validate_oid_type(oid_type): + """ + Determines if the specified oid type is valid. + :param oid_type: + :return: + """ + valid_types = ['b64asn1', 'string', 'ia5string'] + if oid_type.lower() not in [o_type.lower() for o_type in valid_types]: + raise ValidationError('Invalid Oid Type: {0} choose from {1}'.format(oid_type, ",".join(valid_types))) + + +def validate_sub_alt_type(alt_type): + """ + Determines if the specified subject alternate type is valid. + :param alt_type: + :return: + """ + valid_types = ['DNSName', 'IPAddress', 'uniFormResourceIdentifier', 'directoryName', 'rfc822Name', 'registrationID', + 'otherName', 'x400Address', 'EDIPartyName'] + if alt_type.lower() not in [a_type.lower() for a_type in valid_types]: + raise ValidationError('Invalid SubAltName Type: {0} choose from {1}'.format(type, ",".join(valid_types))) + + +def validate_csr(data): + """ + Determines if the CSR is valid. + :param data: + :return: + """ + try: + x509.load_pem_x509_csr(bytes(data), default_backend()) + except Exception: + raise ValidationError('CSR presented is not valid.') + + +class BaseExtensionSchema(LemurSchema): + @pre_load(pass_many=True) + def preprocess(self, data, many): + return self.under(data, many=many) + + @post_dump(pass_many=True) + def post_process(self, data, many): + if data: + data = self.camel(data, many=many) + return data + + +class BasicConstraintsSchema(BaseExtensionSchema): + pass + + +class AuthorityIdentifierSchema(BaseExtensionSchema): + use_authority_cert = fields.Boolean() + + +class AuthorityKeyIdentifierSchema(BaseExtensionSchema): + use_key_identifier = fields.Boolean() + + +class CertificateInfoAccessSchema(BaseExtensionSchema): + include_aia = fields.Boolean() + + @post_dump + def handle_keys(self, data): + return {'includeAIA': data['include_aia']} + + +class KeyUsageSchema(BaseExtensionSchema): + use_crl_sign = fields.Boolean() + use_data_encipherment = fields.Boolean() + use_decipher_only = fields.Boolean() + use_encipher_only = fields.Boolean() + use_key_encipherment = fields.Boolean() + use_digital_signature = fields.Boolean() + use_non_repudiation = fields.Boolean() + + +class ExtendedKeyUsageSchema(BaseExtensionSchema): + use_server_authentication = fields.Boolean() + use_client_authentication = fields.Boolean() + use_eap_over_lan = fields.Boolean() + use_eap_over_ppp = fields.Boolean() + use_ocsp_signing = fields.Boolean() + use_smart_card_authentication = fields.Boolean() + use_timestamping = fields.Boolean() + + +class SubjectKeyIdentifierSchema(BaseExtensionSchema): + include_ski = fields.Boolean() + + @post_dump + def handle_keys(self, data): + return {'includeSKI': data['include_ski']} + + +class SubAltNameSchema(BaseExtensionSchema): + name_type = fields.String(validate=validate_sub_alt_type) + value = fields.String() + + @validates_schema + def check_sensitive(self, data): + if data['name_type'] == 'DNSName': + validate_domain(data['value']) + + +class SubAltNamesSchema(BaseExtensionSchema): + names = fields.Nested(SubAltNameSchema, many=True) + + +class CustomOIDSchema(BaseExtensionSchema): + oid = fields.String() + oid_type = fields.String(validate=validate_oid_type) + value = fields.String() + + +class ExtensionSchema(BaseExtensionSchema): + basic_constraints = fields.Nested(BasicConstraintsSchema) + key_usage = fields.Nested(KeyUsageSchema) + extended_key_usage = fields.Nested(ExtendedKeyUsageSchema) + subject_key_identifier = fields.Nested(SubjectKeyIdentifierSchema) + sub_alt_names = fields.Nested(SubAltNamesSchema) + authority_identifier = fields.Nested(AuthorityIdentifierSchema) + authority_key_identifier = fields.Nested(AuthorityKeyIdentifierSchema) + certificate_info_access = fields.Nested(CertificateInfoAccessSchema) + custom = fields.List(fields.Nested(CustomOIDSchema)) + + +class CertificateInputSchema(LemurInputSchema): + name = fields.String() + owner = fields.Email(required=True) + description = fields.String() + common_name = fields.String(required=True, validate=validate_domain) + authority = fields.Nested(AssociatedAuthoritySchema, required=True) + + validity_start = fields.DateTime() + validity_end = fields.DateTime() + validity_years = fields.Integer() + + destinations = fields.Nested(AssociatedDestinationSchema, missing=[], many=True) + notifications = fields.Nested(AssociatedNotificationSchema, missing=[], many=True) + replacements = fields.Nested(AssociatedCertificateSchema, missing=[], many=True) + + csr = fields.String(validate=validate_csr) + + # certificate body fields + organizational_unit = fields.String(missing=lambda: current_app.config.get('LEMUR_DEFAULT_ORGANIZATIONAL_UNIT')) + organization = fields.String(missing=lambda: current_app.config.get('LEMUR_DEFAULT_ORGANIZATION')) + location = fields.String(missing=lambda: current_app.config.get('LEMUR_DEFAULT_LOCATION')) + country = fields.String(missing=lambda: current_app.config.get('LEMUR_DEFAULT_COUNTRY')) + state = fields.String(missing=lambda: current_app.config.get('LEMUR_DEFAULT_STATE')) + + extensions = fields.Nested(ExtensionSchema) + + @validates_schema + def validate_dates(self, data): + if not data.get('validity_start') and data.get('validity_end'): + raise ValidationError('If validity start is specified so must validity end.') + + if not data.get('validity_end') and data.get('validity_start'): + raise ValidationError('If validity end is specified so must validity start.') + + if data.get('validity_end') and data.get('validity_years'): + raise ValidationError('Cannot specify both validity end and validity years.') + + if data.get('validity_start') and data.get('validity_end'): + if not data['validity_start'] < data['validity_end']: + raise ValidationError('Validity start must be before validity end.') + + if data.get('validity_start').replace(tzinfo=None) < data['authority'].not_before: + raise ValidationError('Validity start must not be before {0}'.format(data['authority'].not_before)) + + if data.get('validity_end').replace(tzinfo=None) > data['authority'].not_after: + raise ValidationError('Validity end must not be after {0}'.format(data['authority'].not_after)) + + if data.get('validity_years'): + now = arrow.utcnow() + end = now.replace(years=+data['validity_years']) + + if now.naive < data['authority'].not_before: + raise ValidationError('Validity start must not be before {0}'.format(data['authority'].not_before)) + + if end.naive > data['authority'].not_after: + raise ValidationError('Validity end must not be after {0}'.format(data['authority'].not_after)) + + +class CertificateOutputSchema(LemurOutputSchema): + id = fields.Integer() + active = fields.Boolean() + bits = fields.Integer() + body = fields.String() + chain = fields.String() + deleted = fields.Boolean(default=False) + description = fields.String() + issuer = fields.String() + name = fields.String() + not_after = fields.DateTime() + not_before = fields.DateTime() + owner = fields.Email() + san = fields.Boolean() + serial = fields.String() + signing_algorithm = fields.String() + status = fields.Boolean() + + +class CertificateUploadInputSchema(LemurInputSchema): + name = fields.String() + owner = fields.Email(required=True) + description = fields.String() + active = fields.Boolean(missing=True) + + private_key = fields.String(validate=validate_private_key) + public_cert = fields.String(required=True, validate=validate_public_certificate) + chain = fields.String(validate=validate_public_certificate) + + destinations = fields.Nested(AssociatedDestinationSchema, missing=[], many=True) + notifications = fields.Nested(AssociatedNotificationSchema, missing=[], many=True) + replacements = fields.Nested(AssociatedCertificateSchema, missing=[], many=True) + + @validates_schema + def keys(self, data): + if data.get('destinations'): + if not data.get('private_key'): + raise ValidationError('Destinations require private key.') + + +class CertificateExportInputSchema(LemurInputSchema): + export = fields.Nested(PluginSchema) + + +certificate_input_schema = CertificateInputSchema() +certificate_output_schema = CertificateOutputSchema() +certificates_output_schema = CertificateOutputSchema(many=True) +certificate_upload_input_schema = CertificateUploadInputSchema() +certificate_export_input_schema = CertificateExportInputSchema() diff --git a/lemur/certificates/service.py b/lemur/certificates/service.py index 6d140d4a..08d23054 100644 --- a/lemur/certificates/service.py +++ b/lemur/certificates/service.py @@ -88,7 +88,6 @@ def export(cert, export_plugin): :return: """ plugin = plugins.get(export_plugin['slug']) - return plugin.export(cert.body, cert.chain, cert.private_key, export_plugin['pluginOptions']) @@ -222,8 +221,8 @@ def upload(**kwargs): g.user.certificates.append(cert) - database.update_list(cert, 'destinations', Destination, kwargs.get('destinations')) - database.update_list(cert, 'notifications', Notification, kwargs.get('notifications')) + database.update_list(cert, 'destinations', Destination, kwargs['destinations']) + database.update_list(cert, 'notifications', Notification, kwargs['notifications']) database.update_list(cert, 'replaces', Certificate, kwargs['replacements']) # create default notifications for this certificate if none are provided @@ -256,9 +255,9 @@ def create(**kwargs): # do this after the certificate has already been created because if it fails to upload to the third party # we do not want to lose the certificate information. - database.update_list(cert, 'destinations', Destination, kwargs.get('destinations')) + database.update_list(cert, 'destinations', Destination, kwargs['destinations']) database.update_list(cert, 'replaces', Certificate, kwargs['replacements']) - database.update_list(cert, 'notifications', Notification, kwargs.get('notifications')) + database.update_list(cert, 'notifications', Notification, kwargs['notifications']) # create default notifications for this certificate if none are provided notifications = cert.notifications @@ -364,9 +363,9 @@ def create_csr(csr_config): # TODO When we figure out a better way to validate these options they should be parsed as str builder = x509.CertificateSigningRequestBuilder() builder = builder.subject_name(x509.Name([ - x509.NameAttribute(x509.OID_COMMON_NAME, csr_config['commonName']), + x509.NameAttribute(x509.OID_COMMON_NAME, csr_config['common_name']), x509.NameAttribute(x509.OID_ORGANIZATION_NAME, csr_config['organization']), - x509.NameAttribute(x509.OID_ORGANIZATIONAL_UNIT_NAME, csr_config['organizationalUnit']), + x509.NameAttribute(x509.OID_ORGANIZATIONAL_UNIT_NAME, csr_config['organizational_unit']), x509.NameAttribute(x509.OID_COUNTRY_NAME, csr_config['country']), x509.NameAttribute(x509.OID_STATE_OR_PROVINCE_NAME, csr_config['state']), x509.NameAttribute(x509.OID_LOCALITY_NAME, csr_config['location']), @@ -378,11 +377,11 @@ def create_csr(csr_config): if csr_config.get('extensions'): for k, v in csr_config.get('extensions', {}).items(): - if k == 'subAltNames': + if k == 'sub_alt_names': # map types to their x509 objects general_names = [] for name in v['names']: - if name['nameType'] == 'DNSName': + if name['name_type'] == 'DNSName': general_names.append(x509.DNSName(name['value'])) builder = builder.add_extension( diff --git a/lemur/certificates/views.py b/lemur/certificates/views.py index 432e6dbd..dd28f21e 100644 --- a/lemur/certificates/views.py +++ b/lemur/certificates/views.py @@ -9,129 +9,24 @@ import base64 from builtins import str from flask import Blueprint, make_response, jsonify -from flask.ext.restful import reqparse, Api, fields +from flask.ext.restful import reqparse, Api -from cryptography import x509 -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import serialization - -from lemur.plugins import plugins +from lemur.common.schema import validate_schema +from lemur.common.utils import paginated_parser from lemur.auth.service import AuthenticatedResource -from lemur.auth.permissions import ViewKeyPermission -from lemur.auth.permissions import AuthorityPermission -from lemur.auth.permissions import UpdateCertificatePermission -from lemur.auth.permissions import SensitiveDomainPermission +from lemur.auth.permissions import ViewKeyPermission, AuthorityPermission, UpdateCertificatePermission from lemur.certificates import service -from lemur.authorities.models import Authority +from lemur.certificates.schemas import certificate_input_schema, certificate_output_schema, \ + certificate_upload_input_schema, certificates_output_schema, certificate_export_input_schema + from lemur.roles import service as role_service -from lemur.domains import service as domain_service -from lemur.common.utils import marshal_items, paginated_parser -from lemur.notifications.views import notification_list + mod = Blueprint('certificates', __name__) api = Api(mod) -FIELDS = { - 'name': fields.String, - 'id': fields.Integer, - 'bits': fields.Integer, - 'deleted': fields.String, - 'issuer': fields.String, - 'serial': fields.String, - 'owner': fields.String, - 'chain': fields.String, - 'san': fields.String, - 'active': fields.Boolean, - 'description': fields.String, - 'notBefore': fields.DateTime(dt_format='iso8601', attribute='not_before'), - 'notAfter': fields.DateTime(dt_format='iso8601', attribute='not_after'), - 'cn': fields.String, - 'signingAlgorithm': fields.String(attribute='signing_algorithm'), - 'status': fields.String, - 'body': fields.String -} - - -def valid_authority(authority_options): - """ - Defends against invalid authorities - - :param authority_options: - :return: :raise ValueError: - """ - name = authority_options['name'] - authority = Authority.query.filter(Authority.name == name).one() - - if not authority: - raise ValueError("Unable to find authority specified") - - if not authority.active: - raise ValueError("Selected authority [{0}] is not currently active".format(name)) - - return authority - - -def get_domains_from_options(options): - """ - Retrive all domains from certificate options - :param options: - :return: - """ - domains = [options['commonName']] - if options.get('extensions'): - if options['extensions'].get('subAltNames'): - for k, v in options['extensions']['subAltNames']['names']: - if k == 'DNSName': - domains.append(v) - return domains - - -def check_sensitive_domains(domains): - """ - Determines if any certificates in the given certificate - are marked as sensitive - :param domains: - :return: - """ - for domain in domains: - domain_objs = domain_service.get_by_name(domain) - for d in domain_objs: - if d.sensitive: - raise ValueError("The domain {0} has been marked as sensitive. Contact an administrator to " - "issue this certificate".format(d.name)) - - -def pem_str(value, name): - """ - Used to validate that the given string is a PEM formatted string - - :param value: - :param name: - :return: :raise ValueError: - """ - try: - x509.load_pem_x509_certificate(bytes(value), default_backend()) - except Exception: - raise ValueError("The parameter '{0}' needs to be a valid PEM string".format(name)) - return value - - -def private_key_str(value, name): - """ - User to validate that a given string is a RSA private key - - :param value: - :param name: - :return: :raise ValueError: - """ - try: - serialization.load_pem_private_key(bytes(value), None, backend=default_backend()) - except Exception: - raise ValueError("The parameter '{0}' needs to be a valid RSA private key".format(name)) - return value - class CertificatesList(AuthenticatedResource): """ Defines the 'certificates' endpoint """ @@ -140,7 +35,7 @@ class CertificatesList(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(CertificatesList, self).__init__() - @marshal_items(FIELDS) + @validate_schema(None, certificates_output_schema) def get(self): """ .. http:get:: /certificates @@ -208,8 +103,8 @@ class CertificatesList(AuthenticatedResource): args = parser.parse_args() return service.render(args) - @marshal_items(FIELDS) - def post(self): + @validate_schema(certificate_input_schema, certificate_output_schema) + def post(self, data=None): """ .. http:post:: /certificates @@ -346,48 +241,24 @@ class CertificatesList(AuthenticatedResource): :arg state: state for the CSR :arg location: location for the CSR :arg organization: organization for CSR - :arg commonName: certiifcate common name + :arg commonName: certificate common name :reqheader Authorization: OAuth token to authenticate :statuscode 200: no error :statuscode 403: unauthenticated """ - self.reqparse.add_argument('extensions', type=dict, location='json') - self.reqparse.add_argument('destinations', type=list, default=[], location='json') - self.reqparse.add_argument('notifications', type=list, default=[], location='json') - self.reqparse.add_argument('replacements', type=list, default=[], location='json') - self.reqparse.add_argument('validityStart', type=str, location='json') # TODO validate - self.reqparse.add_argument('validityEnd', type=str, location='json') # TODO validate - self.reqparse.add_argument('validityYears', type=int, location='json') # TODO validate - self.reqparse.add_argument('authority', type=valid_authority, location='json', required=True) - self.reqparse.add_argument('description', type=str, location='json') - self.reqparse.add_argument('country', type=str, location='json', required=True) - self.reqparse.add_argument('state', type=str, location='json', required=True) - self.reqparse.add_argument('location', type=str, location='json', required=True) - self.reqparse.add_argument('organization', type=str, location='json', required=True) - self.reqparse.add_argument('organizationalUnit', type=str, location='json', required=True) - self.reqparse.add_argument('owner', type=str, location='json', required=True) - self.reqparse.add_argument('commonName', type=str, location='json', required=True) - self.reqparse.add_argument('csr', type=str, location='json') - - args = self.reqparse.parse_args() - - authority = args['authority'] - role = role_service.get_by_name(authority.owner) + role = role_service.get_by_name(data['authority'].owner) # all the authority role members should be allowed - roles = [x.name for x in authority.roles] + roles = [x.name for x in data['authority'].roles] # allow "owner" roles by team DL roles.append(role) - authority_permission = AuthorityPermission(authority.id, roles) + authority_permission = AuthorityPermission(data['authority'].id, roles) if authority_permission.can(): - # if we are not admins lets make sure we aren't issuing anything sensitive - if not SensitiveDomainPermission().can(): - check_sensitive_domains(get_domains_from_options(args)) - return service.create(**args) + return service.create(**data) - return dict(message="You are not authorized to use {0}".format(args['authority'].name)), 403 + return dict(message="You are not authorized to use {0}".format(data['authority'].name)), 403 class CertificatesUpload(AuthenticatedResource): @@ -397,8 +268,8 @@ class CertificatesUpload(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(CertificatesUpload, self).__init__() - @marshal_items(FIELDS) - def post(self): + @validate_schema(certificate_upload_input_schema, certificate_output_schema) + def post(self, data=None): """ .. http:post:: /certificates/upload @@ -460,23 +331,12 @@ class CertificatesUpload(AuthenticatedResource): :statuscode 403: unauthenticated :statuscode 200: no error """ - self.reqparse.add_argument('description', type=str, location='json') - self.reqparse.add_argument('owner', type=str, required=True, location='json') - self.reqparse.add_argument('name', type=str, location='json') - self.reqparse.add_argument('publicCert', type=pem_str, required=True, dest='public_cert', location='json') - self.reqparse.add_argument('destinations', type=list, default=[], location='json') - self.reqparse.add_argument('notifications', type=list, default=[], location='json') - self.reqparse.add_argument('replacements', type=list, default=[], location='json') - self.reqparse.add_argument('intermediateCert', type=pem_str, dest='intermediate_cert', location='json') - self.reqparse.add_argument('privateKey', type=private_key_str, dest='private_key', location='json') - - args = self.reqparse.parse_args() - if args.get('destinations'): - if args.get('private_key'): - return service.upload(**args) + if data.get('destinations'): + if data.get('private_key'): + return service.upload(**data) else: raise Exception("Private key must be provided in order to upload certificate to AWS") - return service.upload(**args) + return service.upload(**data) class CertificatesStats(AuthenticatedResource): @@ -554,7 +414,7 @@ class Certificates(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(Certificates, self).__init__() - @marshal_items(FIELDS) + @validate_schema(None, certificate_output_schema) def get(self, certificate_id): """ .. http:get:: /certificates/1 @@ -603,8 +463,8 @@ class Certificates(AuthenticatedResource): """ return service.get(certificate_id) - @marshal_items(FIELDS) - def put(self, certificate_id): + @validate_schema(certificate_upload_input_schema, certificate_output_schema) + def put(self, certificate_id, data=None): """ .. http:put:: /certificates/1 @@ -657,14 +517,6 @@ class Certificates(AuthenticatedResource): :statuscode 200: no error :statuscode 403: unauthenticated """ - self.reqparse.add_argument('active', type=bool, location='json') - self.reqparse.add_argument('owner', type=str, location='json') - self.reqparse.add_argument('description', type=str, location='json') - self.reqparse.add_argument('destinations', type=list, default=[], location='json') - self.reqparse.add_argument('notifications', type=notification_list, default=[], location='json') - self.reqparse.add_argument('replacements', type=list, default=[], location='json') - args = self.reqparse.parse_args() - cert = service.get(certificate_id) role = role_service.get_by_name(cert.owner) @@ -673,12 +525,12 @@ class Certificates(AuthenticatedResource): if permission.can(): return service.update( certificate_id, - args['owner'], - args['description'], - args['active'], - args['destinations'], - args['notifications'], - args['replacements'] + data['owner'], + data['description'], + data['active'], + data['destinations'], + data['notifications'], + data['replacements'] ) return dict(message='You are not authorized to update this certificate'), 403 @@ -691,7 +543,7 @@ class NotificationCertificatesList(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(NotificationCertificatesList, self).__init__() - @marshal_items(FIELDS) + @validate_schema(None, certificates_output_schema) def get(self, notification_id): """ .. http:get:: /notifications/1/certificates @@ -767,7 +619,7 @@ class CertificatesReplacementsList(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(CertificatesReplacementsList, self).__init__() - @marshal_items(FIELDS) + @validate_schema(None, certificates_output_schema) def get(self, certificate_id): """ .. http:get:: /certificates/1/replacements @@ -822,7 +674,8 @@ class CertificateExport(AuthenticatedResource): self.reqparse = reqparse.RequestParser() super(CertificateExport, self).__init__() - def post(self, certificate_id): + @validate_schema(None, certificate_export_input_schema) + def post(self, certificate_id, data=None): """ .. http:post:: /certificates/1/export @@ -887,22 +740,21 @@ class CertificateExport(AuthenticatedResource): :statuscode 200: no error :statuscode 403: unauthenticated """ - self.reqparse.add_argument('export', type=dict, required=True, location='json') - args = self.reqparse.parse_args() - cert = service.get(certificate_id) role = role_service.get_by_name(cert.owner) permission = UpdateCertificatePermission(certificate_id, getattr(role, 'name', None)) - plugin = plugins.get(args['export']['plugin']['slug']) + options = data['export']['plugin']['plugin_options'] + plugin = data['export']['plugin'] + if plugin.requires_key: if permission.can(): - extension, passphrase, data = plugin.export(cert.body, cert.chain, cert.private_key, args['export']['plugin']['pluginOptions']) + extension, passphrase, data = plugin.export(cert.body, cert.chain, cert.private_key, options) else: return dict(message='You are not authorized to export this certificate'), 403 else: - extension, passphrase, data = plugin.export(cert.body, cert.chain, cert.private_key, args['export']['plugin']['pluginOptions']) + extension, passphrase, data = plugin.export(cert.body, cert.chain, cert.private_key, options) # we take a hit in message size when b64 encoding return dict(extension=extension, passphrase=passphrase, data=base64.b64encode(data)) diff --git a/lemur/common/schema.py b/lemur/common/schema.py new file mode 100644 index 00000000..5c2e8976 --- /dev/null +++ b/lemur/common/schema.py @@ -0,0 +1,146 @@ +""" +.. module: lemur.common.schema + :platform: unix + :copyright: (c) 2015 by Netflix Inc., see AUTHORS for more + :license: Apache, see LICENSE for more details. + +.. moduleauthor:: Kevin Glisson + +""" +from functools import wraps +from flask import request + +from sqlalchemy.orm.collections import InstrumentedList + +from marshmallow import Schema, post_dump, pre_load, pre_dump +from inflection import camelize, underscore + + +class LemurSchema(Schema): + """ + Base schema from which all grouper schema's inherit + """ + __envelope__ = True + + def under(self, data, many=None): + items = [] + if many: + for i in data: + items.append( + {underscore(key): value for key, value in i.items()} + ) + return items + return { + underscore(key): value + for key, value in data.items() + } + + def camel(self, data, many=None): + items = [] + if many: + for i in data: + items.append( + {camelize(key, uppercase_first_letter=False): value for key, value in i.items()} + ) + return items + return { + camelize(key, uppercase_first_letter=False): value + for key, value in data.items() + } + + def wrap_with_envelope(self, data, many): + if many: + if 'total' in self.context.keys(): + return dict(total=self.context['total'], items=data) + return data + + +class LemurInputSchema(LemurSchema): + @pre_load(pass_many=True) + def preprocess(self, data, many): + return self.under(data, many=many) + + +class LemurOutputSchema(LemurSchema): + @pre_load(pass_many=True) + def preprocess(self, data, many): + if many: + data = self.unwrap_envelope(data, many) + return self.under(data, many=many) + + @pre_dump(pass_many=True) + def unwrap_envelope(self, data, many): + if many: + if data: + if isinstance(data, InstrumentedList) or isinstance(data, list): + self.context['total'] = len(data) + return data + else: + self.context['total'] = data['total'] + else: + self.context['total'] = 0 + data = {'items': []} + + return data['items'] + + return data + + @post_dump(pass_many=True) + def post_process(self, data, many): + if data: + data = self.camel(data, many=many) + if self.__envelope__: + return self.wrap_with_envelope(data, many=many) + else: + return data + + +def format_errors(messages): + errors = {} + for k, v in messages.items(): + key = camelize(k, uppercase_first_letter=False) + if isinstance(v, dict): + errors[key] = format_errors(v) + elif isinstance(v, list): + errors[key] = v[0] + return errors + + +def wrap_errors(messages): + errors = dict(message='Validation Error.') + if messages.get('_schema'): + errors['reasons'] = {'Schema': {'rule': messages['_schema']}} + else: + errors['reasons'] = format_errors(messages) + return errors + + +def validate_schema(input_schema, output_schema): + def decorator(f): + @wraps(f) + def decorated_function(*args, **kwargs): + if input_schema: + if request.get_json(): + request_data = request.get_json() + else: + request_data = request.args + + data, errors = input_schema.load(request_data) + + if errors: + return wrap_errors(errors), 400 + + kwargs['data'] = data + + resp = f(*args, **kwargs) + + if not resp: + return dict(message="No data found"), 404 + + if output_schema: + data = output_schema.dump(resp) + return data.data, 200 + return resp, 200 + + return decorated_function + return decorator diff --git a/lemur/database.py b/lemur/database.py index b4df298c..f654d9f6 100644 --- a/lemur/database.py +++ b/lemur/database.py @@ -287,4 +287,7 @@ def sort_and_page(query, model, args): if sort_by and sort_dir: query = sort(query, model, sort_by, sort_dir) - return paginate(query, page, count) + total = query.count() + + items = query.offset(count * page).limit(count).all() + return dict(items=items, total=total) diff --git a/lemur/extensions.py b/lemur/extensions.py index d6325be1..50ae3bc8 100644 --- a/lemur/extensions.py +++ b/lemur/extensions.py @@ -3,16 +3,16 @@ :copyright: (c) 2015 by Netflix Inc., see AUTHORS for more :license: Apache, see LICENSE for more details. """ -from flask.ext.sqlalchemy import SQLAlchemy +from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() -from flask.ext.migrate import Migrate +from flask_migrate import Migrate migrate = Migrate() -from flask.ext.bcrypt import Bcrypt +from flask_bcrypt import Bcrypt bcrypt = Bcrypt() -from flask.ext.principal import Principal +from flask_principal import Principal principal = Principal() from flask_mail import Mail diff --git a/lemur/manage.py b/lemur/manage.py index 4e35d605..eebd01ee 100755 --- a/lemur/manage.py +++ b/lemur/manage.py @@ -391,7 +391,7 @@ class LemurServer(Command): settings = make_settings() options = ( Option(*klass.cli, action=klass.action) - for setting, klass in settings.iteritems() if klass.cli + for setting, klass in settings.items() if klass.cli ) return options diff --git a/lemur/plugins/lemur_aws/tests/test_iam.py b/lemur/plugins/lemur_aws/tests/test_iam.py index bc86feb7..44788cd5 100644 --- a/lemur/plugins/lemur_aws/tests/test_iam.py +++ b/lemur/plugins/lemur_aws/tests/test_iam.py @@ -2,7 +2,7 @@ from moto import mock_iam, mock_sts from lemur.certificates.models import Certificate -from lemur.tests.certs import EXTERNAL_VALID_STR, PRIVATE_KEY_STR +from lemur.tests.vectors import EXTERNAL_VALID_STR, PRIVATE_KEY_STR def test_get_name_from_arn(): diff --git a/lemur/plugins/lemur_email/tests/email.html b/lemur/plugins/lemur_email/tests/email.html new file mode 100644 index 00000000..e69de29b diff --git a/lemur/plugins/lemur_verisign/plugin.py b/lemur/plugins/lemur_verisign/plugin.py index ef0ba0e6..810bdf25 100644 --- a/lemur/plugins/lemur_verisign/plugin.py +++ b/lemur/plugins/lemur_verisign/plugin.py @@ -77,14 +77,14 @@ def process_options(options): 'email': current_app.config.get("VERISIGN_EMAIL") } - if options.get('validityEnd'): + if options.get('validity_end'): end_date, period = get_default_issuance(options) data['specificEndDate'] = str(end_date) data['validityPeriod'] = period - elif options.get('validityYears'): - if options['validityYears'] in [1, 2]: - data['validityPeriod'] = str(options['validityYears']) + 'Y' + elif options.get('validity_years'): + if options['validity_years'] in [1, 2]: + data['validityPeriod'] = str(options['validity_years']) + 'Y' else: raise Exception("Verisign issued certificates cannot exceed two years in validity") @@ -98,10 +98,10 @@ def get_default_issuance(options): :param options: :return: """ - specific_end_date = arrow.get(options['validityEnd']).replace(days=-1).format("MM/DD/YYYY") + specific_end_date = arrow.get(options['validity_end']).replace(days=-1).format("MM/DD/YYYY") now = arrow.utcnow() - then = arrow.get(options['validityEnd']) + then = arrow.get(options['validity_end']) if then < now.replace(years=+1): validity_period = '1Y' diff --git a/lemur/schemas.py b/lemur/schemas.py new file mode 100644 index 00000000..9b05d138 --- /dev/null +++ b/lemur/schemas.py @@ -0,0 +1,73 @@ +""" +.. module: lemur.schemas + :platform: unix + :copyright: (c) 2015 by Netflix Inc., see AUTHORS for more + :license: Apache, see LICENSE for more details. + +.. moduleauthor:: Kevin Glisson + +""" +from marshmallow import fields, post_load +from lemur.authorities.models import Authority +from lemur.destinations.models import Destination +from lemur.certificates.models import Certificate +from lemur.notifications.models import Notification +from lemur.common.schema import LemurInputSchema + +from lemur.plugins import plugins + + +class AssociatedAuthoritySchema(LemurInputSchema): + id = fields.Int(required=True) + + @post_load + def get_object(self, data, many=False): + return Authority.query.filter(Authority.id == data['id']).one() + + +class AssociatedDestinationSchema(LemurInputSchema): + id = fields.Int(required=True) + + @post_load + def get_object(self, data, many=False): + if many: + ids = [d['id'] for d in data] + return Destination.query.filter(Destination.id.in_(ids)).all() + else: + return Destination.query.filter(Destination.id == data['id']).one() + + +class AssociatedNotificationSchema(LemurInputSchema): + id = fields.Int(required=True) + + @post_load + def get_object(self, data, many=False): + if many: + ids = [d['id'] for d in data] + return Notification.query.filter(Notification.id.in_(ids)).all() + else: + return Notification.query.filter(Notification.id == data['id']).one() + + +class AssociatedCertificateSchema(LemurInputSchema): + id = fields.Int(required=True) + + @post_load + def get_object(self, data, many=False): + if many: + ids = [d['id'] for d in data] + return Certificate.query.filter(Certificate.id.in_(ids)).all() + else: + return Certificate.query.filter(Certificate.id == data['id']).one() + + +class PluginSchema(LemurInputSchema): + plugin_options = fields.Dict() + slug = fields.String() + + @post_load + def get_object(self, data, many=False): + if many: + return [plugins.get(plugin['slug']) for plugin in data] + else: + return plugins.get(data['slug']) diff --git a/lemur/static/app/angular/app.js b/lemur/static/app/angular/app.js index 7f0571f7..6b2f1049 100644 --- a/lemur/static/app/angular/app.js +++ b/lemur/static/app/angular/app.js @@ -89,6 +89,15 @@ }; }); + lemur.directive('lemurBadRequest', [function () { + return { + template: '

{{ directiveData.message }}

' + + '
' + + '{{ key | titleCase }} - {{ value }}' + + '
' + }; + }]); + lemur.factory('LemurRestangular', function (Restangular, $location, $auth) { return Restangular.withConfig(function (RestangularConfigurer) { RestangularConfigurer.setBaseUrl('http://localhost:8000/api/1'); @@ -109,18 +118,6 @@ return extractedData; }); - RestangularConfigurer.setErrorInterceptor(function(response) { - if (response.status === 400) { - if (response.data.message) { - var data = ''; - _.each(response.data.message, function (value, key) { - data = data + ' ' + key + ' ' + value; - }); - response.data.message = data; - } - } - }); - RestangularConfigurer.addFullRequestInterceptor(function (element, operation, route, url, headers, params) { // We want to make sure the user is auth'd before any requests if (!$auth.isAuthenticated()) { diff --git a/lemur/static/app/angular/certificates/certificate/certificate.js b/lemur/static/app/angular/certificates/certificate/certificate.js index e8fe2b9f..175cb4bd 100644 --- a/lemur/static/app/angular/certificates/certificate/certificate.js +++ b/lemur/static/app/angular/certificates/certificate/certificate.js @@ -41,7 +41,9 @@ angular.module('lemur') toaster.pop({ type: 'error', title: certificate.name, - body: 'Failed to export ' + response.data.message, + body: 'lemur-bad-request', + bodyOutputType: 'directive', + directiveData: response.data, timeout: 100000 }); }); @@ -73,7 +75,9 @@ angular.module('lemur') toaster.pop({ type: 'error', title: certificate.name, - body: 'Failed to update ' + response.data.message, + body: 'lemur-bad-request', + bodyOutputType: 'directive', + directiveData: response.data, timeout: 100000 }); }); @@ -109,9 +113,12 @@ angular.module('lemur') toaster.pop({ type: 'error', title: certificate.name, - body: 'Was not created! ' + response.data.message, + body: 'lemur-bad-request', + bodyOutputType: 'directive', + directiveData: response.data, timeout: 100000 }); + WizardHandler.wizard().context.loading = false; }); }; diff --git a/lemur/static/app/index.html b/lemur/static/app/index.html index 2ef06c17..34f6ce79 100644 --- a/lemur/static/app/index.html +++ b/lemur/static/app/index.html @@ -71,10 +71,10 @@
  • - + {{ currentUser.username }} - + {{ currentUser.username }} diff --git a/lemur/tests/conf.py b/lemur/tests/conf.py index e94873ec..f7dd5a75 100644 --- a/lemur/tests/conf.py +++ b/lemur/tests/conf.py @@ -38,6 +38,12 @@ LEMUR_SECURITY_TEAM_EMAIL = [] LOG_LEVEL = "DEBUG" LOG_FILE = "lemur.log" +LEMUR_DEFAULT_COUNTRY = 'US' +LEMUR_DEFAULT_STATE = 'California' +LEMUR_DEFAULT_LOCATION = 'Los Gatos' +LEMUR_DEFAULT_ORGANIZATION = 'Example, Inc.' +LEMUR_DEFAULT_ORGANIZATIONAL_UNIT = 'Example' + # Database diff --git a/lemur/tests/conftest.py b/lemur/tests/conftest.py index bc3479bb..f7fd6b28 100644 --- a/lemur/tests/conftest.py +++ b/lemur/tests/conftest.py @@ -1,15 +1,15 @@ import os import pytest +from flask import current_app + +from flask.ext.principal import identity_changed, Identity + from lemur import create_app from lemur.database import db as _db -from lemur.users import service as user_service -from lemur.roles import service as role_service - -def pytest_addoption(parser): - parser.addoption("--lemurconfig", help="override the default test config") - parser.addoption("--runslow", action="store_true", help="run slow tests") +from .factories import AuthorityFactory, NotificationFactory, DestinationFactory, \ + CertificateFactory, UserFactory, RoleFactory def pytest_runtest_setup(item): @@ -35,10 +35,7 @@ def app(request): Creates a new Flask application for a test duration. Uses application factory `create_app`. """ - if request.config.getoption('--lemurconfig'): - _app = create_app(request.config.getoption('--lemurconfig')) - else: - _app = create_app(os.path.dirname(os.path.realpath(__file__)) + '/conf.py') + _app = create_app(os.path.dirname(os.path.realpath(__file__)) + '/conf.py') ctx = _app.app_context() ctx.push() @@ -54,9 +51,10 @@ def db(app, request): _db.app = app - user = user_service.create('user', 'test', 'user@example.com', True, None, []) - admin_role = role_service.create('admin') - admin = user_service.create('admin', 'admin', 'admin@example.com', True, None, [admin_role]) + UserFactory() + r = RoleFactory(name='admin') + UserFactory(roles=[r]) + _db.session.commit() yield _db @@ -68,10 +66,52 @@ def session(db, request): for test duration. """ db.session.begin_nested() - yield session + yield db.session db.session.rollback() @pytest.yield_fixture(scope="function") def client(app, session, client): yield client + + +@pytest.fixture +def authority(session): + a = AuthorityFactory() + session.commit() + return a + + +@pytest.fixture +def destination(session): + d = DestinationFactory() + session.commit() + return d + + +@pytest.fixture +def notification(session): + n = NotificationFactory() + session.commit() + return n + + +@pytest.fixture +def certificate(session): + c = CertificateFactory() + session.commit() + return c + + +@pytest.yield_fixture(scope="function") +def logged_in_user(app, user): + with app.test_request_context(): + identity_changed.send(current_app._get_current_object(), identity=Identity(user.id)) + yield + + +@pytest.yield_fixture(scope="function") +def logged_in_admin(app, admin_user): + with app.test_request_context(): + identity_changed.send(current_app._get_current_object(), identity=Identity(admin_user.id)) + yield diff --git a/lemur/tests/factories.py b/lemur/tests/factories.py new file mode 100644 index 00000000..e57d2e6a --- /dev/null +++ b/lemur/tests/factories.py @@ -0,0 +1,210 @@ + +from datetime import date + +from factory import Sequence, post_generation +from factory.alchemy import SQLAlchemyModelFactory +from factory.fuzzy import FuzzyChoice, FuzzyText, FuzzyDate + + +from lemur.database import db +from lemur.authorities.models import Authority +from lemur.certificates.models import Certificate +from lemur.destinations.models import Destination +from lemur.notifications.models import Notification +from lemur.users.models import User +from lemur.roles.models import Role + +from .vectors import INTERNAL_VALID_SAN_STR, PRIVATE_KEY_STR + + +class BaseFactory(SQLAlchemyModelFactory): + """Base factory.""" + + class Meta: + """Factory configuration.""" + abstract = True + sqlalchemy_session = db.session + + +class AuthorityFactory(BaseFactory): + """Authority factory.""" + name = Sequence(lambda n: 'authority{0}'.format(n)) + owner = 'joe@example.com' + plugin_name = 'TheRing' + body = INTERNAL_VALID_SAN_STR + + class Meta: + """Factory configuration.""" + model = Authority + + @post_generation + def roles(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for role in extracted: + self.roles.append(role) + + +class CertificateFactory(BaseFactory): + """Certificate factory.""" + name = Sequence(lambda n: 'certificate{0}'.format(n)) + chain = INTERNAL_VALID_SAN_STR + body = INTERNAL_VALID_SAN_STR + private_key = PRIVATE_KEY_STR + owner = 'joe@example.com' + status = FuzzyChoice(['valid', 'revoked', 'unknown']) + deleted = False + bits = 2048 + issuer = 'Example' + serial = FuzzyText(length=128) + cn = 'test.example.com' + description = FuzzyText(length=128) + active = True + san = 'true' + not_before = FuzzyDate(date(2016, 1, 1), date(2020, 1, 1)) + not_after = FuzzyDate(date(2016, 1, 1), date(2020, 1, 1)) + date_created = FuzzyDate(date(2016, 1, 1), date(2020, 1, 1)) + + class Meta: + """Factory Configuration.""" + model = Certificate + + @post_generation + def user(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + self.user_id = extracted.id + + @post_generation + def authority(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + self.authority_id = extracted.id + + @post_generation + def notifications(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for notification in extracted: + self.notifications.append(notification) + + @post_generation + def destinations(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for destination in extracted: + self.destintations.append(destination) + + @post_generation + def replaces(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for replace in extracted: + self.replaces.append(replace) + + @post_generation + def sources(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for source in extracted: + self.sources.append(source) + + @post_generation + def domains(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for domain in extracted: + self.domains.append(domain) + + +class DestinationFactory(BaseFactory): + """Destination factory.""" + plugin_name = Sequence(lambda n: 'destination{0}'.format(n)) + label = Sequence(lambda n: 'destination{0}'.format(n)) + + class Meta: + """Factory Configuration.""" + model = Destination + + +class NotificationFactory(BaseFactory): + """Notification factory.""" + plugin_name = Sequence(lambda n: 'notification{0}'.format(n)) + label = Sequence(lambda n: 'notification{0}'.format(n)) + + class Meta: + """Factory Configuration.""" + model = Notification + + +class RoleFactory(BaseFactory): + """Role factory.""" + name = Sequence(lambda n: 'role{0}'.format(n)) + + class Meta: + """Factory Configuration.""" + model = Role + + @post_generation + def users(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for user in extracted: + self.users.append(user) + + +class UserFactory(BaseFactory): + """User Factory.""" + username = Sequence(lambda n: 'user{0}'.format(n)) + email = Sequence(lambda n: 'user{0}@example.com'.format(n)) + active = True + password = FuzzyText(length=24) + + class Meta: + """Factory Configuration.""" + model = User + + @post_generation + def roles(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for role in extracted: + self.roles.append(role) + + @post_generation + def certificates(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for cert in extracted: + self.certificates.append(cert) + + @post_generation + def authorities(self, create, extracted, **kwargs): + if not create: + return + + if extracted: + for authority in extracted: + self.authorities.append(authority) diff --git a/lemur/tests/test_certificates.py b/lemur/tests/test_certificates.py index fd3012d2..4ea5124a 100644 --- a/lemur/tests/test_certificates.py +++ b/lemur/tests/test_certificates.py @@ -1,35 +1,288 @@ from __future__ import unicode_literals # at top of module import pytest +import json from lemur.certificates.views import * # noqa - -def test_pem_str(): - from lemur.tests.certs import INTERNAL_VALID_LONG_STR - assert pem_str(INTERNAL_VALID_LONG_STR, 'test') == INTERNAL_VALID_LONG_STR - - with pytest.raises(ValueError): - pem_str('sdfsdfds', 'test') +from .vectors import VALID_ADMIN_HEADER_TOKEN, VALID_USER_HEADER_TOKEN -def test_private_key_str(): - from lemur.tests.certs import PRIVATE_KEY_STR - assert private_key_str(PRIVATE_KEY_STR, 'test') == PRIVATE_KEY_STR +def test_authority_identifier_schema(): + from lemur.certificates.schemas import AuthorityIdentifierSchema + input_data = {'useAuthorityCert': True} - with pytest.raises(ValueError): - private_key_str('dfsdfsdf', 'test') + data, errors = AuthorityIdentifierSchema().load(input_data) + + assert data == {'use_authority_cert': True} + assert not errors + + data, errors = AuthorityIdentifierSchema().dumps(data) + assert not errors + assert data == json.dumps(input_data) -def test_create_basic_csr(): +def test_authority_key_identifier_schema(): + from lemur.certificates.schemas import AuthorityKeyIdentifierSchema + input_data = {'useKeyIdentifier': True} + + data, errors = AuthorityKeyIdentifierSchema().load(input_data) + + assert data == {'use_key_identifier': True} + assert not errors + + data, errors = AuthorityKeyIdentifierSchema().dumps(data) + assert data == json.dumps(input_data) + assert not errors + + +def test_certificate_info_access_schema(): + from lemur.certificates.schemas import CertificateInfoAccessSchema + input_data = {'includeAIA': True} + + data, errors = CertificateInfoAccessSchema().load(input_data) + assert not errors + assert data == {'include_aia': True} + + data, errors = CertificateInfoAccessSchema().dump(data) + assert not errors + assert data == input_data + + +def test_subject_key_identifier_schema(): + from lemur.certificates.schemas import SubjectKeyIdentifierSchema + + input_data = {'includeSKI': True} + + data, errors = SubjectKeyIdentifierSchema().load(input_data) + assert not errors + assert data == {'include_ski': True} + data, errors = SubjectKeyIdentifierSchema().dump(data) + assert not errors + assert data == input_data + + +def test_extension_schema(): + from lemur.certificates.schemas import ExtensionSchema + + input_data = { + 'keyUsage': { + 'useKeyEncipherment': True, + 'useDigitalSignature': True + }, + 'extendedKeyUsage': { + 'useServerAuthentication': True + }, + 'subjectKeyIdentifier': { + 'includeSKI': True + }, + 'subAltNames': { + 'names': [ + {'nameType': 'DNSName', 'value': 'test.example.com'} + ] + } + } + + data, errors = ExtensionSchema().load(input_data) + assert not errors + + +def test_certificate_input_schema(client, authority): + from lemur.certificates.schemas import CertificateInputSchema + + input_data = { + 'commonName': 'test.example.com', + 'owner': 'jim@example.com', + 'authority': {'id': authority.id}, + 'description': 'testtestest', + } + + data, errors = CertificateInputSchema().load(input_data) + + assert not errors + assert data['authority'].id == authority.id + + # make sure the defaults got set + assert data['common_name'] == 'test.example.com' + assert data['country'] == 'US' + assert data['location'] == 'Los Gatos' + + assert len(data.keys()) == 12 + + +def test_certificate_input_with_extensions(client, authority): + from lemur.certificates.schemas import CertificateInputSchema + + input_data = { + 'commonName': 'test.example.com', + 'owner': 'jim@example.com', + 'authority': {'id': authority.id}, + 'description': 'testtestest', + 'extensions': { + 'keyUsage': { + 'useKeyEncipherment': True, + 'useDigitalSignature': True + }, + 'extendedKeyUsage': { + 'useServerAuthentication': True + }, + 'subjectKeyIdentifier': { + 'includeSKI': True + }, + 'subAltNames': { + 'names': [ + {'nameType': 'DNSName', 'value': 'test.example.com'} + ] + } + } + } + + data, errors = CertificateInputSchema().load(input_data) + assert not errors + + +def test_certificate_out_of_range_date(client, authority): + from lemur.certificates.schemas import CertificateInputSchema + input_data = { + 'commonName': 'test.example.com', + 'owner': 'jim@example.com', + 'authority': {'id': authority.id}, + 'description': 'testtestest', + 'validityYears': 100 + } + + data, errors = CertificateInputSchema().load(input_data) + assert errors + + input_data['validityStart'] = '2017-04-30T00:12:34.513631' + + data, errors = CertificateInputSchema().load(input_data) + assert errors + + input_data['validityEnd'] = '2018-04-30T00:12:34.513631' + + data, errors = CertificateInputSchema().load(input_data) + assert errors + + +def test_certificate_valid_years(client, authority): + from lemur.certificates.schemas import CertificateInputSchema + input_data = { + 'commonName': 'test.example.com', + 'owner': 'jim@example.com', + 'authority': {'id': authority.id}, + 'description': 'testtestest', + 'validityYears': 3 + } + + data, errors = CertificateInputSchema().load(input_data) + assert not errors + + +def test_certificate_valid_dates(client, authority): + from lemur.certificates.schemas import CertificateInputSchema + input_data = { + 'commonName': 'test.example.com', + 'owner': 'jim@example.com', + 'authority': {'id': authority.id}, + 'description': 'testtestest', + 'validityStart': '2017-04-30T00:12:34.513631', + 'validityEnd': '2018-04-30T00:12:34.513631' + } + + data, errors = CertificateInputSchema().load(input_data) + assert not errors + + +def test_sub_alt_name_schema(): + from lemur.certificates.schemas import SubAltNameSchema, SubAltNamesSchema + input_data = {'nameType': 'DNSName', 'value': 'test.example.com'} + + data, errors = SubAltNameSchema().load(input_data) + assert not errors + assert data == {'name_type': 'DNSName', 'value': 'test.example.com'} + + data, errors = SubAltNameSchema().dumps(data) + assert data == json.dumps(input_data) + assert not errors + + input_datas = {'names': [input_data]} + + data, errors = SubAltNamesSchema().load(input_datas) + assert not errors + assert data == {'names': [{'name_type': 'DNSName', 'value': 'test.example.com'}]} + + data, errors = SubAltNamesSchema().dumps(data) + assert data == json.dumps(input_datas) + assert not errors + + +def test_key_usage_schema(): + from lemur.certificates.schemas import KeyUsageSchema + + input_data = { + 'useCRLSign': True, + 'useDataEncipherment': True, + 'useDecipherOnly': True, + 'useEncipherOnly': True, + 'useKeyEncipherment': True, + 'useDigitalSignature': True, + 'useNonRepudiation': True + } + + data, errors = KeyUsageSchema().load(input_data) + + assert not errors + assert data == { + 'use_crl_sign': True, + 'use_data_encipherment': True, + 'use_decipher_only': True, + 'use_encipher_only': True, + 'use_key_encipherment': True, + 'use_digital_signature': True, + 'use_non_repudiation': True + } + + +def test_extended_key_usage_schema(): + from lemur.certificates.schemas import ExtendedKeyUsageSchema + + input_data = { + 'useServerAuthentication': True, + 'useClientAuthentication': True, + 'useEapOverLAN': True, + 'useEapOverPPP': True, + 'useOCSPSigning': True, + 'useSmartCardAuthentication': True, + 'useTimestamping': True + } + + data, errors = ExtendedKeyUsageSchema().load(input_data) + + assert not errors + assert data == { + 'use_server_authentication': True, + 'use_client_authentication': True, + 'use_eap_over_lan': True, + 'use_eap_over_ppp': True, + 'use_ocsp_signing': True, + 'use_smart_card_authentication': True, + 'use_timestamping': True + } + + +def test_create_basic_csr(client): + from cryptography import x509 + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives import serialization from lemur.certificates.service import create_csr csr_config = dict( - commonName='example.com', + common_name='example.com', organization='Example, Inc.', - organizationalUnit='Operations', + organizational_unit='Operations', country='US', state='CA', location='A place', - extensions=dict(names=dict(subAltNames=['test.example.com', 'test2.example.com'])) + extensions=dict(names=dict(sub_alt_names=['test.example.com', 'test2.example.com'])) ) csr, pem = create_csr(csr_config) @@ -39,61 +292,61 @@ def test_create_basic_csr(): assert name.value in csr_config.values() -def test_cert_get_cn(): - from lemur.tests.certs import INTERNAL_VALID_LONG_CERT +def test_cert_get_cn(client): + from .vectors import INTERNAL_VALID_LONG_CERT from lemur.certificates.models import get_cn assert get_cn(INTERNAL_VALID_LONG_CERT) == 'long.lived.com' -def test_cert_get_subAltDomains(): - from lemur.tests.certs import INTERNAL_VALID_SAN_CERT, INTERNAL_VALID_LONG_CERT +def test_cert_get_sub_alt_domains(client): + from .vectors import INTERNAL_VALID_SAN_CERT, INTERNAL_VALID_LONG_CERT from lemur.certificates.models import get_domains assert get_domains(INTERNAL_VALID_LONG_CERT) == [] assert get_domains(INTERNAL_VALID_SAN_CERT) == ['example2.long.com', 'example3.long.com'] -def test_cert_is_san(): - from lemur.tests.certs import INTERNAL_VALID_SAN_CERT, INTERNAL_VALID_LONG_CERT +def test_cert_is_san(client): + from .vectors import INTERNAL_VALID_SAN_CERT, INTERNAL_VALID_LONG_CERT from lemur.certificates.models import is_san - assert is_san(INTERNAL_VALID_LONG_CERT) == None # noqa - assert is_san(INTERNAL_VALID_SAN_CERT) == True # noqa + assert not is_san(INTERNAL_VALID_LONG_CERT) + assert is_san(INTERNAL_VALID_SAN_CERT) -def test_cert_is_wildcard(): - from lemur.tests.certs import INTERNAL_VALID_WILDCARD_CERT, INTERNAL_VALID_LONG_CERT +def test_cert_is_wildcard(client): + from .vectors import INTERNAL_VALID_WILDCARD_CERT, INTERNAL_VALID_LONG_CERT from lemur.certificates.models import is_wildcard - assert is_wildcard(INTERNAL_VALID_WILDCARD_CERT) == True # noqa - assert is_wildcard(INTERNAL_VALID_LONG_CERT) == None # noqa + assert is_wildcard(INTERNAL_VALID_WILDCARD_CERT) + assert not is_wildcard(INTERNAL_VALID_LONG_CERT) -def test_cert_get_bitstrength(): - from lemur.tests.certs import INTERNAL_VALID_LONG_CERT +def test_cert_get_bitstrength(client): + from .vectors import INTERNAL_VALID_LONG_CERT from lemur.certificates.models import get_bitstrength assert get_bitstrength(INTERNAL_VALID_LONG_CERT) == 2048 -def test_cert_get_issuer(): - from lemur.tests.certs import INTERNAL_VALID_LONG_CERT +def test_cert_get_issuer(client): + from .vectors import INTERNAL_VALID_LONG_CERT from lemur.certificates.models import get_issuer assert get_issuer(INTERNAL_VALID_LONG_CERT) == 'Example' -def test_get_name_from_arn(): +def test_get_name_from_arn(client): from lemur.certificates.models import get_name_from_arn arn = 'arn:aws:iam::11111111:server-certificate/mycertificate' assert get_name_from_arn(arn) == 'mycertificate' -def test_get_account_number(): +def test_get_account_number(client): from lemur.certificates.models import get_account_number arn = 'arn:aws:iam::11111111:server-certificate/mycertificate' assert get_account_number(arn) == '11111111' -def test_create_name(): +def test_create_name(client): from lemur.certificates.models import create_name from datetime import datetime assert create_name( @@ -112,203 +365,181 @@ def test_create_name(): ) == 'SAN-example.com-ExampleInc-20150507-20150512' -def test_certificate_get(client): - assert client.get(api.url_for(Certificates, certificate_id=1)).status_code == 401 - - -def test_certificate_post(client): - assert client.post(api.url_for(Certificates, certificate_id=1), data={}).status_code == 405 - - -def test_certificate_put(client): - assert client.put(api.url_for(Certificates, certificate_id=1), data={}).status_code == 401 - - -def test_certificate_delete(client): - assert client.delete(api.url_for(Certificates, certificate_id=1)).status_code == 405 - - -def test_certificate_patch(client): - assert client.patch(api.url_for(Certificates, certificate_id=1), data={}).status_code == 405 - - -def test_certificates_get(client): - assert client.get(api.url_for(CertificatesList)).status_code == 401 - - -def test_certificates_post(client): - assert client.post(api.url_for(CertificatesList), data={}).status_code == 401 - - -def test_certificates_put(client): - assert client.put(api.url_for(CertificatesList), data={}).status_code == 405 - - -def test_certificates_delete(client): - assert client.delete(api.url_for(CertificatesList)).status_code == 405 - - -def test_certificates_patch(client): - assert client.patch(api.url_for(CertificatesList), data={}).status_code == 405 - - -def test_certificate_credentials_get(client): - assert client.get(api.url_for(CertificatePrivateKey, certificate_id=1)).status_code == 401 - - -def test_certificate_credentials_post(client): - assert client.post(api.url_for(CertificatePrivateKey, certificate_id=1), data={}).status_code == 405 - - -def test_certificate_credentials_put(client): - assert client.put(api.url_for(CertificatePrivateKey, certificate_id=1), data={}).status_code == 405 - - -def test_certificate_credentials_delete(client): - assert client.delete(api.url_for(CertificatePrivateKey, certificate_id=1)).status_code == 405 - - -def test_certificate_credentials_patch(client): - assert client.patch(api.url_for(CertificatePrivateKey, certificate_id=1), data={}).status_code == 405 - - -def test_certificates_upload_get(client): - assert client.get(api.url_for(CertificatesUpload)).status_code == 405 - - -def test_certificates_upload_post(client): - assert client.post(api.url_for(CertificatesUpload), data={}).status_code == 401 - - -def test_certificates_upload_put(client): - assert client.put(api.url_for(CertificatesUpload), data={}).status_code == 405 - - -def test_certificates_upload_delete(client): - assert client.delete(api.url_for(CertificatesUpload)).status_code == 405 - - -def test_certificates_upload_patch(client): - assert client.patch(api.url_for(CertificatesUpload), data={}).status_code == 405 - - -VALID_USER_HEADER_TOKEN = { - 'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'} - - -def test_auth_certificate_get(client): - assert client.get(api.url_for(Certificates, certificate_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 200 - - -def test_auth_certificate_post_(client): - assert client.post(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificate_put(client): - assert client.put(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 400 - - -def test_auth_certificate_delete(client): - assert client.delete(api.url_for(Certificates, certificate_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificate_patch(client): - assert client.patch(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificates_get(client): - assert client.get(api.url_for(CertificatesList), headers=VALID_USER_HEADER_TOKEN).status_code == 200 - - -def test_auth_certificates_post(client): - assert client.post(api.url_for(CertificatesList), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 400 - - -def test_auth_certificate_credentials_get(client): - assert client.get(api.url_for(CertificatePrivateKey, certificate_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 404 - - -def test_auth_certificate_credentials_post(client): - assert client.post(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificate_credentials_put(client): - assert client.put(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificate_credentials_delete(client): - assert client.delete(api.url_for(CertificatePrivateKey, certificate_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificate_credentials_patch(client): - assert client.patch(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificates_upload_get(client): - assert client.get(api.url_for(CertificatesUpload), headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificates_upload_post(client): - assert client.post(api.url_for(CertificatesUpload), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 400 - - -def test_auth_certificates_upload_put(client): - assert client.put(api.url_for(CertificatesUpload), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificates_upload_delete(client): - assert client.delete(api.url_for(CertificatesUpload), headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -def test_auth_certificates_upload_patch(client): - assert client.patch(api.url_for(CertificatesUpload), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405 - - -VALID_ADMIN_HEADER_TOKEN = { - 'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyNTAyMTgsInN1YiI6MiwiZXhwIjoxNTIxNTYzODE4fQ.6mbq4-Ro6K5MmuNiTJBB153RDhlM5LGJBjI7GBKkfqA'} - - -def test_admin_certificate_get(client): - assert client.get(api.url_for(Certificates, certificate_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200 - - -def test_admin_certificate_post(client): - assert client.post(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificate_put(client): - assert client.put(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 400 - - -def test_admin_certificate_delete(client): - assert client.delete(api.url_for(Certificates, certificate_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificate_patch(client): - assert client.patch(api.url_for(Certificates, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificates_get(client): - resp = client.get(api.url_for(CertificatesList), headers=VALID_ADMIN_HEADER_TOKEN) - assert resp.status_code == 200 - assert resp.json['total'] == 0 - - -def test_admin_certificate_credentials_get(client): - assert client.get(api.url_for(CertificatePrivateKey, certificate_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 404 - - -def test_admin_certificate_credentials_post(client): - assert client.post(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificate_credentials_put(client): - assert client.put(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificate_credentials_delete(client): - assert client.delete(api.url_for(CertificatePrivateKey, certificate_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 - - -def test_admin_certificate_credentials_patch(client): - assert client.patch(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405 +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 404), + (VALID_ADMIN_HEADER_TOKEN, 404), + ('', 401) +]) +def test_certificate_get(client, token, status): + assert client.get(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_post(client, token, status): + assert client.post(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 400), + (VALID_ADMIN_HEADER_TOKEN, 400), + ('', 401) +]) +def test_certificate_put(client, token, status): + assert client.put(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_delete(client, token, status): + assert client.delete(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_patch(client, token, status): + assert client.patch(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 200), + (VALID_ADMIN_HEADER_TOKEN, 200), + ('', 401) +]) +def test_certificates_get(client, token, status): + assert client.get(api.url_for(CertificatesList), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 400), + (VALID_ADMIN_HEADER_TOKEN, 400), + ('', 401) +]) +def test_certificates_post(client, token, status): + assert client.post(api.url_for(CertificatesList), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_put(client, token, status): + assert client.put(api.url_for(CertificatesList), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_delete(client, token, status): + assert client.delete(api.url_for(CertificatesList), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_patch(client, token, status): + assert client.patch(api.url_for(CertificatesList), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 404), + (VALID_ADMIN_HEADER_TOKEN, 404), + ('', 401) +]) +def test_certificate_credentials_get(client, token, status): + assert client.get(api.url_for(CertificatePrivateKey, certificate_id=1), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_credentials_post(client, token, status): + assert client.post(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_credentials_put(client, token, status): + assert client.put(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_credentials_delete(client, token, status): + assert client.delete(api.url_for(CertificatePrivateKey, certificate_id=1), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificate_credentials_patch(client, token, status): + assert client.patch(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_upload_get(client, token, status): + assert client.get(api.url_for(CertificatesUpload), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 400), + (VALID_ADMIN_HEADER_TOKEN, 400), + ('', 401) +]) +def test_certificates_upload_post(client, token, status): + assert client.post(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_upload_put(client, token, status): + assert client.put(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_upload_delete(client, token, status): + assert client.delete(api.url_for(CertificatesUpload), headers=token).status_code == status + + +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 405), + (VALID_ADMIN_HEADER_TOKEN, 405), + ('', 405) +]) +def test_certificates_upload_patch(client, token, status): + assert client.patch(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status diff --git a/lemur/tests/certs.py b/lemur/tests/vectors.py similarity index 96% rename from lemur/tests/certs.py rename to lemur/tests/vectors.py index 1704fca9..0d3f03b4 100644 --- a/lemur/tests/certs.py +++ b/lemur/tests/vectors.py @@ -1,6 +1,14 @@ from cryptography import x509 from cryptography.hazmat.backends import default_backend +VALID_USER_HEADER_TOKEN = { + 'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'} + + +VALID_ADMIN_HEADER_TOKEN = { + 'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyNTAyMTgsInN1YiI6MiwiZXhwIjoxNTIxNTYzODE4fQ.6mbq4-Ro6K5MmuNiTJBB153RDhlM5LGJBjI7GBKkfqA'} + + INTERNAL_VALID_LONG_STR = b""" -----BEGIN CERTIFICATE----- MIID1zCCAr+gAwIBAgIBATANBgkqhkiG9w0BAQsFADCBjDELMAkGA1UEBhMCVVMx diff --git a/setup.py b/setup.py index cf23db95..64944f85 100644 --- a/setup.py +++ b/setup.py @@ -50,12 +50,15 @@ install_requires = [ 'boto==2.38.0', # we might make this optional 'six==1.10.0', 'gunicorn==19.4.1', + 'marshmallow-sqlalchemy==0.8.0', + 'marshmallow==2.4.0', 'pycrypto==2.6.1', - 'cryptography==1.1.2', + 'cryptography==1.3.1', 'pyopenssl==0.15.1', 'pyjwt==1.4.0', 'xmltodict==0.9.2', 'lockfile==0.12.2', + 'inflection==0.3.1', 'future==0.15.2', ] @@ -64,6 +67,7 @@ tests_require = [ 'moto==0.4.19', 'nose==1.3.7', 'pytest==2.8.5', + 'factory-boy==2.7.0', 'pytest-flask==0.10.0' ] diff --git a/tox.ini b/tox.ini index d8735b72..eaa43109 100644 --- a/tox.ini +++ b/tox.ini @@ -1,2 +1,2 @@ [tox] -envlist = py27,py34 +envlist = py27,py35