Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_ssl_config(celery_check=False):
"""
Return the ssl config based on the configuration specified in the
`app.yaml` config file.
:param celery_check : Return the proper results ssl setting when configuring celery
"""
results_backend = ""
try:
results_backend = CONFIG.results_backend.url.split(":")[0]
except AttributeError:
pass
try:
results_backend = CONFIG.results_backend.name.lower()
except AttributeError:
pass
if results_backend not in BACKENDS:
return False
try:
certs_path = CONFIG.celery.certs
except AttributeError:
certs_path = None
def get_ssl_config(celery_check=False):
"""
Return the ssl config based on the configuration specified in the
`app.yaml` config file.
:param celery_check : Return the proper results ssl setting when configuring celery
"""
results_backend = ""
try:
results_backend = CONFIG.results_backend.url.split(":")[0]
except AttributeError:
pass
try:
results_backend = CONFIG.results_backend.name.lower()
except AttributeError:
pass
if results_backend not in BACKENDS:
return False
try:
certs_path = CONFIG.celery.certs
except AttributeError:
certs_path = None
results_backend_ssl = get_ssl_entries(
"Results Backend", results_backend, CONFIG.results_backend, certs_path
)
if results_backend == "rediss":
def get_redis(certs_path=None, include_password=True, ssl=False):
"""
Return the redis or rediss specific connection
:param certs_path : The path for ssl certificates and passwords
:param include_password : Format the connection for ouput by setting this True
:param ssl : Flag to use rediss output
"""
server = CONFIG.results_backend.server
password_file = ""
urlbase = "rediss" if ssl else "redis"
try:
port = CONFIG.results_backend.port
except (KeyError, AttributeError):
port = 6379
LOG.debug(f"Results backend: redis using default port = {port}")
try:
db_num = CONFIG.results_backend.db_num
except (KeyError, AttributeError):
db_num = 0
LOG.debug(f"Results backend: redis using default db_num = {db_num}")
def _get_key_path():
""" Loads the redis encryption key path from the file described in config."""
try:
key_filepath = CONFIG.results_backend.encryption_key
except AttributeError:
key_filepath = "~/.merlin/encrypt_data_key"
try:
key_filepath = os.path.abspath(os.path.expanduser(key_filepath))
except KeyError:
raise ValueError("Error! No password provided for RabbitMQ")
return key_filepath
try:
results_backend = CONFIG.results_backend.name.lower()
except AttributeError:
pass
if results_backend not in BACKENDS:
return False
try:
certs_path = CONFIG.celery.certs
except AttributeError:
certs_path = None
results_backend_ssl = get_ssl_entries(
"Results Backend", results_backend, CONFIG.results_backend, certs_path
)
if results_backend == "rediss":
if not results_backend_ssl:
results_backend_ssl = True
return results_backend_ssl
if results_backend and "mysql" in results_backend:
if not celery_check:
return results_backend_ssl
return False
def get_mysql(certs_path=None, mysql_certs=None, include_password=True):
"""
Returns the formatted MySQL connection string.
:param certs_path : The path for ssl certificates and passwords
:param mysql_certs : The dict of mysql certificates
:param include_password : Format the connection for ouput by setting this True
"""
dbname = CONFIG.results_backend.dbname
password_file = CONFIG.results_backend.password
server = CONFIG.results_backend.server
# Adding an initial start for printing configurations. This should
# eventually be configured to use a logger. This logic should also
# eventually be decoupled so we can print debug messages similar to our
# Python debugging messages.
LOG.debug(f"Results backend: dbname = {dbname}")
LOG.debug(f"Results backend: password_file = {password_file}")
LOG.debug(f"Results backend: server = {server}")
LOG.debug(f"Results backend: certs_path = {certs_path}")
if not server:
msg = f"Results backend: server {server} does not have a configuration"
raise Exception(msg)
password = get_backend_password(password_file, certs_path=certs_path)
if not server:
msg = f"Results backend: server {server} does not have a configuration"
raise Exception(msg)
password = get_backend_password(password_file, certs_path=certs_path)
if mysql_certs is None:
mysql_certs = MYSQL_CONFIG_FILENAMES
mysql_config = get_mysql_config(certs_path, mysql_certs)
if not mysql_config:
msg = f"The connection information for MySQL could not be set, cannot find:\n {mysql_certs}\ncheck the celery/certs path or set the ssl information in the app.yaml file."
raise Exception(msg)
mysql_config["user"] = CONFIG.results_backend.username
if include_password:
mysql_config["password"] = password
else:
mysql_config["password"] = "******"
mysql_config["server"] = server
return MYSQL_CONNECTION_STRING.format(**mysql_config)
db_num = CONFIG.results_backend.db_num
except (KeyError, AttributeError):
db_num = 0
LOG.debug(f"Results backend: redis using default db_num = {db_num}")
try:
username = CONFIG.results_backend.username
except (KeyError, AttributeError):
username = ""
try:
password_file = CONFIG.results_backend.password
try:
password = get_backend_password(password_file, certs_path=certs_path)
except IOError:
password = CONFIG.results_backend.password
if include_password:
spass = "%s:%s@" % (username, password)
else:
spass = "%s:%s@" % (username, "******")
except (KeyError, AttributeError):
spass = ""
LOG.debug(f"Results backend: redis using default password = {spass}")
LOG.debug(f"Results backend: password_file = {password_file}")
LOG.debug(f"Results backend: server = {server}")
LOG.debug(f"Results backend: certs_path = {certs_path}")
return f"{urlbase}://{spass}{server}:{port}/{db_num}"
def get_mysql(certs_path=None, mysql_certs=None, include_password=True):
"""
Returns the formatted MySQL connection string.
:param certs_path : The path for ssl certificates and passwords
:param mysql_certs : The dict of mysql certificates
:param include_password : Format the connection for ouput by setting this True
"""
dbname = CONFIG.results_backend.dbname
password_file = CONFIG.results_backend.password
server = CONFIG.results_backend.server
# Adding an initial start for printing configurations. This should
# eventually be configured to use a logger. This logic should also
# eventually be decoupled so we can print debug messages similar to our
# Python debugging messages.
LOG.debug(f"Results backend: dbname = {dbname}")
LOG.debug(f"Results backend: password_file = {password_file}")
LOG.debug(f"Results backend: server = {server}")
LOG.debug(f"Results backend: certs_path = {certs_path}")
if not server:
msg = f"Results backend: server {server} does not have a configuration"
raise Exception(msg)
password = get_backend_password(password_file, certs_path=certs_path)