lerentis/bitwarden-crd-operator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

switch to JSON mode of cli

Browse Source 
pass around logger
This commit is contained in:
Nico Angelo 2023-04-21 16:50:33 +02:00
parent 0f518ab28d
commit ad1cc9f646
5 changed files with 25 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/bitwardenCrdOperator.py
View File

@ -9,12 +9,12 @@ from utils.utils import command_wrapper, unlock_bw
def bitwarden_signin(logger, **kwargs): def bitwarden_signin(logger, **kwargs):
if 'BW_HOST' in os.environ: if 'BW_HOST' in os.environ:
try: try:
command_wrapper(f"config server {os.getenv('BW_HOST')}") command_wrapper(logger, f"config server {os.getenv('BW_HOST')}")
except BaseException: except BaseException:
logger.warn("Revieved none zero exit code from server config") logger.warn("Received non-zero exit code from server config")
logger.warn("This is expected from startup") logger.warn("This is expected from startup")
pass pass
else: else:
logger.info("BW_HOST not set. Assuming SaaS installation") logger.info("BW_HOST not set. Assuming SaaS installation")
command_wrapper("login --apikey") command_wrapper(logger, "login --apikey")
unlock_bw(logger) unlock_bw(logger)

4
src/dockerlogin.py
View File

@ -46,7 +46,7 @@ def create_managed_registry_secret(spec, name, namespace, logger, **kwargs):
unlock_bw(logger) unlock_bw(logger)
logger.info(f"Locking up secret with ID: {id}") logger.info(f"Locking up secret with ID: {id}")
secret_json_object = json.loads(get_secret_from_bitwarden(id)) secret_json_object = get_secret_from_bitwarden(logger, id)
api = kubernetes.client.CoreV1Api() api = kubernetes.client.CoreV1Api()
@ -118,7 +118,7 @@ def update_managed_registry_secret(
unlock_bw(logger) unlock_bw(logger)
logger.info(f"Locking up secret with ID: {id}") logger.info(f"Locking up secret with ID: {id}")
secret_json_object = json.loads(get_secret_from_bitwarden(id)) secret_json_object = get_secret_from_bitwarden(logger, id)
api = kubernetes.client.CoreV1Api() api = kubernetes.client.CoreV1Api()

4
src/kv.py
View File

@ -45,7 +45,7 @@ def create_managed_secret(spec, name, namespace, logger, body, **kwargs):
unlock_bw(logger) unlock_bw(logger)
logger.info(f"Locking up secret with ID: {id}") logger.info(f"Locking up secret with ID: {id}")
secret_json_object = json.loads(get_secret_from_bitwarden(id)) secret_json_object = get_secret_from_bitwarden(logger, id)
api = kubernetes.client.CoreV1Api() api = kubernetes.client.CoreV1Api()
@ -106,7 +106,7 @@ def update_managed_secret(
unlock_bw(logger) unlock_bw(logger)
logger.info(f"Locking up secret with ID: {id}") logger.info(f"Locking up secret with ID: {id}")
secret_json_object = json.loads(get_secret_from_bitwarden(id)) secret_json_object = get_secret_from_bitwarden(logger, id)
api = kubernetes.client.CoreV1Api() api = kubernetes.client.CoreV1Api()

2
src/lookups/bitwarden_lookup.py
View File

@ -4,7 +4,7 @@ from utils.utils import get_secret_from_bitwarden, parse_fields_scope, parse_log
def bitwarden_lookup(id, scope, field): def bitwarden_lookup(id, scope, field):
_secret_json = json.loads(get_secret_from_bitwarden(id)) _secret_json = get_secret_from_bitwarden(None, id)
if scope == "login": if scope == "login":
return parse_login_scope(_secret_json, field) return parse_login_scope(_secret_json, field)
if scope == "fields": if scope == "fields":

31
src/utils/utils.py
View File

@ -7,44 +7,47 @@ class BitwardenCommandException(Exception):
pass pass
def get_secret_from_bitwarden(id): def get_secret_from_bitwarden(logger, id):
return command_wrapper(command=f"get item {id}") return command_wrapper(logger, command=f"get item {id}")
def unlock_bw(logger): def unlock_bw(logger):
status_output = command_wrapper("status") status_output = command_wrapper(logger, "status", False)
status = json.loads(status_output)['status'] status = status_output['data']['template']['status']
if status == 'unlocked': if status == 'unlocked':
logger.info("Already unlocked") logger.info("Already unlocked")
return return
token_output = command_wrapper("unlock --passwordenv BW_PASSWORD") token_output = command_wrapper(logger, "unlock --passwordenv BW_PASSWORD")
tokens = token_output.split('"')[1::2] os.environ["BW_SESSION"] = token_output["data"]["raw"]
os.environ["BW_SESSION"] = tokens[1]
logger.info("Signin successful. Session exported") logger.info("Signin successful. Session exported")
def command_wrapper(command): def command_wrapper(logger, command, use_success: bool = True):
system_env = dict(os.environ) system_env = dict(os.environ)
sp = subprocess.Popen( sp = subprocess.Popen(
[f"bw {command}"], [f"bw --response {command}"],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
close_fds=True, close_fds=True,
shell=True, shell=True,
env=system_env) env=system_env)
out, err = sp.communicate() out, err = sp.communicate()
if err: resp = json.loads(out.decode(encoding='UTF-8'))
raise BitwardenCommandException(err) if os.environ["DEBUG"] != None:
return out.decode(encoding='UTF-8') logger.info(resp)
if resp["success"] != None and (not use_success or (use_success and resp["success"] == True)):
return resp
logger.warn(resp)
return None
def parse_login_scope(secret_json, key): def parse_login_scope(secret_json, key):
return secret_json["login"][key] return secret_json["data"]["login"][key]
def parse_fields_scope(secret_json, key): def parse_fields_scope(secret_json, key):
if "fields" not in secret_json: if "fields" not in secret_json:
return None return None
for entry in secret_json["fields"]: for entry in secret_json["data"]["fields"]:
if entry['name'] == key: if entry['name'] == key:
return entry['value'] return entry['value']