Deuxfleurs/infrastructure
Archived
24
2
Fork 2
Code Issues 22 Pull requests Projects 1 Releases Wiki Activity

secretmgr retourne une belle erreur quand il voit un fichier vide #31

Closed
adrien wants to merge 1 commit from master into master
pull from: master
merge into: Deuxfleurs:master
Conversation 1 Commits 1 Files changed 1 +51 -25
1 changed files with 51 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit acbd3b6a54 - Show all commits

76
app/secretmgr.py
View file

@ -78,17 +78,18 @@ consul_server = consul.Consul()
# ---- # ----
USER = "USER" USER = "USER"
USER_LONG = "USER_LONG" USER_LONG = "USER_LONG"
CMD = "CMD" CMD = "CMD"
CONST = "CONST" CONST = "CONST"
CONST_LONG = "CONST_LONG" CONST_LONG = "CONST_LONG"
SERVICE_DN = "SERVICE_DN" SERVICE_DN = "SERVICE_DN"
SERVICE_PASSWORD = "SERVICE_PASSWORD" SERVICE_PASSWORD = "SERVICE_PASSWORD"
SSL_CERT = "SSL_CERT" SSL_CERT = "SSL_CERT"
SSL_KEY = "SSL_KEY" SSL_KEY = "SSL_KEY"
RSA_PUBLIC_KEY = "RSA_PUBLIC_KEY" RSA_PUBLIC_KEY = "RSA_PUBLIC_KEY"
RSA_PRIVATE_KEY = "RSA_PRIVATE_KEY" RSA_PRIVATE_KEY = "RSA_PRIVATE_KEY"
class bcolors: class bcolors:
HEADER = '\033[95m' HEADER = '\033[95m'
@ -101,8 +102,13 @@ class bcolors:
BOLD = '\033[1m' BOLD = '\033[1m'
UNDERLINE = '\033[4m' UNDERLINE = '\033[4m'
def read_secret(key, file_path): def read_secret(key, file_path):
lines = [l.strip() for l in open(file_path, "r")] lines = [l.strip() for l in open(file_path, "r")]
if len(lines) == 0:
print(bcolors.FAIL, "ERROR:", bcolors.ENDC, "Empty file ", file_path)
sys.exit(-1)
l0 = lines[0].split(" ") l0 = lines[0].split(" ")
stype = l0[0] stype = l0[0]
secret = {"type": stype, "key": key} secret = {"type": stype, "key": key}
@ -127,11 +133,13 @@ def read_secret(key, file_path):
if stype == RSA_PUBLIC_KEY: if stype == RSA_PUBLIC_KEY:
secret["key_desc"] = " ".join(l0[2:]) secret["key_desc"] = " ".join(l0[2:])
else: else:
print(bcolors.FAIL, "ERROR:", bcolors.ENDC, "Invalid secret type", stype, "in", file_path) print(bcolors.FAIL, "ERROR:", bcolors.ENDC,
"Invalid secret type", stype, "in", file_path)
sys.exit(-1) sys.exit(-1)
return secret return secret
def read_secrets(module_list): def read_secrets(module_list):
secrets = {} secrets = {}
for mod in module_list: for mod in module_list:
@ -141,6 +149,7 @@ def read_secrets(module_list):
secrets[key] = read_secret(key, file_path) secrets[key] = read_secret(key, file_path)
return secrets return secrets
def get_secrets_services(secrets): def get_secrets_services(secrets):
services = {} services = {}
for key, secret in secrets.items(): for key, secret in secrets.items():
@ -150,7 +159,7 @@ def get_secrets_services(secrets):
print(svc, "@", key, bcolors.OKCYAN, "...", bcolors.ENDC) print(svc, "@", key, bcolors.OKCYAN, "...", bcolors.ENDC)
if svc not in services: if svc not in services:
services[svc] = { services[svc] = {
"dn": "cn=%s,%s"%(svc, SERVICE_DN_SUFFIX), "dn": "cn=%s,%s" % (svc, SERVICE_DN_SUFFIX),
"pass": None, "pass": None,
"dn_at": [], "dn_at": [],
"pass_at": [], "pass_at": [],
@ -164,11 +173,15 @@ def get_secrets_services(secrets):
_, data = consul_server.kv.get(key) _, data = consul_server.kv.get(key)
if data is not None: if data is not None:
if services[svc]["pass"] is None: if services[svc]["pass"] is None:
services[svc]["pass"] = data["Value"].decode('ascii').strip() services[svc]["pass"] = data["Value"].decode(
'ascii').strip()
return services return services
ldap_admin_conn = None ldap_admin_conn = None
def get_ldap_admin_conn(): def get_ldap_admin_conn():
global ldap_admin_conn global ldap_admin_conn
if ldap_admin_conn is None: if ldap_admin_conn is None:
@ -180,6 +193,7 @@ def get_ldap_admin_conn():
# ---- CHECK COMMAND ---- # ---- CHECK COMMAND ----
def check_secrets(module_list): def check_secrets(module_list):
secrets = read_secrets(module_list) secrets = read_secrets(module_list)
print("Found", len(secrets), "secrets to check") print("Found", len(secrets), "secrets to check")
@ -188,16 +202,18 @@ def check_secrets(module_list):
check_secrets_presence(secrets) check_secrets_presence(secrets)
check_secrets_services(secrets) check_secrets_services(secrets)
def check_secrets_presence(secrets): def check_secrets_presence(secrets):
print("Checking secrets presence...") print("Checking secrets presence...")
for key in secrets.keys(): for key in secrets.keys():
_, data = consul_server.kv.get(key) _, data = consul_server.kv.get(key)
if data is None: if data is None:
print(key, bcolors.FAIL, "x", bcolors.ENDC) print(key, bcolors.FAIL, "x", bcolors.ENDC)
else: else:
print(key, bcolors.OKGREEN, "✓", bcolors.ENDC) print(key, bcolors.OKGREEN, "✓", bcolors.ENDC)
print() print()
def check_secrets_services(secrets): def check_secrets_services(secrets):
print("Checking secrets for LDAP service users...") print("Checking secrets for LDAP service users...")
services = get_secrets_services(secrets) services = get_secrets_services(secrets)
@ -208,7 +224,8 @@ def check_secrets_services(secrets):
if data is not None: if data is not None:
got_val = data["Value"].decode('ascii').strip() got_val = data["Value"].decode('ascii').strip()
if got_val != svc["dn"]: if got_val != svc["dn"]:
print(svc_name, "wrong DN at", dn_key, bcolors.FAIL, "x", bcolors.ENDC) print(svc_name, "wrong DN at", dn_key,
bcolors.FAIL, "x", bcolors.ENDC)
print("got:", got_val, "instead of:", svc["dn"]) print("got:", got_val, "instead of:", svc["dn"])
if svc["pass"] is None: if svc["pass"] is None:
@ -219,7 +236,8 @@ def check_secrets_services(secrets):
if data is not None: if data is not None:
got_val = data["Value"].decode('ascii').strip() got_val = data["Value"].decode('ascii').strip()
if got_val != svc["pass"]: if got_val != svc["pass"]:
print(svc_name, "wrong pass at", dn_key, bcolors.FAIL, "x", bcolors.ENDC) print(svc_name, "wrong pass at", dn_key,
bcolors.FAIL, "x", bcolors.ENDC)
l = ldap.initialize(LDAP_URL) l = ldap.initialize(LDAP_URL)
try: try:
@ -243,6 +261,7 @@ def gen_secrets(module_list, regen):
check_secrets_presence(secrets) check_secrets_presence(secrets)
check_secrets_services(secrets) check_secrets_services(secrets)
def gen_secrets_base(secrets, regen): def gen_secrets_base(secrets, regen):
print("Filling in user secrets and cmd secrets...") print("Filling in user secrets and cmd secrets...")
@ -268,7 +287,8 @@ def gen_secrets_base(secrets, regen):
print(key) print(key)
print("Description:", secret["desc"]) print("Description:", secret["desc"])
print("Enter value for secret, or ^C to skip:") print("Enter value for secret, or ^C to skip:")
print("THIS IS A LONG VALUE, ENTER SEVERAL LINES AND FINISH WITH A LINE CONTAINING A SINGLE .") print(
"THIS IS A LONG VALUE, ENTER SEVERAL LINES AND FINISH WITH A LINE CONTAINING A SINGLE .")
try: try:
lines = [] lines = []
while True: while True:
@ -299,6 +319,7 @@ def gen_secrets_base(secrets, regen):
print() print()
def gen_secrets_services(secrets, regen): def gen_secrets_services(secrets, regen):
print("Generating LDAP service accounts...") print("Generating LDAP service accounts...")
services = get_secrets_services(secrets) services = get_secrets_services(secrets)
@ -316,7 +337,8 @@ def gen_secrets_services(secrets, regen):
if svc["pass"] is None or regen: if svc["pass"] is None or regen:
print(bcolors.OKCYAN, "Generating new password", bcolors.ENDC) print(bcolors.OKCYAN, "Generating new password", bcolors.ENDC)
svc["pass"] = base64.urlsafe_b64encode(token_bytes(12)).decode('ascii') svc["pass"] = base64.urlsafe_b64encode(
token_bytes(12)).decode('ascii')
l = ldap.initialize(LDAP_URL) l = ldap.initialize(LDAP_URL)
try: try:
@ -327,11 +349,13 @@ def gen_secrets_services(secrets, regen):
for pass_key in svc["pass_at"]: for pass_key in svc["pass_at"]:
_, data = consul_server.kv.get(pass_key) _, data = consul_server.kv.get(pass_key)
if data is None or data["Value"].decode('ascii').strip() != svc["pass"]: if data is None or data["Value"].decode('ascii').strip() != svc["pass"]:
print(bcolors.OKCYAN, "Setting password", bcolors.ENDC, "at", pass_key) print(bcolors.OKCYAN, "Setting password",
bcolors.ENDC, "at", pass_key)
consul_server.kv.put(pass_key, svc["pass"]) consul_server.kv.put(pass_key, svc["pass"])
print() print()
def fix_service_user(svc): def fix_service_user(svc):
print("Fixing service user", svc["dn"], "...") print("Fixing service user", svc["dn"], "...")
l = get_ldap_admin_conn() l = get_ldap_admin_conn()
@ -344,16 +368,17 @@ def fix_service_user(svc):
("objectclass", [b"person", b"top"]), ("objectclass", [b"person", b"top"]),
("displayname", [svc["desc"].encode('ascii')]), ("displayname", [svc["desc"].encode('ascii')]),
("userpassword", [pass_crypt.encode('ascii')]), ("userpassword", [pass_crypt.encode('ascii')]),
]) ])
else: else:
print(bcolors.OKCYAN, "Resetting entity password", bcolors.ENDC) print(bcolors.OKCYAN, "Resetting entity password", bcolors.ENDC)
l.modify_s(svc["dn"], l.modify_s(svc["dn"],
[ [
(ldap.MOD_REPLACE, "userpassword", [pass_crypt.encode('ascii')]) (ldap.MOD_REPLACE, "userpassword", [pass_crypt.encode('ascii')])
]) ])
# ---- MAIN ---- # ---- MAIN ----
if __name__ == "__main__": if __name__ == "__main__":
for i, val in enumerate(sys.argv): for i, val in enumerate(sys.argv):
if val == "check": if val == "check":
@ -365,5 +390,6 @@ if __name__ == "__main__":
elif val == "regen": elif val == "regen":
gen_secrets(sys.argv[i+1:], True) gen_secrets(sys.argv[i+1:], True)
break break
n":
gen_secrets(sys.argv[i+1:], True)
break