diff --git a/install/generate_secrets.py b/install/generate_secrets.py
index 774c97a93f9bf98c9af469207de3f8c4f02805b4..1909cf171d25e25a10ad6fc4a6fd382d005ae617 100644
--- a/install/generate_secrets.py
+++ b/install/generate_secrets.py
@@ -58,92 +58,102 @@ def main():
def get_templates_dir():
"""Returns directory that contains the Jinja templates used to create app secrets."""
- return os.path.join(os.path.dirname(os.path.realpath(__file__)), 'templates')
+ return os.path.join(os.path.dirname(os.path.realpath(__file__)), "templates")
def create_variables_secret(app_name, variables_filename, env):
"""Checks if a variables secret for app_name already exists, generates it if necessary."""
- variables_filepath = \
- os.path.join(get_templates_dir(), variables_filename)
+ variables_filepath = os.path.join(get_templates_dir(), variables_filename)
if os.path.exists(variables_filepath):
# Check if k8s secret already exists, if not, generate it
- with open(variables_filepath, encoding='UTF-8') as template_file:
+ with open(variables_filepath, encoding="UTF-8") as template_file:
lines = template_file.read()
secret_name, secret_namespace = get_secret_metadata(lines)
new_secret_dict = yaml.safe_load(
- env.from_string(
- lines,
- globals={"app": app_name}
- ).render())
- current_secret_data = get_kubernetes_secret_data(secret_name,
- secret_namespace)
+ env.from_string(lines, globals={"app": app_name}).render()
+ )
+ current_secret_data = get_kubernetes_secret_data(
+ secret_name, secret_namespace
+ )
if current_secret_data is None:
# Create new secret
update_secret = False
- elif current_secret_data.keys() != new_secret_dict['data'].keys():
+ elif current_secret_data.keys() != new_secret_dict["data"].keys():
# Update current secret with new keys
update_secret = True
- print(f"Secret {secret_name} in namespace {secret_namespace}"
- " already exists. Merging...")
+ print(
+ f"Secret {secret_name} in namespace {secret_namespace}"
+ " already exists. Merging..."
+ )
# Merge dicts. Values from current_secret_data take precedence
- new_secret_dict['data'] |= current_secret_data
+ new_secret_dict["data"] |= current_secret_data
else:
# Do Nothing
- print(f"Secret {secret_name} in namespace {secret_namespace}"
- " is already in a good state, doing nothing.")
+ print(
+ f"Secret {secret_name} in namespace {secret_namespace}"
+ " is already in a good state, doing nothing."
+ )
return
- print(f"Storing secret {secret_name} in namespace"
- f" {secret_namespace} in cluster.")
- store_kubernetes_secret(new_secret_dict, secret_namespace,
- update=update_secret)
+ print(
+ f"Storing secret {secret_name} in namespace"
+ f" {secret_namespace} in cluster."
+ )
+ store_kubernetes_secret(
+ new_secret_dict, secret_namespace, update=update_secret
+ )
else:
print(
- f'Template {variables_filename} does not exist, no action needed')
+ f"Template {variables_filename} does not exist, no action needed")
def create_basic_auth_secret(app_name, env):
"""Checks if a basic auth secret for app_name already exists, generates it if necessary."""
- basic_auth_filename = \
- os.path.join(get_templates_dir(),
- f"stackspin-{app_name}-basic-auth.yaml.jinja")
+ basic_auth_filename = os.path.join(
+ get_templates_dir(), f"stackspin-{app_name}-basic-auth.yaml.jinja"
+ )
if os.path.exists(basic_auth_filename):
- with open(basic_auth_filename, encoding='UTF-8') as template_file:
+ with open(basic_auth_filename, encoding="UTF-8") as template_file:
lines = template_file.read()
secret_name, secret_namespace = get_secret_metadata(lines)
if get_kubernetes_secret_data(secret_name, secret_namespace) is None:
- basic_auth_username = 'admin'
+ basic_auth_username = "admin"
basic_auth_password = generate_password(32)
basic_auth_htpasswd = gen_htpasswd(
- basic_auth_username,
- basic_auth_password)
- print(f"Adding secret {secret_name} in namespace"
- f" {secret_namespace} to cluster.")
+ basic_auth_username, basic_auth_password
+ )
+ print(
+ f"Adding secret {secret_name} in namespace"
+ f" {secret_namespace} to cluster."
+ )
template = env.from_string(
lines,
globals={
- 'pass': basic_auth_password,
- 'htpasswd': basic_auth_htpasswd
- })
+ "pass": basic_auth_password,
+ "htpasswd": basic_auth_htpasswd,
+ },
+ )
secret_dict = yaml.safe_load(template.render())
store_kubernetes_secret(secret_dict, secret_namespace)
else:
- print(f"Secret {secret_name} in namespace {secret_namespace}"
- " already exists. Not generating new secrets.")
+ print(
+ f"Secret {secret_name} in namespace {secret_namespace}"
+ " already exists. Not generating new secrets."
+ )
else:
- print(f'File {basic_auth_filename} does not exist, no action needed')
+ print(f"File {basic_auth_filename} does not exist, no action needed")
def get_secret_metadata(yaml_string):
"""Returns secret name and namespace from metadata field in a yaml string."""
secret_dict = yaml.safe_load(yaml_string)
- secret_name = secret_dict['metadata']['name']
+ secret_name = secret_dict["metadata"]["name"]
# default namespace is flux-system, but other namespace can be
# provided in secret metadata
- if 'namespace' in secret_dict['metadata']:
- secret_namespace = secret_dict['metadata']['namespace']
+ if "namespace" in secret_dict["metadata"]:
+ secret_namespace = secret_dict["metadata"]["namespace"]
else:
- secret_namespace = 'flux-system'
+ secret_namespace = "flux-system"
return secret_name, secret_namespace
@@ -168,9 +178,8 @@ def store_kubernetes_secret(secret_dict, namespace, update=False):
else:
verb = "created"
api_response = create_from_yaml(
- apiclient,
- yaml_objects=[secret_dict],
- namespace=namespace)
+ apiclient, yaml_objects=[secret_dict], namespace=namespace
+ )
print(f"Secret {verb} with api response: {api_response}")
@@ -178,16 +187,16 @@ def patch_kubernetes_secret(secret_dict, namespace):
"""Patches secret in the cluster with new data."""
apiclient = api_client.ApiClient()
api_instance = client.CoreV1Api(apiclient)
- name = secret_dict['metadata']['name']
+ name = secret_dict["metadata"]["name"]
body = {}
- body['data'] = secret_dict['data']
+ body["data"] = secret_dict["data"]
return api_instance.patch_namespaced_secret(name, namespace, body)
def generate_password(length):
"""Generates a password of "length" characters."""
length = int(length)
- password = ''.join((secrets.choice(string.ascii_letters)
+ password = "".join((secrets.choice(string.ascii_letters)
for i in range(length)))
return password