rougail-user-data-bitwarden/src/rougail/user_data_bitwarden/data.py

273 lines
8.4 KiB
Python

"""
Silique (https://www.silique.fr)
Copyright (C) 2025
distribued with GPL-2 or later license
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""
from subprocess import run
from json import loads
from os import environ
from shutil import which
from tiramisu.error import ConfigError, display_list
from rougail.error import ExtensionError
from .i18n import _
class FakeBW:
def get_key(
self,
key_bitwarden: str,
multiple: bool,
):
return [{'name': key_bitwarden, 'login': {'username': "example_login", 'password': "Ex4mpL3_P4ssw0rD"}}]
class RBW:
def unlocked(self):
try:
cpe = run(["rbw", "unlocked"], capture_output=True)
except Exception as exc:
return False
return cpe.returncode == 0
def get_key(
self,
key_bitwarden: str,
multiple: bool,
):
keys = []
items = run_commandline(["rbw", "search", key_bitwarden]).strip()
if items:
items = items.split("\n")
else:
items = []
for item in items:
if "@" in item:
keys.append(item.split("@", 1)[-1])
else:
keys.append(item.rsplit("/", 1)[-1])
if not multiple:
if not items:
return []
if len(items) > 1:
return [{"name": key} for key in keys]
keys = [key_bitwarden]
datas = []
for key in keys:
data = loads(
run_commandline(
["rbw", "get", key, "--raw", "--ignorecase"]
).strip()
)
datas.append({"name": key, "login": data["data"]})
return datas
class BW:
def unlocked(self):
try:
cpe = run(["bw", "status"], capture_output=True)
except Exception as exc:
return False
if cpe.returncode != 0:
return False
try:
data = loads(cpe.stdout.decode("utf8"))
if data["status"] == "unlocked":
return True
except:
pass
return False
def get_key(
self,
key_bitwarden: str,
*args,
):
return loads(
run_commandline(
["bw", "list", "items", "--search", key_bitwarden, "--nointeraction"]
)
)
class RougailUserDataBitwarden:
def __init__(
self,
config: "Config",
*,
rougailconfig: "RougailConfig" = None,
):
# this is the tiramisu config object
self.config = config
if rougailconfig is None:
from rougail.config import RougailConfig
rougailconfig = RougailConfig
user_data = rougailconfig["step.user_data"]
if "bitwarden" not in user_data:
user_data.append("bitwarden")
rougailconfig["step.user_data"] = user_data
else:
user_data = rougailconfig["step.user_data"]
self.rougailconfig = rougailconfig
if "bitwarden" not in user_data:
raise ExtensionError(_('"bitwarden" is not set in step.user_data'))
self.errors = []
self.warnings = []
self.leader_informations = {}
self.commands = {'rbw': RBW(),
'bw': BW(),
}
def run(self):
values = {}
self.command = self.get_command()
self.set_passwords(self.config.unrestraint, values)
return [
{
"source": 'Bitwarden',
"errors": self.errors,
"warnings": self.warnings,
"values": values,
"options": {
"secret_manager": True,
}
}
]
def get_command(self):
if self.rougailconfig["bitwarden.mock_enable"]:
return FakeBW()
one_is_find = False
for command_name in self.rougailconfig["bitwarden.command"]:
if not which(command_name):
continue
command = self.commands[command_name]
if not command.unlocked():
one_is_find = True
continue
return command
if one_is_find:
msg = _("please unlock Bitwarden password database with {0} command line")
else:
msg = _("cannot find Bitwarden command line {0} please install it")
raise ExtensionError(msg.format(display_list(list(self.commands))))
def set_passwords(self, optiondescription, values):
for option in optiondescription.list(uncalculated=True):
if option.isoptiondescription():
self.set_passwords(option, values)
elif option.information.get("bitwarden", False):
values[option.path(uncalculated=True)] = (set_password, self.leader_informations, self.command)
def set_password(leader_informations, command, *, option):
path = option.path()
leader_key = None
if option.isfollower():
leader_path = option.parent().leader().path()
if leader_path in leader_informations:
leader_key = leader_informations[leader_path][option.index()]
if not option.isleader():
return get_values(command, option, leader_key=leader_key)
leader_informations[path], option_values = get_values(
command, option, multiple=True, leader_key=option.value.default()[0]
)
return option_values
def get_values(command, option, *, multiple=False, leader_key=None):
if leader_key:
key = leader_key
else:
key = option.value.default()
type_ = option.information.get("type")
if "validator" not in option.property.get():
option.property.add("validator")
if not isinstance(key, str):
raise ConfigError(
_('the default value must be the Bitwarden item name')
)
try:
data = command.get_key(key, multiple)
except Exception as exc:
raise ConfigError(
_(
'cannot execute the "{0}" commandline from Bitwarden: {1}'
).format(command, exc)
)
if not data:
raise ConfigError(
_('item "{0}" in Bitwarden is not found"').format(
key
)
)
if len(data) != 1:
names = [d["name"] for d in data]
if multiple:
ret = []
return names, [
get_value(key, type_, d) for d in data
]
raise ConfigError(
_(
'several items found with name "{0}" in Bitwarden: "{1}"'
).format(key, '", "'.join(names))
)
value = get_value(key, type_, data[0])
if multiple:
return [data[0]["name"]], [value]
return value
def get_value( key_bitwarden: str, type_: str, data: dict) -> str:
try:
if type_ == "secret":
value = data["login"]["password"]
else:
value = data["login"]["username"]
except Exception as exc:
raise ConfigError(
_('unexpected datas "{0}" from Bitwarden: {2}').format(
key_bitwarden, exc
)
)
else:
if value is None:
if type_ == "secret":
bw_type = _("password")
else:
bw_type = _("username")
raise ConfigError(
_('item "{0}" in Bitwarden has no {1}').format(
key_bitwarden, bw_type
)
)
return value
def run_commandline(cmd) -> str:
cpe = run(cmd, capture_output=True)
returncode = cpe.returncode
err = cpe.stderr.decode("utf8")
if returncode != 0 or err:
raise ExtensionError("{0} ({1})".format(err, returncode))
return cpe.stdout.decode("utf8")