Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Adding tests to AWS plugin

This commit is contained in:
kevgliss 2015-07-23 08:52:56 -07:00
parent a4ed83cb62
commit 017eab6e39
8 changed files with 119 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
lemur/plugins/lemur_aws/iam.py
View File

@ -1,5 +1,5 @@
"""
.. module: lemur.common.services.aws.iam
.. module: lemur.plugins.lemur_aws.iam
:platform: Unix
:synopsis: Contains helper functions for interactive with AWS IAM Apis.
:copyright: (c) 2015 by Netflix Inc., see AUTHORS for more
@ -19,21 +19,6 @@ def get_name_from_arn(arn):
return arn.split("/", 1)[1]
def ssl_split(param_string):
"""
:param param_string:
:return:
"""
output = {}
parts = str(param_string).split("/")
for part in parts:
if "=" in part:
key, value = part.split("=", 1)
output[key] = value
return output
def upload_cert(account_number, cert, private_key, cert_chain=None):
"""
Upload a certificate to AWS

20
lemur/plugins/lemur_aws/plugin.py
View File

@ -58,10 +58,22 @@ class AWSSourcePlugin(SourcePlugin):
author = 'Kevin Glisson'
author_url = 'https://github.com/netflix/lemur'
options = {
'accountNumber': {'type': 'int'},
'pollRate': {'type': 'int', 'default': '60'}
}
options = [
{
'name': 'accountNumber',
'type': 'int',
'required': True,
'validation': '/^[0-9]{12,12}$/',
'helpMessage': 'Must be a valid AWS account number!',
},
{
'name': 'pollRate',
'type': 'int',
'required': False,
'helpMessage': 'Rate in seconds to poll source for new information.',
'default': '60',
}
]
def get_certificates(self, **kwargs):
certs = []

34
lemur/plugins/lemur_aws/tests/test_iam.py Normal file
View File

@ -0,0 +1,34 @@
import pytest
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.conftest import app # noqa
def test_get_name_from_arn():
from lemur.plugins.lemur_aws.iam import get_name_from_arn
arn = 'arn:aws:iam::123456789012:server-certificate/tttt2.netflixtest.net-NetflixInc-20150624-20150625'
assert get_name_from_arn(arn) == 'tttt2.netflixtest.net-NetflixInc-20150624-20150625'
@mock_sts()
@mock_iam()
def test_get_all_server_certs(app):
from lemur.plugins.lemur_aws.iam import upload_cert, get_all_server_certs
cert = Certificate(EXTERNAL_VALID_STR)
upload_cert('123456789012', cert, PRIVATE_KEY_STR)
certs = get_all_server_certs('123456789012')
assert len(certs) == 1
@mock_sts()
@mock_iam()
def test_get_cert_from_arn(app):
from lemur.plugins.lemur_aws.iam import upload_cert, get_cert_from_arn
cert = Certificate(EXTERNAL_VALID_STR)
upload_cert('123456789012', cert, PRIVATE_KEY_STR)
body, chain = get_cert_from_arn('arn:aws:iam::123456789012:server-certificate/tttt2.netflixtest.net-NetflixInc-20150624-20150625')
assert body.replace('\n', '') == EXTERNAL_VALID_STR.replace('\n', '')

76
lemur/plugins/lemur_verisign/plugin.py
View File

@ -16,6 +16,7 @@ from flask import current_app
from lemur.plugins.bases import IssuerPlugin
from lemur.plugins import lemur_verisign as verisign
from lemur.plugins.lemur_verisign import constants
from lemur.common.utils import get_psuedo_random_string
# https://support.venafi.com/entries/66445046-Info-VeriSign-Error-Codes
@ -58,9 +59,57 @@ VERISIGN_ERRORS = {
}
def process_options(options):
"""
Processes and maps the incoming issuer options to fields/options that
verisign understands
:param options:
:return: dict or valid verisign options
"""
data = {
'challenge': get_psuedo_random_string(),
'serverType': 'Apache',
'certProductType': 'Server',
'firstName': current_app.config.get("VERISIGN_FIRST_NAME"),
'lastName': current_app.config.get("VERISIGN_LAST_NAME"),
'signatureAlgorithm': 'sha256WithRSAEncryption',
'email': current_app.config.get("VERISIGN_EMAIL")
}
if options.get('validityEnd'):
end_date, period = get_default_issuance(options)
data['specificEndDate'] = end_date
data['validityPeriod'] = period
return data
def get_default_issuance(options):
"""
Gets the default time range for certificates
:param options:
:return:
"""
specific_end_date = arrow.get(options['validityEnd']).replace(days=-1).format("MM/DD/YYYY")
now = arrow.utcnow()
then = arrow.get(options['validityEnd'])
if then < now.replace(years=+1):
validity_period = '1Y'
elif then < now.replace(years=+2):
validity_period = '2Y'
else:
raise Exception("Verisign issued certificates cannot exceed two years in validity")
return specific_end_date, validity_period
def handle_response(content):
"""
Helper function that helps with parsing responses from the Verisign API.
Helper function for parsing responses from the Verisign API.
:param content:
:return: :raise Exception:
"""
@ -99,29 +148,8 @@ class VerisignIssuerPlugin(IssuerPlugin):
"""
url = current_app.config.get("VERISIGN_URL") + '/enroll'
data = {
'csr': csr,
'challenge': issuer_options['challenge'],
'serverType': 'Apache',
'certProductType': 'Server',
'firstName': current_app.config.get("VERISIGN_FIRST_NAME"),
'lastName': current_app.config.get("VERISIGN_LAST_NAME"),
'signatureAlgorithm': 'sha256WithRSAEncryption',
'email': current_app.config.get("VERISIGN_EMAIL")
}
if issuer_options.get('validityEnd'):
data['specificEndDate'] = arrow.get(issuer_options['validityEnd']).replace(days=-1).format("MM/DD/YYYY")
now = arrow.utcnow()
then = arrow.get(issuer_options['validityEnd'])
if then < now.replace(years=+1):
data['validityPeriod'] = '1Y'
elif then < now.replace(years=+2):
data['validityPeriod'] = '2Y'
else:
raise Exception("Verisign issued certificates cannot exceed two years in validity")
data = process_options(issuer_options)
data['csr'] = csr
current_app.logger.info("Requesting a new verisign certificate: {0}".format(data))

4
lemur/tests/__init__.py
View File

@ -3,3 +3,7 @@ import unittest
class LemurTestCase(unittest.TestCase):
pass
class LemurPluginTestCase(LemurTestCase):
pass

44
lemur/tests/conf.py
View File

@ -8,9 +8,7 @@ ADMINS = frozenset([''])
THREADS_PER_PAGE = 8
#############
## General ##
#############
# General
# These will need to be set to `True` if you are developing locally
CORS = False
@ -28,48 +26,30 @@ LEMUR_ENCRYPTION_KEY = 'jPd2xwxgVGXONqghHNq7/S761sffYSrT3UAgKwgtMxbqa0gmKYCfag==
# this is a list of domains as regexes that only admins can issue
LEMUR_RESTRICTED_DOMAINS = []
#################
## Mail Server ##
#################
# Mail Server
# Lemur currently only supports SES for sending email, this address
# needs to be verified
LEMUR_EMAIL = ''
LEMUR_SECURITY_TEAM_EMAIL = []
#############
## Logging ##
#############
# Logging
LOG_LEVEL = "DEBUG"
LOG_FILE = "lemur.log"
##############
## Database ##
##############
# Database
# modify this if you are not using a local database
SQLALCHEMY_DATABASE_URI = 'postgresql://lemur:lemur@localhost:5432/lemur'
#########
## AWS ##
#########
# AWS
# Lemur will need STS assume role access to every destination you want to monitor
#AWS_ACCOUNT_MAPPINGS = {
# '1111111111': 'myawsacount'
#}
LEMUR_INSTANCE_PROFILE = 'Lemur'
## This is useful if you know you only want to monitor one destination
#AWS_REGIONS = ['us-east-1']
#LEMUR_INSTANCE_PROFILE = 'Lemur'
#############
## Issuers ##
#############
# Issuers
# These will be dependent on which 3rd party that Lemur is
# configured to use.
@ -81,8 +61,8 @@ SQLALCHEMY_DATABASE_URI = 'postgresql://lemur:lemur@localhost:5432/lemur'
# number of years to issue if not specified
#CLOUDCA_DEFAULT_VALIDITY = 2
#VERISIGN_URL = ''
#VERISIGN_PEM_PATH = ''
#VERISIGN_FIRST_NAME = ''
#VERISIGN_LAST_NAME = ''
#VERSIGN_EMAIL = ''
VERISIGN_URL = 'http://example.com'
VERISIGN_PEM_PATH = '~/'
VERISIGN_FIRST_NAME = 'Jim'
VERISIGN_LAST_NAME = 'Bob'
VERSIGN_EMAIL = 'jim@example.com'

51
lemur/tests/test_elb.py
View File

@ -1,51 +0,0 @@
# import boto
# from lemur.tests import LemurTestCase
# from moto import mock_elb, mock_sts
# class ELBTestCase(LemurTestCase):
# @mock_sts
# @mock_elb
# def test_add_listener(self):
# from lemur.common.services.aws.elb import create_new_listeners
# conn = boto.connect_elb()
# zones = ['us-east-1a', 'us-east-1b']
# ports = [(80, 8080, 'http')]
# conn.create_load_balancer('my-lb', zones, ports)
# create_new_listeners('111', 'us-east-1', 'my-lb', listeners=[('443', '80', 'HTTP')])
# balancer = conn.get_all_load_balancers()[0]
# self.assertEqual(balancer.name, "my-lb")
# self.assertEqual(len(balancer.listeners), 2)
#
# @mock_sts
# @mock_elb
# def test_update_listener(self):
# from lemur.common.services.aws.elb import update_listeners
# conn = boto.connect_elb()
# zones = ['us-east-1a', 'us-east-1b']
# ports = [(80, 8080, 'http')]
# conn.create_load_balancer('my-lb', zones, ports)
# update_listeners('111', 'us-east-1', 'my-lb', listeners=[('80', '7001', 'http')])
# balancer = conn.get_all_load_balancers()[0]
# listener = balancer.listeners[0]
# self.assertEqual(listener.load_balancer_port, 80)
# self.assertEqual(listener.instance_port, 7001)
# self.assertEqual(listener.protocol, "HTTP")
#
# @mock_sts
# @mock_elb
# def test_set_certificate(self):
# from lemur.common.services.aws.elb import attach_certificate
# conn = boto.connect_elb()
# zones = ['us-east-1a', 'us-east-1b']
# ports = [(443, 7001, 'https', 'sslcert')]
# conn.create_load_balancer('my-lb', zones, ports)
# attach_certificate('1111', 'us-east-1', 'my-lb', 443, 'somecert')
# balancer = conn.get_all_load_balancers()[0]
# listener = balancer.listeners[0]
# self.assertEqual(listener.load_balancer_port, 443)
# self.assertEqual(listener.instance_port, 7001)
# self.assertEqual(listener.protocol, "HTTPS")
# self.assertEqual(listener.ssl_certificate_id, 'somecert')
#

35
lemur/tests/test_iam.py
View File

@ -1,35 +0,0 @@
# from lemur.tests import LemurTestCase
# from lemur.certificates.models import Certificate
# from moto import mock_iam, mock_sts
# class IAMTestCase(LemurTestCase):
# @mock_sts
# @mock_iam
# def test_get_all_server_certs(self):
# from lemur.common.services.aws.iam import upload_cert, get_all_server_certs
# cert = Certificate(TEST_CERT)
# upload_cert('1111', cert, TEST_KEY)
# certs = get_all_server_certs('1111')
# self.assertEquals(len(certs), 1)
#
# @mock_sts
# @mock_iam
# def test_get_server_cert(self):
# from lemur.common.services.aws.iam import upload_cert, get_cert_from_arn
# cert = Certificate(TEST_CERT)
# upload_cert('1111', cert, TEST_KEY)
# body, chain = get_cert_from_arn('arn:aws:iam::123456789012:server-certificate/AHB-dfdsflkj.net-NetflixInc-20140525-20150525')
# self.assertTrue(body)
#
# @mock_sts
# @mock_iam
# def test_upload_server_cert(self):
# from lemur.common.services.aws.iam import upload_cert
# cert = Certificate(TEST_CERT)
# response = upload_cert('1111', cert, TEST_KEY)
# self.assertEquals(response['upload_server_certificate_response']['upload_server_certificate_result']['server_certificate_metadata']['server_certificate_name'], 'AHB-dfdsflkj.net-NetflixInc-20140525-20150525')
#
#