Skip to content
Snippets Groups Projects
Unverified Commit 70294868 authored by Max Adamo's avatar Max Adamo
Browse files

refactor: remove unused imports and clean up code formatting

parent 9cab2c06
No related branches found
No related tags found
No related merge requests found
"""Wile Coyote is a Python library for creating and managing virtual environments."""
import pkgutil
__version__ = "0.8.3"
""" Common package for Wile Coyote project. """
import pkgutil
__all__ = []
for loader, module_name, is_pkg in pkgutil.walk_packages(__path__):
__all__.append(module_name)
_module = loader.find_module(module_name).load_module(module_name)
globals()[module_name] = _module
"""Module providing a function to combine certificate, CA and private key"""
import os
import wile_coyote.tools
import configparser
import wile_coyote.common.constants
def keys(cert_path, provider, key_path, outpath):
......@@ -13,7 +14,9 @@ def keys(cert_path, provider, key_path, outpath):
else:
raise NotImplementedError("OS not supported")
ca_file = wile_coyote.tools.PROVIDERS_CA[provider]
config = configparser.RawConfigParser()
config.read(wile_coyote.common.constants.CRED_CONF, encoding="utf8")
ca_file = config.get("ca", provider)
ca_path = os.path.join(ssl_dir, ca_file)
file_names = [cert_path, ca_path, key_path]
......
......@@ -14,21 +14,21 @@ for loader, module_name, is_pkg in pkgutil.walk_packages(__path__):
config = configparser.RawConfigParser()
config.read(wile_coyote.common.constants.CRED_CONF, encoding="utf8")
ATOMIC_MSG = 'Atomic Check:'
VAULT_HOST = config.get('acme', 'vault_host')
REDIS_HOST = config.get('acme', 'redis_host')
REDIS_TOKEN = config.get('acme', 'redis_token')
CONSUL_SERVERS = config.get('acme', 'consul_servers')
CONSUL_TOKEN = config.get('acme', 'consul_token')
WEB_BASE = config.get('acme', 'web_base')
ACME_PROVIDERS = config.get('acme', 'acme_providers')
PROVIDERS_CA = l_eval(config.get('acme', 'providers_ca'))
ATOMIC_MSG = "Atomic Check:"
VAULT_HOST = config.get("acme", "vault_host")
REDIS_HOST = config.get("acme", "redis_host")
REDIS_TOKEN = config.get("acme", "redis_token")
CONSUL_SERVERS = config.get("acme", "consul_servers")
CONSUL_TOKEN = config.get("acme", "consul_token")
WEB_BASE = config.get("acme", "web_base")
ACME_PROVIDERS = config.get("acme", "acme_providers")
PROVIDERS_CA = l_eval(config.get("acme", "providers_ca"))
# these parameters only work in test
if 'unit-test' in config.sections():
VAULT_ROOT_TOKEN = config.get('acme', 'vault_token_root')
MOUNT_POINTS_V1 = l_eval(config.get('unit-test', 'mount_points_v1'))
MOUNT_POINTS_V2 = l_eval(config.get('unit-test', 'mount_points_v2'))
REDIS_KEYS = sorted(l_eval(config.get('unit-test', 'redis_keys')))
VAULT_KEYS = sorted(l_eval(config.get('unit-test', 'vault_keys')))
CONSUL_KEYS = sorted(l_eval(config.get('unit-test', 'consul_keys')))
if "unit-test" in config.sections():
VAULT_ROOT_TOKEN = config.get("acme", "vault_token_root")
MOUNT_POINTS_V1 = l_eval(config.get("unit-test", "mount_points_v1"))
MOUNT_POINTS_V2 = l_eval(config.get("unit-test", "mount_points_v2"))
REDIS_KEYS = sorted(l_eval(config.get("unit-test", "redis_keys")))
VAULT_KEYS = sorted(l_eval(config.get("unit-test", "vault_keys")))
CONSUL_KEYS = sorted(l_eval(config.get("unit-test", "consul_keys")))
......@@ -13,9 +13,9 @@ import wile_coyote.tools
def delete(key, leader):
""" Download key from consul """
"""Download key from consul"""
token = wile_coyote.tools.CONSUL_TOKEN
c_client = Consul(host=leader, port='443', token=token, scheme='https')
c_client = Consul(host=leader, port="443", token=token, scheme="https")
try:
c_client.kv.delete(key)
......@@ -24,42 +24,42 @@ def delete(key, leader):
def get(keyname, log_file, leader):
""" Get key from consul """
"""Get key from consul"""
token = wile_coyote.tools.CONSUL_TOKEN
c_client = Consul(host=leader, port='443', token=token, scheme='https')
c_client = Consul(host=leader, port="443", token=token, scheme="https")
try:
_, data = c_client.kv.get(keyname)
except Exception as err: # pylint: disable=w0703
log.handler(
f'could not connect to Consul {leader}', log_file, True)
log.handler(f"could not connect to Consul {leader}", log_file, True)
return err
if data:
return data['Value'].decode('utf-8')
return data["Value"].decode("utf-8")
return 'BOFH'
return "BOFH"
def get_leader(log_file):
""" detect consul leader to ensure that we
are picking a node which is up & running """
"""detect consul leader to ensure that we
are picking a node which is up & running"""
consul_servers = wile_coyote.tools.CONSUL_SERVERS
ctx = ssl.create_default_context()
server_list = consul_servers.replace(' ', '').split(',')
server_list = consul_servers.replace(" ", "").split(",")
leader = None
for server in server_list:
try:
_leader = urllib.request.urlopen(
f'https://{server}/v1/status/leader', context=ctx)
f"https://{server}/v1/status/leader", context=ctx
)
except Exception: # pylint: disable=W0703
print(111)
else:
leader = _leader.read().decode('utf-8').split(':')[0].replace('"', '')
leader = _leader.read().decode("utf-8").split(":")[0].replace('"', "")
break
if not leader:
log.handler('could not find a Consul leader', log_file, True)
log.handler("could not find a Consul leader", log_file, True)
log.handler(wile_coyote.common.constants.GIVEUP, log_file)
os._exit(1)
......@@ -67,31 +67,33 @@ def get_leader(log_file):
def put(key, value, log_file, leader):
""" Upload key to consul """
"""Upload key to consul"""
token = wile_coyote.tools.CONSUL_TOKEN
c_client = Consul(host=leader, port='443', token=token, scheme='https')
c_client = Consul(host=leader, port="443", token=token, scheme="https")
try:
c_client.kv.put(key, value)
except Exception as err: # pylint: disable=W0703
log.handler(f'could not write key {key} to {leader}: {err}',
log_file, True)
log.handler(f"could not write key {key} to {leader}: {err}", log_file, True)
os._exit(1)
def test(log_file, leader):
""" Check consul connection """
"""Check consul connection"""
atomic_msg = wile_coyote.tools.ATOMIC_MSG
token = wile_coyote.tools.CONSUL_TOKEN
rnd = ''.join([random.choice(string.ascii_letters) for _ in range(8)])
c_client = Consul(host=leader, port='443', token=token, scheme='https')
rnd = "".join([random.choice(string.ascii_letters) for _ in range(8)])
c_client = Consul(host=leader, port="443", token=token, scheme="https")
try:
_ = c_client.kv.put('do_not_delete', rnd)
_ = c_client.kv.put("do_not_delete", rnd)
except Exception as err: # pylint: disable=w0703
log.handler(f'{atomic_msg} could not write test key to {leader}: {err}',
log_file, True)
log.handler(
f"{atomic_msg} could not write test key to {leader}: {err}", log_file, True
)
log.handler(wile_coyote.common.constants.GIVEUP, log_file)
os._exit(1)
else:
log.handler(f'{atomic_msg} successfully wrote Consul test key to {leader}', log_file)
log.handler(
f"{atomic_msg} successfully wrote Consul test key to {leader}", log_file
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment