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

Merge branch 'master' into linuxdst

Browse Source
This commit is contained in:
Curtis
2018-05-08 12:09:05 -07:00
committed by GitHub
parent 28614b5793 c26ae16060
commit 642dbd4098
38 changed files with 446 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
lemur/__init__.py
View File

@ -8,7 +8,8 @@
"""
from __future__ import absolute_import, division, print_function
import time
from flask import g, request
from lemur import factory
from lemur.extensions import metrics
@ -73,17 +74,6 @@ def configure_hook(app):
"""
from flask import jsonify
from werkzeug.exceptions import HTTPException
from lemur.decorators import crossdomain
if app.config.get('CORS'):
@app.after_request
@crossdomain(origin=u"http://localhost:3000", methods=['PUT', 'HEAD', 'GET', 'POST', 'OPTIONS', 'DELETE'])
def after(response):
return response
@app.after_request
def log_status(response):
metrics.send('status_code_{}'.format(response.status_code), 'counter', 1)
return response
@app.errorhandler(Exception)
def handle_error(e):
@ -93,3 +83,29 @@ def configure_hook(app):
app.logger.exception(e)
return jsonify(error=str(e)), code
@app.before_request
def before_request():
g.request_start_time = time.time()
@app.after_request
def after_request(response):
# Return early if we don't have the start time
if not hasattr(g, 'request_start_time'):
return response
# Get elapsed time in milliseconds
elapsed = time.time() - g.request_start_time
elapsed = int(round(1000 * elapsed))
# Collect request/response tags
tags = {
'endpoint': request.endpoint,
'request_method': request.method.lower(),
'status_code': response.status_code
}
# Record our response time metric
metrics.send('response_time', 'TIMER', elapsed, metric_tags=tags)
metrics.send('status_code_{}'.format(response.status_code), 'counter', 1)
return response

5
lemur/authorities/service.py
View File

@ -9,6 +9,7 @@
"""
from lemur import database
from lemur.common.utils import truthiness
from lemur.extensions import metrics
from lemur.authorities.models import Authority
from lemur.roles import service as role_service
@ -170,8 +171,8 @@ def render(args):
if filt:
terms = filt.split(';')
if 'active' in filt: # this is really weird but strcmp seems to not work here??
query = query.filter(Authority.active == terms[1])
if 'active' in filt:
query = query.filter(Authority.active == truthiness(terms[1]))
else:
query = database.filter(query, Authority, terms)

8
lemur/certificates/service.py
View File

@ -8,7 +8,7 @@
import arrow
from flask import current_app
from sqlalchemy import func, or_, not_, cast, Boolean, Integer
from sqlalchemy import func, or_, not_, cast, Integer
from cryptography import x509
from cryptography.hazmat.backends import default_backend
@ -17,7 +17,7 @@ from cryptography.hazmat.primitives import hashes, serialization
from lemur import database
from lemur.extensions import metrics, signals
from lemur.plugins.base import plugins
from lemur.common.utils import generate_private_key
from lemur.common.utils import generate_private_key, truthiness
from lemur.roles.models import Role
from lemur.domains.models import Domain
@ -319,9 +319,9 @@ def render(args):
elif 'destination' in terms:
query = query.filter(Certificate.destinations.any(Destination.id == terms[1]))
elif 'notify' in filt:
query = query.filter(Certificate.notify == cast(terms[1], Boolean))
query = query.filter(Certificate.notify == truthiness(terms[1]))
elif 'active' in filt:
query = query.filter(Certificate.active == terms[1])
query = query.filter(Certificate.active == truthiness(terms[1]))
elif 'cn' in terms:
query = query.filter(
or_(

6
lemur/certificates/views.py
View File

@ -300,12 +300,14 @@ class CertificatesUpload(AuthenticatedResource):
{
"owner": "joe@example.com",
"publicCert": "-----BEGIN CERTIFICATE-----...",
"intermediateCert": "-----BEGIN CERTIFICATE-----...",
"body": "-----BEGIN CERTIFICATE-----...",
"chain": "-----BEGIN CERTIFICATE-----...",
"privateKey": "-----BEGIN RSA PRIVATE KEY-----..."
"destinations": [],
"notifications": [],
"replacements": [],
"roles": [],
"notify": true,
"name": "cert1"
}

48
lemur/common/utils.py
View File

@ -14,10 +14,11 @@ from sqlalchemy import and_, func
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from flask_restful.reqparse import RequestParser
from lemur.constants import CERTIFICATE_KEY_TYPES
from lemur.exceptions import InvalidConfiguration
paginated_parser = RequestParser()
@ -78,17 +79,43 @@ def generate_private_key(key_type):
"""
Generates a new private key based on key_type.
Valid key types: RSA2048, RSA4096
Valid key types: RSA2048, RSA4096', 'ECCPRIME192V1', 'ECCPRIME256V1', 'ECCSECP192R1',
'ECCSECP224R1', 'ECCSECP256R1', 'ECCSECP384R1', 'ECCSECP521R1', 'ECCSECP256K1',
'ECCSECT163K1', 'ECCSECT233K1', 'ECCSECT283K1', 'ECCSECT409K1', 'ECCSECT571K1',
'ECCSECT163R2', 'ECCSECT233R1', 'ECCSECT283R1', 'ECCSECT409R1', 'ECCSECT571R2'
:param key_type:
:return:
"""
valid_key_types = ['RSA2048', 'RSA4096']
if key_type not in valid_key_types:
_CURVE_TYPES = {
"ECCPRIME192V1": ec.SECP192R1(),
"ECCPRIME256V1": ec.SECP256R1(),
"ECCSECP192R1": ec.SECP192R1(),
"ECCSECP224R1": ec.SECP224R1(),
"ECCSECP256R1": ec.SECP256R1(),
"ECCSECP384R1": ec.SECP384R1(),
"ECCSECP521R1": ec.SECP521R1(),
"ECCSECP256K1": ec.SECP256K1(),
"ECCSECT163K1": ec.SECT163K1(),
"ECCSECT233K1": ec.SECT233K1(),
"ECCSECT283K1": ec.SECT283K1(),
"ECCSECT409K1": ec.SECT409K1(),
"ECCSECT571K1": ec.SECT571K1(),
"ECCSECT163R2": ec.SECT163R2(),
"ECCSECT233R1": ec.SECT233R1(),
"ECCSECT283R1": ec.SECT283R1(),
"ECCSECT409R1": ec.SECT409R1(),
"ECCSECT571R2": ec.SECT571R1(),
}
if key_type not in CERTIFICATE_KEY_TYPES:
raise Exception("Invalid key type: {key_type}. Supported key types: {choices}".format(
key_type=key_type,
choices=",".join(valid_key_types)
choices=",".join(CERTIFICATE_KEY_TYPES)
))
if 'RSA' in key_type:
@ -98,6 +125,11 @@ def generate_private_key(key_type):
key_size=key_size,
backend=default_backend()
)
elif 'ECC' in key_type:
return ec.generate_private_key(
curve=_CURVE_TYPES[key_type],
backend=default_backend()
)
def is_weekend(date):
@ -175,3 +207,9 @@ def windowed_query(q, column, windowsize):
column, windowsize):
for row in q.filter(whereclause).order_by(column):
yield row
def truthiness(s):
"""If input string resembles something truthy then return True, else False."""
return s.lower() in ('true', 'yes', 'on', 't', '1')

23
lemur/constants.py
View File

@ -9,3 +9,26 @@ NONSTANDARD_NAMING_TEMPLATE = "{issuer}-{not_before}-{not_after}"
SUCCESS_METRIC_STATUS = 'success'
FAILURE_METRIC_STATUS = 'failure'
CERTIFICATE_KEY_TYPES = [
'RSA2048',
'RSA4096',
'ECCPRIME192V1',
'ECCPRIME256V1',
'ECCSECP192R1',
'ECCSECP224R1',
'ECCSECP256R1',
'ECCSECP384R1',
'ECCSECP521R1',
'ECCSECP256K1',
'ECCSECT163K1',
'ECCSECT233K1',
'ECCSECT283K1',
'ECCSECT409K1',
'ECCSECT571K1',
'ECCSECT163R2',
'ECCSECT233R1',
'ECCSECT283R1',
'ECCSECT409R1',
'ECCSECT571R2'
]

56
lemur/decorators.py
View File

@ -1,56 +0,0 @@
"""
.. module: lemur.decorators
:copyright: (c) 2015 by Netflix Inc., see AUTHORS for more
:license: Apache, see LICENSE for more details.
"""
from builtins import str
from datetime import timedelta
from flask import make_response, request, current_app
from functools import update_wrapper
# this is only used for dev
def crossdomain(origin=None, methods=None, headers=None,
max_age=21600, attach_to_all=True,
automatic_options=True): # pragma: no cover
if methods is not None:
methods = ', '.join(sorted(x.upper() for x in methods))
if headers is not None and not isinstance(headers, str):
headers = ', '.join(x.upper() for x in headers)
if not isinstance(origin, str):
origin = ', '.join(origin)
if isinstance(max_age, timedelta):
max_age = max_age.total_seconds()
def get_methods():
if methods is not None:
return methods
options_resp = current_app.make_default_options_response()
return options_resp.headers['allow']
def decorator(f):
def wrapped_function(*args, **kwargs):
if automatic_options and request.method == 'OPTIONS':
resp = current_app.make_default_options_response()
else:
resp = make_response(f(*args, **kwargs))
if not attach_to_all and request.method != 'OPTIONS':
return resp
h = resp.headers
h['Access-Control-Allow-Origin'] = origin
h['Access-Control-Allow-Methods'] = get_methods()
h['Access-Control-Max-Age'] = str(max_age)
h['Access-Control-Allow-Headers'] = "Origin, X-Requested-With, Content-Type, Accept, Authorization "
h['Access-Control-Allow-Credentials'] = 'true'
return resp
f.provide_automatic_options = False
return update_wrapper(wrapped_function, f)
return decorator

3
lemur/endpoints/service.py
View File

@ -13,6 +13,7 @@ import arrow
from sqlalchemy import func
from lemur import database
from lemur.common.utils import truthiness
from lemur.endpoints.models import Endpoint, Policy, Cipher
from lemur.extensions import metrics
@ -142,7 +143,7 @@ def render(args):
if filt:
terms = filt.split(';')
if 'active' in filt: # this is really weird but strcmp seems to not work here??
query = query.filter(Endpoint.active == terms[1])
query = query.filter(Endpoint.active == truthiness(terms[1]))
elif 'port' in filt:
if terms[1] != 'null': # ng-table adds 'null' if a number is removed
query = query.filter(Endpoint.port == terms[1])

3
lemur/extensions.py
View File

@ -26,3 +26,6 @@ sentry = Sentry()
from blinker import Namespace
signals = Namespace()
from flask_cors import CORS
cors = CORS()

6
lemur/factory.py
View File

@ -21,7 +21,7 @@ from flask import Flask
from lemur.certificates.hooks import activate_debug_dump
from lemur.common.health import mod as health
from lemur.extensions import db, migrate, principal, smtp_mail, metrics, sentry
from lemur.extensions import db, migrate, principal, smtp_mail, metrics, sentry, cors
DEFAULT_BLUEPRINTS = (
@ -125,6 +125,10 @@ def configure_extensions(app):
metrics.init_app(app)
sentry.init_app(app)
if app.config['CORS']:
app.config['CORS_HEADERS'] = 'Content-Type'
cors.init_app(app, resources=r'/api/*', headers='Content-Type', origin='*', supports_credentials=True)
def configure_blueprints(app, blueprints):
"""

26
lemur/manage.py
View File

@ -208,16 +208,16 @@ class InitializeApp(Command):
if operator_role:
sys.stdout.write("[-] Operator role already created, skipping...!\n")
else:
# we create an admin role
# we create an operator role
operator_role = role_service.create('operator', description='This is the Lemur operator role.')
sys.stdout.write("[+] Created 'operator' role\n")
read_only_role = role_service.get_by_name('read-only')
if read_only_role:
sys.stdout.write("[-] Operator role already created, skipping...!\n")
sys.stdout.write("[-] Read only role already created, skipping...!\n")
else:
# we create an admin role
# we create an read only role
read_only_role = role_service.create('read-only', description='This is the Lemur read only role.')
sys.stdout.write("[+] Created 'read-only' role\n")
@ -237,9 +237,6 @@ class InitializeApp(Command):
else:
sys.stdout.write("[-] Default user has already been created, skipping...!\n")
sys.stdout.write("[+] Creating expiration email notifications!\n")
sys.stdout.write("[!] Using {0} as specified by LEMUR_SECURITY_TEAM_EMAIL for notifications\n".format("LEMUR_SECURITY_TEAM_EMAIL"))
intervals = current_app.config.get("LEMUR_DEFAULT_EXPIRATION_NOTIFICATION_INTERVALS", [])
sys.stdout.write(
"[!] Creating {num} notifications for {intervals} days as specified by LEMUR_DEFAULT_EXPIRATION_NOTIFICATION_INTERVALS\n".format(
@ -249,14 +246,21 @@ class InitializeApp(Command):
)
recipients = current_app.config.get('LEMUR_SECURITY_TEAM_EMAIL')
sys.stdout.write("[+] Creating expiration email notifications!\n")
sys.stdout.write("[!] Using {0} as specified by LEMUR_SECURITY_TEAM_EMAIL for notifications\n".format(recipients))
notification_service.create_default_expiration_notifications("DEFAULT_SECURITY", recipients=recipients)
days = current_app.config.get("LEMUR_DEFAULT_ROTATION_INTERVAL", 30)
sys.stdout.write("[+] Creating default certificate rotation policy of {days} days before issuance.\n".format(
days=days
))
_DEFAULT_ROTATION_INTERVAL = 'default'
default_rotation_interval = policy_service.get_by_name(_DEFAULT_ROTATION_INTERVAL)
if default_rotation_interval:
sys.stdout.write("[-] Default rotation interval policy already created, skipping...!\n")
else:
days = current_app.config.get("LEMUR_DEFAULT_ROTATION_INTERVAL", 30)
sys.stdout.write("[+] Creating default certificate rotation policy of {days} days before issuance.\n".format(
days=days))
policy_service.create(days=days, name=_DEFAULT_ROTATION_INTERVAL)
policy_service.create(days=days, name='default')
sys.stdout.write("[/] Done!\n")

7
lemur/notifications/service.py
View File

@ -12,6 +12,7 @@ from flask import current_app
from lemur import database
from lemur.certificates.models import Certificate
from lemur.common.utils import truthiness
from lemur.notifications.models import Notification
@ -169,10 +170,8 @@ def render(args):
if filt:
terms = filt.split(';')
if terms[0] == 'active' and terms[1] == 'false':
query = query.filter(Notification.active == False) # noqa
elif terms[0] == 'active' and terms[1] == 'true':
query = query.filter(Notification.active == True) # noqa
if terms[0] == 'active':
query = query.filter(Notification.active == truthiness(terms[1]))
else:
query = database.filter(query, Notification, terms)

7
lemur/pending_certificates/service.py
View File

@ -5,9 +5,10 @@
"""
import arrow
from sqlalchemy import or_, cast, Boolean, Integer
from sqlalchemy import or_, cast, Integer
from lemur import database
from lemur.common.utils import truthiness
from lemur.plugins.base import plugins
from lemur.roles.models import Role
@ -181,9 +182,9 @@ def render(args):
elif 'destination' in terms:
query = query.filter(PendingCertificate.destinations.any(Destination.id == terms[1]))
elif 'notify' in filt:
query = query.filter(PendingCertificate.notify == cast(terms[1], Boolean))
query = query.filter(PendingCertificate.notify == truthiness(terms[1]))
elif 'active' in filt:
query = query.filter(PendingCertificate.active == terms[1])
query = query.filter(PendingCertificate.active == truthiness(terms[1]))
elif 'cn' in terms:
query = query.filter(
or_(

7
lemur/plugins/lemur_digicert/plugin.py
View File

@ -157,7 +157,7 @@ def map_cis_fields(options, csr):
"csr": csr,
"signature_hash": signature_hash(options.get('signing_algorithm')),
"validity": {
"valid_to": options['validity_end'].format('YYYY-MM-DD')
"valid_to": options['validity_end'].format('YYYY-MM-DDTHH:MM') + 'Z'
},
"organization": {
"name": options['organization'],
@ -491,6 +491,11 @@ class DigiCertCISIssuerPlugin(IssuerPlugin):
self.session.headers.pop('Accept')
end_entity = pem.parse(certificate_pem)[0]
if 'ECC' in issuer_options['key_type']:
return "\n".join(str(end_entity).splitlines()), current_app.config.get('DIGICERT_ECC_CIS_INTERMEDIATE'), data['id']
# By default return RSA
return "\n".join(str(end_entity).splitlines()), current_app.config.get('DIGICERT_CIS_INTERMEDIATE'), data['id']
def revoke_certificate(self, certificate, comments):

4
lemur/plugins/lemur_digicert/tests/test_digicert.py
View File

@ -103,7 +103,7 @@ def test_map_cis_fields(app):
'signature_hash': 'sha256',
'organization': {'name': 'Example, Inc.', 'units': ['Example Org']},
'validity': {
'valid_to': arrow.get(2017, 5, 7).format('YYYY-MM-DD')
'valid_to': arrow.get(2017, 5, 7).format('YYYY-MM-DDTHH:MM') + 'Z'
},
'profile_name': None
}
@ -132,7 +132,7 @@ def test_map_cis_fields(app):
'signature_hash': 'sha256',
'organization': {'name': 'Example, Inc.', 'units': ['Example Org']},
'validity': {
'valid_to': arrow.get(2018, 11, 3).format('YYYY-MM-DD')
'valid_to': arrow.get(2018, 11, 3).format('YYYY-MM-DDTHH:MM') + 'Z'
},
'profile_name': None
}

1
lemur/plugins/lemur_statsd/docs/requirements.txt Normal file
View File

@ -0,0 +1 @@
datadog==0.14.0

4
lemur/plugins/lemur_statsd/lemur_statsd/__init__.py Normal file
View File

@ -0,0 +1,4 @@
try:
VERSION = __import__('pkg_resources').get_distribution(__name__).version
except Exception as e:
VERSION = 'Unknown'

45
lemur/plugins/lemur_statsd/lemur_statsd/plugin.py Normal file
View File

@ -0,0 +1,45 @@
import lemur_statsd as plug
from flask import current_app
from lemur.plugins.bases.metric import MetricPlugin
from datadog import DogStatsd
class StatsdMetricPlugin(MetricPlugin):
title = 'Statsd'
slug = 'statsd-metrics'
description = 'Adds support for sending metrics to Statsd'
version = plug.VERSION
def __init__(self):
host = current_app.config.get('STATSD_HOST')
port = current_app.config.get('STATSD_PORT')
prefix = current_app.config.get('STATSD_PREFIX')
self.statsd = DogStatsd(host=host, port=port, namespace=prefix)
def submit(self, metric_name, metric_type, metric_value, metric_tags=None, options=None):
valid_types = ['COUNTER', 'GAUGE', 'TIMER']
tags = []
if metric_type.upper() not in valid_types:
raise Exception(
"Invalid Metric Type for Statsd, '{metric}' choose from: {options}".format(
metric=metric_type, options=','.join(valid_types)
)
)
if metric_tags:
if not isinstance(metric_tags, dict):
raise Exception("Invalid Metric Tags for Statsd: Tags must be in dict format")
else:
tags = map(lambda e: "{0}:{1}".format(*e), metric_tags.items())
if metric_type.upper() == 'COUNTER':
self.statsd.increment(metric_name, metric_value, tags)
elif metric_type.upper() == 'GAUGE':
self.statsd.gauge(metric_name, metric_value, tags)
elif metric_type.upper() == 'TIMER':
self.statsd.timing(metric_name, metric_value, tags)
return

24
lemur/plugins/lemur_statsd/setup.py Normal file
View File

@ -0,0 +1,24 @@
"""Basic package information"""
from __future__ import absolute_import
from setuptools import setup, find_packages
install_requires = [
'lemur',
'datadog'
]
setup(
name='lemur_statsd',
version='1.0.0',
author='Cloudflare Security Engineering',
author_email='',
include_package_data=True,
packages=find_packages(),
zip_safe=False,
install_requires=install_requires,
entry_points={
'lemur.plugins': [
'statsd = lemur_statsd.plugin:StatsdMetricPlugin',
]
}
)

9
lemur/policies/service.py
View File

@ -18,6 +18,15 @@ def get(policy_id):
return database.get(RotationPolicy, policy_id)
def get_by_name(policy_name):
"""
Retrieves policy by its name.
:param policy_name:
:return:
"""
return database.get_all(RotationPolicy, policy_name, field='name').all()
def delete(policy_id):
"""
Delete a rotation policy.

3
lemur/static/app/angular/authorities/authority/options.tpl.html
View File

@ -20,7 +20,8 @@
Key Type
</label>
<div class="col-sm-10">
<select class="form-control" ng-model="authority.keyType" ng-options="option for option in ['RSA2048', 'RSA4096']" ng-init="authority.keyType = 'RSA2048'"></select>
<select class="form-control" ng-model="authority.keyType" ng-options="option for option in ['RSA2048', 'RSA4096', 'ECCPRIME192V1', 'ECCPRIME256V1', 'ECCSECP192R1', 'ECCSECP224R1', 'ECCSECP256R1', 'ECCSECP384R1', 'ECCSECP521R1', 'ECCSECP256K1',
'ECCSECT163K1', 'ECCSECT233K1', 'ECCSECT283K1', 'ECCSECT409K1', 'ECCSECT571K1', 'ECCSECT163R2', 'ECCSECT233R1', 'ECCSECT283R1', 'ECCSECT409R1', 'ECCSECT571R2']" ng-init="authority.keyType = 'RSA2048'"></select>
</div>
</div>
<div ng-show="authority.sensitivity == 'high'" class="form-group">

5
lemur/static/app/angular/certificates/certificate/options.tpl.html
View File

@ -32,7 +32,10 @@
</label>
<div class="col-sm-10">
<select class="form-control" ng-model="certificate.keyType"
ng-options="option for option in ['RSA2048', 'RSA4096']"
ng-options="option for option in ['RSA2048', 'RSA4096', 'ECCPRIME192V1', 'ECCPRIME256V1', 'ECCSECP192R1',
'ECCSECP224R1', 'ECCSECP256R1', 'ECCSECP384R1', 'ECCSECP521R1', 'ECCSECP256K1',
'ECCSECT163K1', 'ECCSECT233K1', 'ECCSECT283K1', 'ECCSECT409K1', 'ECCSECT571K1',
'ECCSECT163R2', 'ECCSECT233R1', 'ECCSECT283R1', 'ECCSECT409R1', 'ECCSECT571R2']"
ng-init="certificate.keyType = 'RSA2048'"></select>
</div>
</div>

8
lemur/tests/test_certificates.py
View File

@ -717,3 +717,11 @@ def test_certificates_upload_patch(client, token, status):
def test_sensitive_sort(client):
resp = client.get(api.url_for(CertificatesList) + '?sortBy=private_key&sortDir=asc', headers=VALID_ADMIN_HEADER_TOKEN)
assert "'private_key' is not sortable or filterable" in resp.json['message']
def test_boolean_filter(client):
resp = client.get(api.url_for(CertificatesList) + '?filter=notify;true', headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200
# Also don't crash with invalid input (we currently treat that as false)
resp = client.get(api.url_for(CertificatesList) + '?filter=notify;whatisthis', headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200

20
lemur/tests/test_utils.py
View File

@ -6,9 +6,27 @@ def test_generate_private_key():
assert generate_private_key('RSA2048')
assert generate_private_key('RSA4096')
assert generate_private_key('ECCPRIME192V1')
assert generate_private_key('ECCPRIME256V1')
assert generate_private_key('ECCSECP192R1')
assert generate_private_key('ECCSECP224R1')
assert generate_private_key('ECCSECP256R1')
assert generate_private_key('ECCSECP384R1')
assert generate_private_key('ECCSECP521R1')
assert generate_private_key('ECCSECP256K1')
assert generate_private_key('ECCSECT163K1')
assert generate_private_key('ECCSECT233K1')
assert generate_private_key('ECCSECT283K1')
assert generate_private_key('ECCSECT409K1')
assert generate_private_key('ECCSECT571K1')
assert generate_private_key('ECCSECT163R2')
assert generate_private_key('ECCSECT233R1')
assert generate_private_key('ECCSECT283R1')
assert generate_private_key('ECCSECT409R1')
assert generate_private_key('ECCSECT571R2')
with pytest.raises(Exception):
generate_private_key('ECC')
generate_private_key('LEMUR')
def test_get_authority_key():