Per domain authorization for certificate issuance

docs/administration.rst

@@ -1025,6 +1025,23 @@ For more information about how to use social logins, see: `Satellizer <https://g
 
             USER_MEMBERSHIP_PROVIDER = "<yourmembershippluginslug>"
 
+Authorization Providers
+~~~~~~~~~~~~~~~~~~~~~~~
+
+
+If you are not using a custom authorization provider you do not need to configure any of these options
+
+.. data:: USER_DOMAIN_AUTHORIZATION_PROVIDER
+    :noindex:
+
+        An optional plugin to perform domain level authorization during certificate issuance. Provide plugin slug here.
+        Plugin is used to check if caller is authorized to issue a certificate for a given Common Name and Subject Alternative
+        Name (SAN) of type DNSName. Plugin shall be an implementation of DomainAuthorizationPlugin.
+
+        ::
+
+            USER_DOMAIN_AUTHORIZATION_PROVIDER = "<yourauthorizationpluginslug>"
+
 Metric Providers
 ~~~~~~~~~~~~~~~~
 
@@ -2015,6 +2032,8 @@ get it added.
 Want to create your own extension? See :doc:`../developer/plugins/index` to get started.
 
 
+.. _iam_target:
+
 Identity and Access Management
 ==============================
 
@@ -2041,3 +2060,30 @@ These permissions are applied to the user upon login and refreshed on every requ
 .. seealso::
 
     `Flask-Principal <https://pythonhosted.org/Flask-Principal>`_
+
+To allow integration with external access/membership management tools that may exist in your organization, lemur offers
+below plugins in addition to it's own RBAC implementation.
+
+Membership Plugin
+-----------------
+
+:Authors:
+    Sayali Charhate <[email protected]>
+:Type:
+    User Membership
+:Description:
+    Adds support to learn and validate user membership details from an external service. User memberships are used to
+    create user roles dynamically as described in :ref:`iam_target`. Configure this plugin slug as `USER_MEMBERSHIP_PROVIDER`
+
+Authorization Plugins
+---------------------
+
+:Authors:
+    Sayali Charhate <[email protected]>
+:Type:
+    External Authorization
+:Description:
+    Adds support to implement custom authorization logic that is best suited for your enterprise. Lemur offers `AuthorizationPlugin`
+    and its extended version `DomainAuthorizationPlugin`. One can implement `DomainAuthorizationPlugin` and configure its
+    slug as `USER_DOMAIN_AUTHORIZATION_PROVIDER` to check if caller is authorized to issue a certificate for a given Common
+    Name and Subject Alternative Name (SAN) of type DNSName

lemur/api_keys/models.py

@@ -19,6 +19,7 @@ class ApiKey(db.Model):
     ttl = Column(BigInteger)
     issued_at = Column(BigInteger)
     revoked = Column(Boolean)
+    application_name = Column(String, nullable=True)
 
     def __repr__(self):
         return "ApiKey(name={name}, user_id={user_id}, ttl={ttl}, issued_at={iat}, revoked={revoked})".format(

lemur/auth/service.py

@@ -120,6 +120,8 @@ def decorated_function(*args, **kwargs):
                 expired_time = datetime.fromtimestamp(access_key.issued_at) + timedelta(days=access_key.ttl)
                 if current_time >= expired_time:
                     return dict(message="Token has expired"), 403
+            if access_key.application_name:
+                g.caller_application = access_key.application_name
 
         user = user_service.get(payload["sub"])
 

lemur/certificates/service.py

@@ -28,6 +28,7 @@
 from lemur.constants import SUCCESS_METRIC_STATUS, FAILURE_METRIC_STATUS
 from lemur.destinations.models import Destination
 from lemur.domains.models import Domain
+from lemur.domains.service import is_authorized_for_domain
 from lemur.endpoints import service as endpoint_service
 from lemur.extensions import metrics, signals
 from lemur.notifications.messaging import send_revocation_notification
@@ -1240,3 +1241,19 @@ def is_valid_owner(email):
 
     # expecting owner to be an existing team DL
     return user_membership_provider.does_group_exist(email)
+
+
+def allowed_issuance_for_domain(common_name, extensions):
+    check_permission_for_cn = True if common_name else False
+
+    # authorize issuance for every x509.DNSName SAN
+    if extensions and extensions.get("sub_alt_names"):
+        for san in extensions["sub_alt_names"]["names"]:
+            if isinstance(san, x509.DNSName):
+                if san.value == common_name:
+                    check_permission_for_cn = False
+                is_authorized_for_domain(san.value)
+
+    # lemur UI copies CN as SAN (x509.DNSName). Permission check for CN might already be covered above.
+    if check_permission_for_cn:
+        is_authorized_for_domain(common_name)

lemur/certificates/views.py

@@ -10,6 +10,7 @@
 
 from flask import Blueprint, make_response, jsonify, g, current_app
 from flask_restful import reqparse, Api, inputs
+from lemur.plugins.bases.authorization import UnauthorizedError
 from sentry_sdk import capture_exception
 
 from lemur.common.schema import validate_schema
@@ -511,23 +512,21 @@ def post(self, data=None):
         roles.append(role)
         authority_permission = AuthorityPermission(data["authority"].id, roles)
 
-        if authority_permission.can():
-            data["creator"] = g.user
+        if not authority_permission.can():
+            return dict(message=f"You are not authorized to use the authority: {data['authority'].name}"), 403
+
+        data["creator"] = g.user
+        try:
+            service.allowed_issuance_for_domain(data["common_name"], data["extensions"])
+        except UnauthorizedError as e:
+            return dict(message=str(e)), 403
+        else:
             cert = service.create(**data)
             if isinstance(cert, Certificate):
                 # only log if created, not pending
                 log_service.create(g.user, "create_cert", certificate=cert)
             return cert
 
-        return (
-            dict(
-                message="You are not authorized to use the authority: {0}".format(
-                    data["authority"].name
-                )
-            ),
-            403,
-        )
-
 
 class CertificatesUpload(AuthenticatedResource):
     """ Defines the 'certificates' upload endpoint """

lemur/domains/service.py

@@ -8,9 +8,13 @@
 """
 from sqlalchemy import and_
 
+from flask import current_app, g
+
 from lemur import database
 from lemur.certificates.models import Certificate
 from lemur.domains.models import Domain
+from lemur.plugins.base import plugins
+from lemur.plugins.bases.authorization import UnauthorizedError
 
 
 def get(domain_id):
@@ -57,6 +61,31 @@ def is_domain_sensitive(name):
     return database.find_all(query, Domain, {}).all()
 
 
+def is_authorized_for_domain(name):
+    """
+    If authorization plugin is available, perform the check to see if current user can issue certificate for a given
+    domain.
+    Raises UnauthorizedError if unauthorized.
+    If authorization plugin is not available, it returns without performing any check
+
+    :param name: domain (string) for which authorization check is being done
+    """
+    if current_app.config.get("USER_DOMAIN_AUTHORIZATION_PROVIDER") is None:
+        # nothing to check since USER_DOMAIN_AUTHORIZATION_PROVIDER is not configured
+        return
+
+    user_domain_authorization_provider = plugins.get(current_app.config.get("USER_DOMAIN_AUTHORIZATION_PROVIDER"))
+    caller = g.caller_application if hasattr(g, 'caller_application') else g.user.email
+    # if the caller can be mapped to an application name, use that to perform authorization
+    # this could be true when using API key to call lemur (migration script e2d406ada25c_.py)
+    authorized, error = user_domain_authorization_provider.is_authorized(domain=name, caller=caller)
+
+    if error:
+        raise error
+    if not authorized:
+        raise UnauthorizedError(user=caller, resource=name, action="issue_certificate")
+
+
 def create(name, sensitive):
     """
     Create a new domain

lemur/migrations/versions/e2d406ada25c_.py

@@ -0,0 +1,22 @@
+"""Add column application_name to api_keys table. Null values are allowed.
+
+Revision ID: e2d406ada25c
+Revises: 189e5fda5bf8
+Create Date: 2021-11-24 14:48:18.747487
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'e2d406ada25c'
+down_revision = '189e5fda5bf8'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.add_column("api_keys", sa.Column("application_name", sa.String(256), nullable=True))
+
+
+def downgrade():
+    op.drop_column("api_keys", "application_name")

lemur/plugins/bases/__init__.py

@@ -4,4 +4,6 @@
 from .notification import NotificationPlugin, ExpirationNotificationPlugin  # noqa
 from .export import ExportPlugin  # noqa
 from .tls import TLSPlugin  # noqa
-from .membership import MembershipPlugin # noqa
+from .membership import MembershipPlugin  # noqa
+from .authorization import AuthorizationPlugin  # noqa
+from .authorization import DomainAuthorizationPlugin  # noqa

lemur/plugins/bases/authorization.py

@@ -0,0 +1,45 @@
+"""
+.. module: lemur.plugins.bases.authorization
+    :platform: Unix
+    :copyright: (c) 2021 by Netflix Inc., see AUTHORS for more
+    :license: Apache, see LICENSE for more details.
+
+.. moduleauthor:: Sayali Charhate <[email protected]>
+"""
+
+from lemur.exceptions import LemurException
+from lemur.plugins.base import Plugin
+
+
+class AuthorizationPlugin(Plugin):
+    """
+    This is the base class for authorization providers. Check if the caller is authorized to access a resource.
+    """
+    type = "authorization"
+
+    def is_authorized(self, resource, caller):
+        raise NotImplementedError
+
+
+class DomainAuthorizationPlugin(AuthorizationPlugin):
+    """
+    This is the base class for domain authorization providers. Check if the caller can issue certificates for a domain.
+    """
+    type = "domain-authorization"
+
+    def is_authorized(self, domain, caller):
+        raise NotImplementedError
+
+
+class UnauthorizedError(LemurException):
+    """
+    Raised when user is unauthorized to perform an action on the resource
+    """
+    def __init__(self, user, resource, action, details="no additional details"):
+        self.user = user
+        self.resource = resource
+        self.action = action
+        self.details = details
+
+    def __str__(self):
+        return repr(f"{self.user} is not authorized to perform {self.action} on {self.resource}: {self.details}")

lemur/tests/test_certificates.py

@@ -825,8 +825,8 @@ def test_create_basic_csr(client):
         owner="[email protected]",
         key_type="RSA2048",
         extensions=dict(
-            names=dict(
-                sub_alt_names=x509.SubjectAlternativeName(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
                     [
                         x509.DNSName("test.example.com"),
                         x509.DNSName("test2.example.com"),
@@ -1578,3 +1578,86 @@ def start_server():
     daemon.setDaemon(True)  # Set as a daemon so it will be killed once the main thread is dead.
     daemon.start()
     return daemon
+
+
+def mocked_is_authorized_for_domain(name):
+    domain_in_error = "fail.lemur.com"
+    if name == domain_in_error:
+        raise UnauthorizedError(user="dummy_user", resource=domain_in_error, action="issue_certificate",
+                                details="unit test, mocked failure")
+
+
+@pytest.mark.parametrize(
+    "common_name, extensions, expected_error, authz_check_count",
+    [
+        ("fail.lemur.com", None, True, 1),
+        ("fail.lemur.com", dict(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
+                    [
+                        x509.DNSName("test.example.com"),
+                        x509.DNSName("test2.example.com"),
+                    ]
+                )
+            )
+        ), True, 3),  # CN is checked after SAN
+        ("test.example.com", dict(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
+                    [
+                        x509.DNSName("fail.lemur.com"),
+                        x509.DNSName("test2.example.com"),
+                    ]
+                )
+            )
+        ), True, 1),
+        (None, dict(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
+                    [
+                        x509.DNSName("fail.lemur.com"),
+                        x509.DNSName("test2.example.com"),
+                    ]
+                )
+            )
+        ), True, 1),
+        ("pass.lemur.com", None, False, 1),
+        ("pass.lemur.com", dict(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
+                    [
+                        x509.DNSName("test.example.com"),
+                        x509.DNSName("test2.example.com"),
+                    ]
+                )
+            )
+        ), False, 3),
+        ("pass.lemur.com", dict(
+            sub_alt_names=dict(
+                names=x509.SubjectAlternativeName(
+                    [
+                        x509.DNSName("test.example.com"),
+                        x509.DNSName("pass.lemur.com"),
+                    ]
+                )
+            )
+        ), False, 2),  # CN repeated in SAN
+    ],
+)
+def test_allowed_issuance_for_domain(common_name, extensions, expected_error, authz_check_count):
+    from lemur.certificates.service import allowed_issuance_for_domain
+
+    with patch(
+        'lemur.certificates.service.is_authorized_for_domain', side_effect=mocked_is_authorized_for_domain
+    ) as wrapper:
+        try:
+            allowed_issuance_for_domain(common_name, extensions)
+            if expected_error:
+                assert False, f"UnauthorizedError did not occur, input: CN({common_name}), SAN({extensions})"
+        except UnauthorizedError as e:
+            if expected_error:
+                pass
+            else:
+                assert False, f"UnauthorizedError occured, input: CN({common_name}), SAN({extensions})"
+
+        assert wrapper.call_count == authz_check_count