Enhance domains query and sensitive domain checking code; Allow creation of opt-out roles via config

This commit is contained in:
Curtis Castrapel 2019-05-30 10:21:44 -07:00
parent 8bc23f6deb
commit 8b821d0023
4 changed files with 28 additions and 5 deletions

View File

@ -9,6 +9,7 @@
from functools import partial from functools import partial
from collections import namedtuple from collections import namedtuple
from flask import current_app
from flask_principal import Permission, RoleNeed from flask_principal import Permission, RoleNeed
# Permissions # Permissions
@ -21,7 +22,14 @@ CertificateOwnerNeed = partial(CertificateOwner, "role")
class SensitiveDomainPermission(Permission): class SensitiveDomainPermission(Permission):
def __init__(self): def __init__(self):
super(SensitiveDomainPermission, self).__init__(RoleNeed("admin")) needs = [RoleNeed("admin")]
sensitive_domain_roles = current_app.config.get("SENSITIVE_DOMAIN_ROLES", [])
if sensitive_domain_roles:
for role in sensitive_domain_roles:
needs.append(RoleNeed(role))
super(SensitiveDomainPermission, self).__init__(*needs)
class CertificatePermission(Permission): class CertificatePermission(Permission):

View File

@ -16,13 +16,13 @@ from celery.exceptions import SoftTimeLimitExceeded
from flask import current_app from flask import current_app
from lemur.authorities.service import get as get_authority from lemur.authorities.service import get as get_authority
from lemur.destinations import service as destinations_service
from lemur.extensions import metrics, sentry from lemur.extensions import metrics, sentry
from lemur.factory import create_app from lemur.factory import create_app
from lemur.notifications.messaging import send_pending_failure_notification from lemur.notifications.messaging import send_pending_failure_notification
from lemur.pending_certificates import service as pending_certificate_service from lemur.pending_certificates import service as pending_certificate_service
from lemur.plugins.base import plugins from lemur.plugins.base import plugins
from lemur.sources.cli import clean, sync, validate_sources from lemur.sources.cli import clean, sync, validate_sources
from lemur.destinations import service as destinations_service
from lemur.sources.service import add_aws_destination_to_sources from lemur.sources.service import add_aws_destination_to_sources
if current_app: if current_app:

View File

@ -40,7 +40,7 @@ def sensitive_domain(domain):
# Avoid circular import. # Avoid circular import.
from lemur.domains import service as domain_service from lemur.domains import service as domain_service
if any(d.sensitive for d in domain_service.get_by_name(domain)): if domain_service.is_domain_sensitive(domain):
raise ValidationError( raise ValidationError(
"Domain {0} has been marked as sensitive. " "Domain {0} has been marked as sensitive. "
"Contact an administrator to issue the certificate.".format(domain) "Contact an administrator to issue the certificate.".format(domain)

View File

@ -6,10 +6,11 @@
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com> .. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
""" """
from lemur.domains.models import Domain from sqlalchemy import and_
from lemur.certificates.models import Certificate
from lemur import database from lemur import database
from lemur.certificates.models import Certificate
from lemur.domains.models import Domain
def get(domain_id): def get(domain_id):
@ -42,6 +43,20 @@ def get_by_name(name):
return database.get_all(Domain, name, field="name").all() return database.get_all(Domain, name, field="name").all()
def is_domain_sensitive(name):
"""
Return True if domain is marked sensitive
:param name:
:return:
"""
query = database.session_query(Domain)
query = query.filter(and_(Domain.sensitive, Domain.name == name))
return database.find_all(query, Domain, {}).all()
def create(name, sensitive): def create(name, sensitive):
""" """
Create a new domain Create a new domain