secretmgr retourne une belle erreur quand il voit un fichier vide #31
1 changed files with 51 additions and 25 deletions
|
|
@ -78,17 +78,18 @@ consul_server = consul.Consul()
|
|||
|
||||
# ----
|
||||
|
||||
USER = "USER"
|
||||
USER_LONG = "USER_LONG"
|
||||
CMD = "CMD"
|
||||
CONST = "CONST"
|
||||
CONST_LONG = "CONST_LONG"
|
||||
SERVICE_DN = "SERVICE_DN"
|
||||
USER = "USER"
|
||||
USER_LONG = "USER_LONG"
|
||||
CMD = "CMD"
|
||||
CONST = "CONST"
|
||||
CONST_LONG = "CONST_LONG"
|
||||
SERVICE_DN = "SERVICE_DN"
|
||||
SERVICE_PASSWORD = "SERVICE_PASSWORD"
|
||||
SSL_CERT = "SSL_CERT"
|
||||
SSL_KEY = "SSL_KEY"
|
||||
RSA_PUBLIC_KEY = "RSA_PUBLIC_KEY"
|
||||
RSA_PRIVATE_KEY = "RSA_PRIVATE_KEY"
|
||||
SSL_CERT = "SSL_CERT"
|
||||
SSL_KEY = "SSL_KEY"
|
||||
RSA_PUBLIC_KEY = "RSA_PUBLIC_KEY"
|
||||
RSA_PRIVATE_KEY = "RSA_PRIVATE_KEY"
|
||||
|
||||
|
||||
class bcolors:
|
||||
HEADER = '\033[95m'
|
||||
|
|
@ -101,8 +102,13 @@ class bcolors:
|
|||
BOLD = '\033[1m'
|
||||
UNDERLINE = '\033[4m'
|
||||
|
||||
|
||||
def read_secret(key, file_path):
|
||||
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(" ")
|
||||
stype = l0[0]
|
||||
secret = {"type": stype, "key": key}
|
||||
|
|
@ -127,11 +133,13 @@ def read_secret(key, file_path):
|
|||
if stype == RSA_PUBLIC_KEY:
|
||||
secret["key_desc"] = " ".join(l0[2:])
|
||||
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)
|
||||
|
||||
return secret
|
||||
|
||||
|
||||
def read_secrets(module_list):
|
||||
secrets = {}
|
||||
for mod in module_list:
|
||||
|
|
@ -141,6 +149,7 @@ def read_secrets(module_list):
|
|||
secrets[key] = read_secret(key, file_path)
|
||||
return secrets
|
||||
|
||||
|
||||
def get_secrets_services(secrets):
|
||||
services = {}
|
||||
for key, secret in secrets.items():
|
||||
|
|
@ -150,7 +159,7 @@ def get_secrets_services(secrets):
|
|||
print(svc, "@", key, bcolors.OKCYAN, "...", bcolors.ENDC)
|
||||
if svc not in services:
|
||||
services[svc] = {
|
||||
"dn": "cn=%s,%s"%(svc, SERVICE_DN_SUFFIX),
|
||||
"dn": "cn=%s,%s" % (svc, SERVICE_DN_SUFFIX),
|
||||
"pass": None,
|
||||
"dn_at": [],
|
||||
"pass_at": [],
|
||||
|
|
@ -164,11 +173,15 @@ def get_secrets_services(secrets):
|
|||
_, data = consul_server.kv.get(key)
|
||||
if data is not 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
|
||||
|
||||
|
||||
ldap_admin_conn = None
|
||||
|
||||
|
||||
def get_ldap_admin_conn():
|
||||
global ldap_admin_conn
|
||||
if ldap_admin_conn is None:
|
||||
|
|
@ -180,6 +193,7 @@ def get_ldap_admin_conn():
|
|||
|
||||
# ---- CHECK COMMAND ----
|
||||
|
||||
|
||||
def check_secrets(module_list):
|
||||
secrets = read_secrets(module_list)
|
||||
print("Found", len(secrets), "secrets to check")
|
||||
|
|
@ -188,6 +202,7 @@ def check_secrets(module_list):
|
|||
check_secrets_presence(secrets)
|
||||
check_secrets_services(secrets)
|
||||
|
||||
|
||||
def check_secrets_presence(secrets):
|
||||
print("Checking secrets presence...")
|
||||
for key in secrets.keys():
|
||||
|
|
@ -198,6 +213,7 @@ def check_secrets_presence(secrets):
|
|||
print(key, bcolors.OKGREEN, "✓", bcolors.ENDC)
|
||||
print()
|
||||
|
||||
|
||||
def check_secrets_services(secrets):
|
||||
print("Checking secrets for LDAP service users...")
|
||||
services = get_secrets_services(secrets)
|
||||
|
|
@ -208,7 +224,8 @@ def check_secrets_services(secrets):
|
|||
if data is not None:
|
||||
got_val = data["Value"].decode('ascii').strip()
|
||||
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"])
|
||||
|
||||
if svc["pass"] is None:
|
||||
|
|
@ -219,7 +236,8 @@ def check_secrets_services(secrets):
|
|||
if data is not None:
|
||||
got_val = data["Value"].decode('ascii').strip()
|
||||
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)
|
||||
try:
|
||||
|
|
@ -243,6 +261,7 @@ def gen_secrets(module_list, regen):
|
|||
check_secrets_presence(secrets)
|
||||
check_secrets_services(secrets)
|
||||
|
||||
|
||||
def gen_secrets_base(secrets, regen):
|
||||
print("Filling in user secrets and cmd secrets...")
|
||||
|
||||
|
|
@ -268,7 +287,8 @@ def gen_secrets_base(secrets, regen):
|
|||
print(key)
|
||||
print("Description:", secret["desc"])
|
||||
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:
|
||||
lines = []
|
||||
while True:
|
||||
|
|
@ -299,6 +319,7 @@ def gen_secrets_base(secrets, regen):
|
|||
|
||||
print()
|
||||
|
||||
|
||||
def gen_secrets_services(secrets, regen):
|
||||
print("Generating LDAP service accounts...")
|
||||
services = get_secrets_services(secrets)
|
||||
|
|
@ -316,7 +337,8 @@ def gen_secrets_services(secrets, regen):
|
|||
|
||||
if svc["pass"] is None or regen:
|
||||
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)
|
||||
try:
|
||||
|
|
@ -327,11 +349,13 @@ def gen_secrets_services(secrets, regen):
|
|||
for pass_key in svc["pass_at"]:
|
||||
_, data = consul_server.kv.get(pass_key)
|
||||
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"])
|
||||
|
||||
print()
|
||||
|
||||
|
||||
def fix_service_user(svc):
|
||||
print("Fixing service user", svc["dn"], "...")
|
||||
l = get_ldap_admin_conn()
|
||||
|
|
@ -344,16 +368,17 @@ def fix_service_user(svc):
|
|||
("objectclass", [b"person", b"top"]),
|
||||
("displayname", [svc["desc"].encode('ascii')]),
|
||||
("userpassword", [pass_crypt.encode('ascii')]),
|
||||
])
|
||||
])
|
||||
else:
|
||||
print(bcolors.OKCYAN, "Resetting entity password", bcolors.ENDC)
|
||||
l.modify_s(svc["dn"],
|
||||
[
|
||||
(ldap.MOD_REPLACE, "userpassword", [pass_crypt.encode('ascii')])
|
||||
])
|
||||
[
|
||||
(ldap.MOD_REPLACE, "userpassword", [pass_crypt.encode('ascii')])
|
||||
])
|
||||
|
||||
# ---- MAIN ----
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
for i, val in enumerate(sys.argv):
|
||||
if val == "check":
|
||||
|
|
@ -365,5 +390,6 @@ if __name__ == "__main__":
|
|||
elif val == "regen":
|
||||
gen_secrets(sys.argv[i+1:], True)
|
||||
break
|
||||
|
||||
|
||||
n":
|
||||
gen_secrets(sys.argv[i+1:], True)
|
||||
break
|
||||
|
|
|
|||
Reference in a new issue