7x31
/
cmdb
mirror of https://github.com/veops/cmdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: add secrets feature

This commit is contained in:
fxiang21 2023-10-29 21:40:49 +08:00
parent ed0c151d45
commit 5fa8c438f4
3 changed files with 53 additions and 27 deletions
cmdb-api/api
commands
click_cmdb.py
lib/secrets
inner.py
views/cmdb
inner_secrets.py

63
cmdb-api/api/commands/click_cmdb.py
View File

@ -329,15 +329,28 @@ def cmdb_inner_secrets_init(address):
"""
init inner secrets for password feature
"""
KeyManage(backend=InnerKVManger).init()
res, ok = KeyManage(backend=InnerKVManger).init()
if not ok:
if res.get("status") == "failed":
KeyManage.print_response(res)
if address and address.startswith("http") and current_app.config.get("INNER_TRIGGER_TOKEN", "") != "":
resp = requests.post("{}/api/v0.1/secrets/auto_seal".format(address.strip("/")),
headers={"Inner-Token": current_app.config.get("INNER_TRIGGER_TOKEN", "")})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
token = res.get("details", {}).get("root_token", "")
if address:
if not address.startswith(("http://127.0.0.1", "https://127.0.0.1")):
response = {
"message": "Address should start with http://127.0.0.1 or https://127.0.0.1",
"status": "failed"
}
KeyManage.print_response(response)
else:
KeyManage.print_response({"message": resp.text, "status": "failed"})
resp = requests.post("{}/api/v0.1/secrets/auto_seal".format(address.strip("/")),
headers={"Inner-Token": token})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
else:
KeyManage.print_response({"message": resp.text or resp.status_code, "status": "failed"})
else:
KeyManage.print_response(res)
@click.command()
@ -353,8 +366,12 @@ def cmdb_inner_secrets_unseal(address):
unseal the secrets feature
"""
address = "{}/api/v0.1/secrets/unseal".format(address.strip("/"))
if not address.startswith("http"):
KeyManage.print_response({"message": "invalid address, should start with http", "status": "failed"})
if not address.startswith(("http://127.0.0.1", "https://127.0.0.1")):
response = {
"message": "Address should start with http://127.0.0.1 or https://127.0.0.1",
"status": "failed"
}
KeyManage.print_response(response)
return
for i in range(global_key_threshold):
token = click.prompt(f'Enter unseal token {i + 1}', hide_input=True, confirmation_prompt=False)
@ -362,8 +379,10 @@ def cmdb_inner_secrets_unseal(address):
resp = requests.post(address, headers={"Unseal-Token": token})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
if resp.json().get("status") == "success":
return
else:
KeyManage.print_response({"message": resp.text, "status": "failed"})
KeyManage.print_response({"message": resp.status_code, "status": "failed"})
return
@ -388,15 +407,21 @@ def cmdb_inner_secrets_seal(address, token):
"""
assert address is not None
assert token is not None
if address.startswith("http"):
address = "{}/api/v0.1/secrets/seal".format(address.strip("/"))
resp = requests.post(address, headers={
"Inner-Token": token,
})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
else:
KeyManage.print_response({"message": resp.text, "status": "failed"})
if not address.startswith(("http://127.0.0.1", "https://127.0.0.1")):
response = {
"message": "Address should start with http://127.0.0.1 or https://127.0.0.1",
"status": "failed"
}
KeyManage.print_response(response)
return
address = "{}/api/v0.1/secrets/seal".format(address.strip("/"))
resp = requests.post(address, headers={
"Inner-Token": token,
})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
else:
KeyManage.print_response({"message": resp.status_code, "status": "failed"})
@click.command()

16
cmdb-api/api/lib/secrets/inner.py
View File

@ -204,33 +204,33 @@ class KeyManage:
"""
root_key = self.backend.get(backend_root_key_name)
if root_key:
return {"message": "already init, skip"}, False
return {"message": "already init, skip", "status": "skip"}, False
else:
root_key, shares, status = self.generate_unseal_keys()
if not status:
return {"message": root_key}, False
return {"message": root_key, "status": "failed"}, False
# hash root key and store in backend
root_key_hash, ok = self.hash_root_key(root_key)
if not ok:
return {"message": root_key_hash}, False
return {"message": root_key_hash, "status": "failed"}, False
msg, ok = self.backend.add(backend_root_key_name, root_key_hash)
if not ok:
return {"message": msg}, False
return {"message": msg, "status": "failed"}, False
# generate encrypt key from root_key and store in backend
encrypt_key, ok = self.generate_encrypt_key(root_key)
if not ok:
return {"message": encrypt_key}
return {"message": encrypt_key, "status": "failed"}
encrypt_key_aes, status = InnerCrypt.aes_encrypt(root_key, encrypt_key)
if not status:
return {"message": encrypt_key_aes}
return {"message": encrypt_key_aes, "status": "failed"}
msg, ok = self.backend.add(backend_encrypt_key_name, encrypt_key_aes)
if not ok:
return {"message": msg}, False
return {"message": msg, "status": "failed"}, False
current_app.config["secrets_root_key"] = root_key
current_app.config["secrets_encrypt_key"] = encrypt_key
@ -330,7 +330,7 @@ class KeyManage:
for i, v in enumerate(shares):
print(
"unseal token " + str(i + 1) + ": " + Fore.RED + Back.CYAN + v.decode("utf-8") + Style.RESET_ALL)
"unseal token " + str(i + 1) + ": " + Fore.RED + Back.BLACK + v.decode("utf-8") + Style.RESET_ALL)
print()
print(Fore.GREEN + "root token: " + root_token.decode("utf-8") + Style.RESET_ALL)

1
cmdb-api/api/views/cmdb/inner_secrets.py
View File

@ -1,3 +1,4 @@
from api.lib.perm.auth import auth_abandoned
from api.resource import APIView
from api.lib.secrets.inner import KeyManage
from api.lib.secrets.secrets import InnerKVManger