def rotate()

in lemur/certificates/cli.py [0:0]
91 lines of code
10 McCabe index (conditional complexity)

def rotate(endpoint_name, new_certificate_name, old_certificate_name, message, commit):
    """
    Rotates an endpoint and reissues it if it has not already been replaced. If it has
    been replaced, will use the replacement certificate for the rotation.
    """
    if commit:
        print("[!] Running in COMMIT mode.")

    print("[+] Starting endpoint rotation.")

    status = FAILURE_METRIC_STATUS

    log_data = {
        "function": f"{__name__}.{sys._getframe().f_code.co_name}",
    }

    try:
        old_cert = validate_certificate(old_certificate_name)
        new_cert = validate_certificate(new_certificate_name)
        endpoint = validate_endpoint(endpoint_name)

        if endpoint and new_cert:
            print(
                f"[+] Rotating endpoint: {endpoint.name} to certificate {new_cert.name}"
            )
            log_data["message"] = "Rotating endpoint"
            log_data["endpoint"] = endpoint.dnsname
            log_data["certificate"] = new_cert.name
            request_rotation(endpoint, new_cert, message, commit)
            current_app.logger.info(log_data)

        elif old_cert and new_cert:
            print(f"[+] Rotating all endpoints from {old_cert.name} to {new_cert.name}")

            log_data["message"] = "Rotating all endpoints"
            log_data["certificate"] = new_cert.name
            log_data["certificate_old"] = old_cert.name
            log_data["message"] = "Rotating endpoint from old to new cert"
            for endpoint in old_cert.endpoints:
                print(f"[+] Rotating {endpoint.name}")
                log_data["endpoint"] = endpoint.dnsname
                request_rotation(endpoint, new_cert, message, commit)
                current_app.logger.info(log_data)

        else:
            print("[+] Rotating all endpoints that have new certificates available")
            log_data["message"] = "Rotating all endpoints that have new certificates available"
            for endpoint in endpoint_service.get_all_pending_rotation():
                log_data["endpoint"] = endpoint.dnsname
                if len(endpoint.certificate.replaced) == 1:
                    print(
                        f"[+] Rotating {endpoint.name} to {endpoint.certificate.replaced[0].name}"
                    )
                    log_data["certificate"] = endpoint.certificate.replaced[0].name
                    request_rotation(
                        endpoint, endpoint.certificate.replaced[0], message, commit
                    )
                    current_app.logger.info(log_data)

                else:
                    log_data["message"] = "Failed to rotate endpoint due to Multiple replacement certificates found"
                    print(log_data)
                    metrics.send(
                        "endpoint_rotation",
                        "counter",
                        1,
                        metric_tags={
                            "status": FAILURE_METRIC_STATUS,
                            "old_certificate_name": str(old_cert),
                            "new_certificate_name": str(
                                endpoint.certificate.replaced[0].name
                            ),
                            "endpoint_name": str(endpoint.name),
                            "message": str(message),
                        },
                    )
                    print(
                        f"[!] Failed to rotate endpoint {endpoint.name} reason: "
                        "Multiple replacement certificates found."
                    )

        status = SUCCESS_METRIC_STATUS
        print("[+] Done!")

    except Exception as e:
        sentry.captureException(
            extra={
                "old_certificate_name": str(old_certificate_name),
                "new_certificate_name": str(new_certificate_name),
                "endpoint_name": str(endpoint_name),
                "message": str(message),
            }
        )

    metrics.send(
        "endpoint_rotation_job",
        "counter",
        1,
        metric_tags={
            "status": status,
            "old_certificate_name": str(old_certificate_name),
            "new_certificate_name": str(new_certificate_name),
            "endpoint_name": str(endpoint_name),
            "message": str(message),
            "endpoint": str(globals().get("endpoint")),
        },
    )