def create()

in lemur/authorities/service.py [0:0]

• 30 lines of code
• 11 McCabe index (conditional complexity)

def create(**kwargs):
    """
    Creates a new authority.
    """
    ca_name = kwargs.get("name")
    if get_by_name(ca_name):
        raise Exception(f"Authority with name {ca_name} already exists")
    if role_service.get_by_name(f"{ca_name}_admin") or role_service.get_by_name(f"{ca_name}_operator"):
        raise Exception(f"Admin and/or operator roles for authority {ca_name} already exist")

    body, private_key, chain, roles = mint(**kwargs)

    kwargs["body"] = body
    kwargs["private_key"] = private_key
    kwargs["chain"] = chain

    if not kwargs.get("roles"):
        kwargs["roles"] = []
    kwargs["roles"] += [role for role in roles if role not in kwargs["roles"]]

    cert = upload(**kwargs)
    kwargs["authority_certificate"] = cert
    if kwargs.get("plugin", {}).get("plugin_options", []):
        # encrypt the private key before persisting in DB
        for option in kwargs.get("plugin").get("plugin_options"):
            if option["name"] == "acme_private_key" and option["value"]:
                option["value"] = data_encrypt(option["value"])
        kwargs["options"] = json.dumps(kwargs["plugin"]["plugin_options"])

    authority = Authority(**kwargs)
    authority = database.create(authority)
    kwargs["creator"].authorities.append(authority)

    log_service.audit_log("create_authority", ca_name, "Created new authority")

    issuer = kwargs["plugin"]["plugin_object"]
    current_app.logger.warning(f"Created new authority {ca_name} with issuer {issuer.title}")

    metrics.send(
        "authority_created", "counter", 1, metric_tags=dict(owner=authority.owner)
    )
    return authority