From 8dc8b8c096b2fd1fe99d67474960e8e4b8f846a9 Mon Sep 17 00:00:00 2001
From: Varac <varac@varac.net>
Date: Thu, 17 Feb 2022 13:56:25 +0100
Subject: [PATCH] Lint generate_secrets.py
---
install/generate_secrets.py | 81 ++++++++++++++++++++-----------------
1 file changed, 44 insertions(+), 37 deletions(-)
diff --git a/install/generate_secrets.py b/install/generate_secrets.py
index 3640fd75d..774c97a93 100644
--- a/install/generate_secrets.py
+++ b/install/generate_secrets.py
@@ -1,5 +1,4 @@
-"""
-Generates Kubernetes secrets based on a provided app name.
+"""Generates Kubernetes secrets based on a provided app name.
If the `templates` directory contains a secret called `stackspin-{app}-variables`, it
will check if that secret already exists in the cluster, and if not: generate
@@ -19,9 +18,9 @@ import string
import sys
import jinja2
-import jinja2_base64_filters # pylint: disable=unused-import
import yaml
from kubernetes import client, config
+from kubernetes.client import api_client
from kubernetes.client.exceptions import ApiException
from kubernetes.utils import create_from_yaml
@@ -34,10 +33,12 @@ APPS_WITHOUT_OAUTH = [
"alertmanager",
]
+
def main():
- """Run everything"""
+ """Run everything."""
# Add jinja filters we want to use
- env = jinja2.Environment(extensions=["jinja2_base64_filters.Base64Filters"])
+ env = jinja2.Environment(
+ extensions=["jinja2_base64_filters.Base64Filters"])
env.filters["generate_password"] = generate_password
if len(sys.argv) < 2:
@@ -46,26 +47,27 @@ def main():
app_name = sys.argv[1]
# Create app variables secret
- create_variables_secret(app_name, f"stackspin-{app_name}-variables.yaml.jinja", env)
+ create_variables_secret(
+ app_name, f"stackspin-{app_name}-variables.yaml.jinja", env)
# Create a secret that contains the oauth variables for Hydra Maester
if app_name not in APPS_WITHOUT_OAUTH:
- create_variables_secret(app_name, "stackspin-oauth-variables.yaml.jinja", env)
+ create_variables_secret(
+ app_name, "stackspin-oauth-variables.yaml.jinja", env)
create_basic_auth_secret(app_name, env)
def get_templates_dir():
- """Returns directory that contains the Jinja templates used to create app
- secrets"""
+ """Returns directory that contains the Jinja templates used to create app secrets."""
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"""
+ """Checks if a variables secret for app_name already exists, generates it if necessary."""
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) 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(
@@ -74,7 +76,7 @@ def create_variables_secret(app_name, variables_filename, env):
globals={"app": app_name}
).render())
current_secret_data = get_kubernetes_secret_data(secret_name,
- secret_namespace)
+ secret_namespace)
if current_secret_data is None:
# Create new secret
update_secret = False
@@ -95,15 +97,17 @@ def create_variables_secret(app_name, variables_filename, env):
store_kubernetes_secret(new_secret_dict, secret_namespace,
update=update_secret)
else:
- print(f'Template {variables_filename} does not exist, no action needed')
+ print(
+ 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"""
+ """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")
+ 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) 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)
@@ -116,11 +120,11 @@ def create_basic_auth_secret(app_name, env):
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
- })
+ lines,
+ globals={
+ 'pass': basic_auth_password,
+ 'htpasswd': basic_auth_htpasswd
+ })
secret_dict = yaml.safe_load(template.render())
store_kubernetes_secret(secret_dict, secret_namespace)
else:
@@ -129,8 +133,9 @@ def create_basic_auth_secret(app_name, env):
else:
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"""
+ """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']
# default namespace is flux-system, but other namespace can be
@@ -153,42 +158,44 @@ def get_kubernetes_secret_data(secret_name, namespace):
return None
return secret
+
def store_kubernetes_secret(secret_dict, namespace, update=False):
- """Stores either a new secret in the cluster, or updates an existing one"""
- api_client = client.api_client.ApiClient()
+ """Stores either a new secret in the cluster, or updates an existing one."""
+ apiclient = api_client.ApiClient()
if update:
verb = "updated"
api_response = patch_kubernetes_secret(secret_dict, namespace)
else:
verb = "created"
api_response = create_from_yaml(
- api_client,
- yaml_objects=[secret_dict],
- namespace=namespace)
+ apiclient,
+ yaml_objects=[secret_dict],
+ namespace=namespace)
print(f"Secret {verb} with api response: {api_response}")
+
def patch_kubernetes_secret(secret_dict, namespace):
- """Patches secret in the cluster with new data"""
- api_client = client.api_client.ApiClient()
- api_instance = client.CoreV1Api(api_client)
+ """Patches secret in the cluster with new data."""
+ apiclient = api_client.ApiClient()
+ api_instance = client.CoreV1Api(apiclient)
name = secret_dict['metadata']['name']
body = {}
body['data'] = secret_dict['data']
return api_instance.patch_namespaced_secret(name, namespace, body)
+
def generate_password(length):
- """Generates a password of "length" characters"""
+ """Generates a password of "length" characters."""
length = int(length)
- password = ''.join((secrets.choice(string.ascii_letters) for i in range(length)))
+ password = ''.join((secrets.choice(string.ascii_letters)
+ for i in range(length)))
return password
def gen_htpasswd(user, password):
- """generate htpasswd entry for user with password"""
- return "{}:{}".format(user, crypt.crypt(
- password, crypt.mksalt(crypt.METHOD_SHA512)
- ),
- )
+ """Generate htpasswd entry for user with password."""
+ return f"{user}:{crypt.crypt(password, crypt.mksalt(crypt.METHOD_SHA512))}"
+
if __name__ == "__main__":
config.load_kube_config()
--
GitLab