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-28 13:31:58 +08:00
parent bfb1cb14b3
commit 7bc62ba66b
7 changed files with 123 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
cmdb-api/api/app.py
View File

@ -21,6 +21,7 @@ from api.extensions import (bcrypt, cache, celery, cors, db, es, login_manager,
from api.extensions import inner_secrets
from api.flask_cas import CAS
from api.models.acl import User
from api.lib.secrets.secrets import InnerKVManger
HERE = os.path.abspath(os.path.dirname(__file__))
PROJECT_ROOT = os.path.join(HERE, os.pardir)
@ -126,7 +127,7 @@ def register_extensions(app):
app.config.update(app.config.get("CELERY"))
celery.conf.update(app.config)
inner_secrets.init_app(app)
inner_secrets.init_app(app, InnerKVManger())
def register_blueprints(app):

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

@ -7,6 +7,7 @@ import json
import time
import click
import requests
from flask import current_app
from flask.cli import with_appcontext
from flask_login import login_user
@ -318,28 +319,61 @@ def cmdb_index_table_upgrade():
@click.command()
@click.option(
'-a',
'--address',
help='inner cmdb api, http://127.0.0.1:8000',
)
@with_appcontext
def cmdb_inner_secrets_init():
def cmdb_inner_secrets_init(address):
"""
init inner secrets for password feature
"""
KeyManage(backend=InnerKVManger).init()
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())
else:
KeyManage.print_response({"message": resp.text, "status": "failed"})
@click.command()
@click.option(
'-a',
'--address',
help='inner cmdb api, http://127.0.0.1:8000',
required=True,
)
@with_appcontext
def cmdb_inner_secrets_unseal():
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"})
return
for i in range(global_key_threshold):
token = click.prompt(f'Enter token {i + 1}', hide_input=True, confirmation_prompt=False)
token = click.prompt(f'Enter unseal token {i + 1}', hide_input=True, confirmation_prompt=False)
assert token is not None
res = KeyManage(backend=InnerKVManger).unseal(token)
KeyManage.print_response(res)
resp = requests.post(address, headers={"Unseal-Token": token})
if resp.status_code == 200:
KeyManage.print_response(resp.json())
else:
KeyManage.print_response({"message": resp.text, "status": "failed"})
return
@click.command()
@click.option(
'-a',
'--address',
help='inner cmdb api, http://127.0.0.1:8000',
required=True,
)
@click.option(
'-k',
'--token',
@ -348,23 +382,21 @@ def cmdb_inner_secrets_unseal():
hide_input=True,
)
@with_appcontext
def cmdb_inner_secrets_seal(token):
def cmdb_inner_secrets_seal(address, token):
"""
seal the secrets feature
"""
assert address is not None
assert token is not None
res = KeyManage(backend=InnerKVManger()).seal(token)
KeyManage.print_response(res)
@click.command()
@with_appcontext
def cmdb_inner_secrets_auto_seal():
"""
auto seal the secrets feature
"""
res = KeyManage(current_app.config.get("INNER_TRIGGER_TOKEN"), backend=InnerKVManger()).auto_unseal()
KeyManage.print_response(res)
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"})
@click.command()

1
cmdb-api/api/extensions.py
View File

@ -14,6 +14,7 @@ from api.lib.utils import RedisHandler
from api.lib.secrets.inner import KeyManage
bcrypt = Bcrypt()
login_manager = LoginManager()
db = SQLAlchemy(session_options={"autoflush": False})

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

@ -60,8 +60,13 @@ class KeyManage:
if backend:
self.backend = Backend(backend)
def init_app(self, app):
self.auto_unseal()
def init_app(self, app, backend=None):
self.trigger = app.config.get("INNER_TRIGGER_TOKEN")
if not self.trigger:
return
self.backend = backend
# resp = self.auto_unseal()
# self.print_response(resp)
def hash_root_key(self, value):
algorithm = hashes.SHA256()
@ -133,15 +138,17 @@ class KeyManage:
"message": "encrypt key is empty",
"status": "failed"
}
secrets_encrypt_key = InnerCrypt.aes_decrypt(string_to_bytes(root_key), encrypt_key_aes)
setattr(current_app, 'secrets_encrypt_key', secrets_encrypt_key)
setattr(current_app, 'secrets_root_key', root_key)
setattr(current_app, 'secrets_shares', [])
return {
"message": success,
"status": success
}
secrets_encrypt_key, ok = InnerCrypt.aes_decrypt(string_to_bytes(root_key), encrypt_key_aes)
if ok:
current_app.config["secrets_encrypt_key"] = secrets_encrypt_key
current_app.config["secrets_root_key"] = root_key
current_app.config["secrets_shares"] = []
return {"message": success, "status": success}
else:
return {
"message": secrets_encrypt_key,
"status": "failed"
}
def unseal(self, key):
if not self.is_seal():
@ -152,10 +159,14 @@ class KeyManage:
try:
t = [i for i in b64decode(key)]
v = (int("".join([chr(i) for i in t[-2:]])), bytes(t[:-2]))
shares = getattr(current_app, "secrets_shares", [])
print("............")
# shares = getattr(current_app.config, "secrets_shares", [])
shares = current_app.config.get("secrets_shares", [])
print("222222222222")
if v not in shares:
shares.append(v)
setattr(current_app, "secrets_shares", shares)
current_app.config["secrets_shares"] = shares
print("shares:", shares)
if len(shares) >= global_key_threshold:
recovered_secret = Shamir.combine(shares[:global_key_threshold])
return self.auth_root_secret(b64encode(recovered_secret))
@ -209,10 +220,10 @@ class KeyManage:
msg, ok = self.backend.add(backend_encrypt_key_name, encrypt_key_aes)
if not ok:
return {"message": msg}, False
#
setattr(current_app, 'secrets_root_key', root_key)
setattr(current_app, 'secrets_encrypt_key', encrypt_key)
current_app.config["secrets_root_key"] = root_key
current_app.config["secrets_encrypt_key"] = encrypt_key
print(".....", current_app.config["secrets_root_key"], current_app.config["secrets_encrypt_key"])
self.print_token(shares, root_token=root_key)
return {"message": "OK",
@ -271,8 +282,8 @@ class KeyManage:
"status": "failed"
}
else:
setattr(current_app, 'secrets_root_key', '')
setattr(current_app, 'secrets_encrypt_key', '')
current_app.config["secrets_root_key"] = ''
current_app.config["secrets_encrypt_key"] = ''
return {
"message": success,
@ -284,7 +295,7 @@ class KeyManage:
If there is no initialization or the root key is inconsistent, it is considered to be in a sealed state.
:return:
"""
secrets_root_key = getattr(current_app, 'secrets_root_key')
secrets_root_key = current_app.config.get("secrets_root_key")
root_key = self.backend.get(backend_root_key_name)
if root_key == "" or root_key != secrets_root_key:
return "invalid root key", True
@ -326,7 +337,8 @@ class KeyManage:
class InnerCrypt:
def __init__(self):
secrets_encrypt_key = getattr(current_app, 'secrets_encrypt_key', '')
secrets_encrypt_key = current_app.config.get("secrets_encrypt_key", "")
print("secrets_encrypt_key:", secrets_encrypt_key)
self.encrypt_key = b64decode(secrets_encrypt_key.encode("utf-8"))
def encrypt(self, plaintext):

1
cmdb-api/api/resource.py
View File

@ -46,5 +46,4 @@ def register_resources(resource_path, rest_api):
resource_cls.url_prefix = ("",)
if isinstance(resource_cls.url_prefix, six.string_types):
resource_cls.url_prefix = (resource_cls.url_prefix,)
rest_api.add_resource(resource_cls, *resource_cls.url_prefix)

38
cmdb-api/api/views/cmdb/inner_secrets.py Normal file
View File

@ -0,0 +1,38 @@
from api.resource import APIView
from api.lib.secrets.inner import KeyManage
from api.lib.secrets.secrets import InnerKVManger
from flask import request, abort
class InnerSecretUnSealView(APIView):
url_prefix = "/secrets/unseal"
def post(self):
unseal_key = request.headers.get("Unseal-Token")
res = KeyManage(backend=InnerKVManger()).unseal(unseal_key)
# if res.get("status") == "failed":
# return abort(400, res.get("message"))
return self.jsonify(**res)
class InnerSecretSealView(APIView):
url_prefix = "/secrets/seal"
def post(self):
unseal_key = request.headers.get("Inner-Token")
res = KeyManage(backend=InnerKVManger()).seal(unseal_key)
# if res.get("status") == "failed":
# return abort(400, res.get("message"))
return self.jsonify(**res)
class InnerSecretAutoSealView(APIView):
url_prefix = "/secrets/auto_seal"
def post(self):
unseal_key = request.headers.get("Inner-Token")
res = KeyManage(backend=InnerKVManger()).seal(unseal_key)
# if res.get("status") == "failed":
# return abort(400, res.get("message"))
return self.jsonify(**res)

27
cmdb-api/api/views/cmdb/secrets.py
View File

@ -1,27 +0,0 @@
from api.resource import APIView
from api.models.cmdb import InnerKV
from api.lib.secrets.inner import KeyManage
from flask import request, abort
class InnerSecretUnSealView(APIView):
url_prefix = "/secrets/unseal"
def post(self):
unseal_key = request.headers.get("Inner-Token")
res = KeyManage(InnerKV()).unseal(unseal_key)
if res.get("status") == "failed":
return abort(400, res.get("message"))
return self.jsonify(**res)
class InnerSecretSealView(APIView):
url_prefix = "/secrets/seal"
def post(self):
unseal_key = request.headers.get("Inner-Token")
res = KeyManage(InnerKV()).seal(unseal_key)
if res.get("status") == "failed":
return abort(400, res.get("message"))
return self.jsonify(**res)