in lemur/certificates/cli.py [0:0]
def rotate_region(endpoint_name, new_certificate_name, old_certificate_name, message, commit, region):
"""
Rotates an endpoint in a defined region it if it has not already been replaced. If it has
been replaced, will use the replacement certificate for the rotation.
:param old_certificate_name: Name of the certificate you wish to rotate.
:param new_certificate_name: Name of the certificate you wish to rotate to.
:param endpoint_name: Name of the endpoint you wish to rotate.
:param message: Send a rotation notification to the certificates owner.
:param commit: Persist changes.
:param region: Region in which to rotate the endpoint.
#todo: merge this method with rotate()
"""
if commit:
click.echo("[!] Running in COMMIT mode.")
click.echo("[+] Starting endpoint rotation.")
status = FAILURE_METRIC_STATUS
log_data = {
"function": f"{__name__}.{sys._getframe().f_code.co_name}",
"region": region,
}
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:
log_data["endpoint"] = endpoint.dnsname
log_data["certificate"] = new_cert.name
request_rotation_region(endpoint, new_cert, message, commit, log_data, region)
elif old_cert and new_cert:
log_data["certificate"] = new_cert.name
log_data["certificate_old"] = old_cert.name
log_data["message"] = "Rotating endpoint from old to new cert"
click.echo(log_data)
current_app.logger.info(log_data)
for endpoint in old_cert.endpoints:
log_data["endpoint"] = endpoint.dnsname
request_rotation_region(endpoint, new_cert, message, commit, log_data, region)
else:
log_data["message"] = "Rotating all endpoints that have new certificates available"
click.echo(log_data)
current_app.logger.info(log_data)
all_pending_rotation_endpoints = endpoint_service.get_all_pending_rotation()
for endpoint in all_pending_rotation_endpoints:
log_data["endpoint"] = endpoint.dnsname
if region not in endpoint.dnsname:
log_data["message"] = "Skipping rotation, region mismatch"
click.echo(log_data)
current_app.logger.info(log_data)
metrics.send(
"endpoint_rotation_region_skipped",
"counter",
1,
metric_tags={
"region": region,
"new_certificate_name": str(endpoint.certificate.replaced[0].name),
"endpoint_name": str(endpoint.dnsname),
},
)
continue
log_data["certificate"] = endpoint.certificate.replaced[0].name
log_data["message"] = "Rotating all endpoints in region"
if len(endpoint.certificate.replaced) > 1:
log_data["message"] = f"Multiple replacement certificates found, going with the first one out of " \
f"{len(endpoint.certificate.replaced)}"
request_rotation(endpoint, endpoint.certificate.replaced[0], message, commit)
current_app.logger.info(log_data)
metrics.send(
"endpoint_rotation_region",
"counter",
1,
metric_tags={
"status": FAILURE_METRIC_STATUS,
"new_certificate_name": str(log_data["certificate"]),
"endpoint_name": str(endpoint.dnsname),
"message": str(message),
"region": str(region),
},
)
status = SUCCESS_METRIC_STATUS
click.echo("[+] Done!")
except Exception as e:
capture_exception(
extra={
"old_certificate_name": str(old_certificate_name),
"new_certificate_name": str(new_certificate_name),
"endpoint": str(endpoint_name),
"message": str(message),
"region": str(region),
}
)
metrics.send(
"endpoint_rotation_region_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")),
"region": str(region),
},
)