From 8aaad929aca21ca2353e13f07868ea23956b14b8 Mon Sep 17 00:00:00 2001 From: Daniel Weschke Date: Tue, 28 Jul 2026 23:08:59 +0200 Subject: [PATCH] initial commit --- pb.py | 1051 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1051 insertions(+) create mode 100644 pb.py diff --git a/pb.py b/pb.py new file mode 100644 index 0000000..feb0c2a --- /dev/null +++ b/pb.py @@ -0,0 +1,1051 @@ +import sys +import os +import requests +import gnupg +import fnmatch +import json +import urllib +import yaml +import time +import re +from functools import singledispatchmethod +from typing import overload + +__version__ = '20260724' + +def str_presenter(dumper, data): + """configures yaml for dumping multiline strings + Ref: https://stackoverflow.com/questions/8640959/how-can-i-control-what-scalar-form-pyyaml-uses-for-my-data""" + if len(data.splitlines()) > 1: # check for multiline string + return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|') + return dumper.represent_scalar('tag:yaml.org,2002:str', data) +yaml.add_representer(str, str_presenter) +yaml.representer.SafeRepresenter.add_representer(str, str_presenter) # to use with safe_dum + +def get_valid_filename(name): + s = str(name).strip().replace(" ", "_") + s = re.sub(r"(?u)[^-\w.]", "", s) + if s in {"", ".", ".."}: + raise SuspiciousFileOperation(f"Could not derive file name from '{name}'") + return s + +class Passbolt: + """ + urlbase + + either secretkey and passphrase or fingerprint + """ + + @overload + def __init__(self, urlbase:str, fingerprint:str): + self.__init__(urlbase=urlbase, fingerprint=fingerprint) + + @overload + def __init__(self, urlbase:str, secretkey:str, passphrase:str): + self.__init__(urlbase=urlbase, secretkey=secretkey, passphrase=passphrase) + + def __init__( + self, urlbase:str, *, + secretkey:str=None, passphrase:str=None, + fingerprint:str=None, + requests_session_kwargs:dict=None, + login=True, + ): + """ + urlbase: api base part of the url, i.e. scheme://netloc, e.g. https://passbolt.com + requests_session_kwargs: e.g. {'verify': True} + """ + if fingerprint: + self.keyfingerprint = fingerprint + self.secretkey = None + self.passphrase = None + self.gpg = gnupg.GPG(use_agent=True) + elif secretkey and passphrase: + self.secretkey = secretkey + self.passphrase = passphrase + self.gpg = gnupg.GPG() + self.imported_keys = self.gpg.import_keys(secretkey) # Importing our private key + self.keyfingerprint = self.imported_keys.fingerprints[0] # Getting the fingerprint of our first private key + else: + raise ValueError("either secretkey and passphrase or fingerprint has to be defined") + self.gpg.trust_keys(self.keyfingerprint, "TRUST_ULTIMATE") # Giving our key Ultimate trust so that we can encrypt secrets with it. + + self.session = requests.Session() + if requests_session_kwargs is not None: + for k,v in requests_session_kwargs.items(): + print(k, v) + setattr(self.session, k, v) + + self.urlbase = urlbase + self.apiversion = {'api-version': 'v3'} + self.headers = {"Content-Type": "application/json"} + + if login: + self._login() + + def list_secret_keys(self): + res = '' + for i in self.gpg.list_keys(True): + res += f"{i['fingerprint']}, {', '.join(i['uids'])}\n" + return res + + def _login(self): + """ + https://help.passbolt.com/api/authentication + """ + path = "/auth/login.json" + + # stage1 + body = {"gpg_auth": {"keyid": self.keyfingerprint}} + login = self.request('post', path, data=body) + logindata = login.json() + # state2 + server_verify_token = urllib.parse.unquote( + login.headers["X-GPGAuth-User-Auth-Token"]).replace("\\+", " ") + user_token_result = self.decrypt(server_verify_token, getdata=True) + body["gpg_auth"]["user_token_result"] = user_token_result + login = self.request('post', path, data=body) + logindata = login.json() + + if login.headers['X-GPGAuth-Authenticated']: + self.userid = logindata['body']['profile']['user_id'] + self.fingerprint = logindata['body']['gpgkey']['fingerprint'] + # get shared keys and csrfToken + mdsk = self.metadata_secretkeys() + for k in mdsk: + self.gpg.import_keys(k['armored_key']) + self.gpg.trust_keys(k['fingerprint'], "TRUST_ULTIMATE") # Giving shared key Ultimate trust so that we can encrypt secrets with it. + csrfToken = self.session.cookies.get_dict()["csrfToken"] # csrfToken filled after a request + self.headers = {"Content-Type": "application/json", + "X-CSRF-Token": csrfToken} + else: + raise ValueError(logindata['header']['message']) + + def decrypt(self, encdata, getdata=False, castjson=False): + res = self.gpg.decrypt(encdata, passphrase=self.passphrase) + if getdata: + res = res.data.decode("utf-8") + if castjson: + res = json.loads(res) + return res + + def encrypt(self, data, recipients=None, getdata=True): + if recipients is None: + recipients = self.fingerprint + try: + res = self.gpg.encrypt(data.encode('utf-8'), recipients, armor=True) + if getdata: + return res.data.decode('utf-8') + return res + except Exception as e: + print(data) + print(e) + + def url(self, path:str, query:dict|str = None): + qd = self.apiversion + _query_as_str = False + if query is not None: + if isinstance(query, dict): + qd.update(query) + else: + _query_as_str = True + if _query_as_str: + _query = query + else: + _query = urllib.parse.urlencode(qd) + return f"{self.urlbase}{path}?{_query}" + + def request(self, rtype:str, path:str, query:dict|str = None, headers=None, data=None): + """ + rtype: get, put, post, delete + """ + if headers is None: + headers = self.headers + return getattr(self.session, rtype)( + self.url(path=path, query=query), headers=headers, json=data) + + def request_ret( + self, rtype:str, path:str, query:dict = None, headers=None, data=None, + ret=None): + """ + ret: + - None: response + - 'json': response.json() + - 'json-body': response.json()['body'] + - 'json-header-message': response.json()['header']['message'] + """ + res = self.request( + rtype=rtype, path=path, query=query, headers=headers, data=data) + if ret: + if ret == 'json': + return res.json() + elif ret == 'json-body': + return res.json()['body'] + elif ret == 'json-header-message': + return res.json()['header']['message'] + return res + + def get_folders(self, ret='json-body'): + return self.request_ret("get", f"/folders.json", ret=ret) + + def _folders(self): + self.folders = {d['id']:d for d in self.get_folders()} + + def folder_path(self, folder_parent_id:str|None): + res = None + if folder_parent_id is not None: + res = self.folders[folder_parent_id]['name'] + return res + + def get_gpgkeys(self, ret='json-body'): + return self.request_ret("get", "/gpgkeys.json", ret=ret) + + def get_groups(self, ret='json-body'): + return self.request_ret("get", "/groups.json", ret=ret) + + def get_healthcheck(self, ret='json-body'): + return self.request_ret("get", "/healthcheck.json", ret=ret) + + def get_healthcheck_status(self, ret='json-body'): + return self.request_ret("get", "/healthcheck/status.json", ret=ret) + + def get_metadata_keys( + self, query=None, contain_privates_keys=False, ret='json-body'): + if query is None: + query = {} + if contain_privates_keys: + query.update({'contain[metadata_private_keys]': 1}) + return self.request_ret("get", "/metadata/keys.json", query, ret=ret) + + def get_metadata_keys_settings(self, ret='json-body'): + return self.request_ret("get", "/metadata/keys/settings.json", ret=ret) + + def metadata_secretkeys(self): + req = self.get_metadata_keys(contain_privates_keys=True) + data = [] + for mdk in req: + if 'metadata_private_keys' in mdk: + for mdsk in mdk['metadata_private_keys']: + data.append(self.decrypt(mdsk['data'], getdata=True, castjson=True)) + return data + + def get_metadata_types_settings(self, ret='json-body'): + return self.request_ret("get", "/metadata/types/settings.json", ret=ret) + + def get_metadata_rotate_key_folders(self, ret='json-body'): + return self.request_ret("get", "/metadata/rotate-key/folders.json", ret=ret) + + def get_metadata_rotate_key_resources(self, ret='json-body'): + return self.request_ret("get", "/metadata/rotate-key/resources.json", ret=ret) + + def get_metadata_rotate_key_tags(self, ret='json-body'): + # tags => status_code: 404 + return self.request_ret("get", "/metadata/rotate-key/tags.json", ret=ret) + + def get_metadata_session_keys(self, ret='json-body'): + return self.request_ret("get", "/metadata/session-keys.json", ret=ret) + + def get_permissions(self, resource_id:str, ret='json-body'): + return self.request_ret("get", f"/permissions/resource/{resource_id}.json", ret=ret) + + def get_resources( + self, contain_creator=False, contain_favorite=False, + contain_modifier=False, contain_secret=False, contain_resource_type=False, + contain_permission=False, contain_permissions=False, + contain_permissions_user_profile=False, contain_permissions_group=False, + contain_all=False, filter_is_favorite=False, + filter_is_shared_with_group:str = None, filter_is_owned_by_me=False, + filter_is_shared_with_me=False, filter_has_id:str = None, + filter_has_parent:str = None, filter_metadata_key_type:str = None, + query=None, ret='json-body'): + if query is None: + query = {} + if contain_creator: + query.update({'contain[creator]': 1}) + if contain_favorite: + query.update({'contain[favorite]': 1}) + if contain_modifier: + query.update({'contain[modifier]': 1}) + if contain_secret: + query.update({'contain[secret]': 1}) + if contain_resource_type: + query.update({'contain[resource-type]': 1}) + if contain_permission: + query.update({'contain[permission]': 1}) + if contain_permissions: + query.update({'contain[permissions]': 1}) + if contain_permissions_user_profile: + query.update({'contain[permissions.user.profile]': 1}) + if contain_permissions_group: + query.update({'contain[permissions.group]': 1}) + if contain_all: + query.update({ + 'contain[creator]': 1, 'contain[favorite]': 1, + 'contain[modifier]': 1, 'contain[secret]': 1, + 'contain[resource-type]': 1, 'contain[permission]': 1, + 'contain[permissions]': 1, 'contain[permissions.user.profile]': 1, + 'contain[permissions.group]': 1, + }) + if filter_is_favorite: + query.update({'filter[is-favorite]': True}) + if filter_is_shared_with_group: + query.update({'filter[is-shared-with-group]': filter_is_shared_with_group}) + if filter_is_owned_by_me: + query.update({'filter[is-owned-by-me]': True}) + if filter_is_shared_with_me: + query.update({'filter[is-shared-with-me]': True}) + if filter_has_id: + query.update({'filter[has-id]': filter_has_id}) + if filter_has_parent: + query.update({'filter[has-parent]': filter_has_parent}) + if filter_metadata_key_type: + query.update({'filter[metadata_key_type]': filter_metadata_key_type}) + return self.request_ret('get', '/resources.json', query, ret=ret) + + def get_resource( + self, resourceid:str, contain_creator=False, contain_favorite=False, + contain_modifier=False, contain_secret=False, contain_resource_type=False, + contain_permission=False, contain_permissions=False, + contain_permissions_user_profile=False, contain_permissions_group=False, + contain_all=False, query=None, ret='json-body'): + return self.request_ret('get', f'/resources/{resourceid}.json', ret=ret) + + def dec_resource_metadata(self, resource): + return json.loads(self.decrypt(resource['metadata']).data) + + def _resources(self) -> dict: + """ + """ + resources = self.get_resources(contain_all=True) + d = {} + for resource in resources: + id = resource['id'] + resource['metadata'] = self.dec_resource_metadata(resource) + resource['secrets'] = self.dec_secrets_resource(resource) + d[id] = resource + self.resources = d + + def get_resource_types(self, resource_type_id:str=None, ret='json-body'): + if resource_type_id: + return self.request_ret("get", f"/resource-types/{resource_type_id}.json", ret=ret) + return self.request_ret("get", f"/resource-types.json", ret=ret) + + def _resource_types(self): + self.resource_types = { + i['id']:{'slug': i['slug'], 'name': i['name'], 'description': i['description']} + for i in self.get_resource_types()} + + def get_roles(self, ret='json-body'): + return self.request_ret("get", f"/roles.json", ret=ret) + + def get_secrets_resource(self, resource_id:str, ret='json-body'): + return self.request_ret("get", f"/secrets/resource/{resource_id}.json", ret=ret) + + def dec_secrets_resource(self, resource): + return json.loads(self.decrypt(self.get_secrets_resource(resource['id'])["data"], getdata=True)) + + def get_settings(self, ret='json-body'): + return self.request_ret("get", f"/settings.json", ret=ret) + + def get_share_aros(self, ret='json-body'): + return self.request_ret("get", "/share/search-aros.json", ret=ret) + + def get_tags(self, ret='json-body'): + # /tags.json => status_code: 404 + return self.request_ret("get", "/tags.json", ret=ret) + + def get_users( + self, contain_last_logged_in=False, contain_groups_users=False, + contain_gpgkey=False, contain_profile=False, contain_role=False, + contain_missing_metadata_key_ids=False, contain_all=False, + filter_search:str|None = None, filter_has_groups=None, + filter_has_access=None, filter_is_admin=False, filter_is_active=False, + user_id:str|None = None, query=None, ret='json-body'): + if query is None: + query = {} + if contain_last_logged_in: + query.update({'contain[last_logged_in]': 1}) + if contain_groups_users: + query.update({'contain[groups_users]': 1}) + if contain_gpgkey: + query.update({'contain[gpgkey]': 1}) + if contain_profile: + query.update({'contain[profile]': 1}) + if contain_role: + query.update({'contain[role]': 1}) + if contain_missing_metadata_key_ids: + query.update({'contain[missing_metadata_key_ids]': 1}) + if contain_all: + query.update({ + 'contain[last_logged_in]': 1, 'contain[groups_users]': 1, + 'contain[gpgkey]': 1, 'contain[profile]': 1, 'contain[role]': 1, + 'contain[missing_metadata_key_ids]': 1, + }) + if filter_search: + query.update({'filter[search]': filter_search}) + if filter_has_groups: + query.update({'filter[has-groups]': filter_has_groups}) + if filter_has_access: + query.update({'filter[has-access]': filter_has_access}) + if filter_is_admin: + query.update({'filter[is-admin]': True}) + if filter_is_active: + query.update({'filter[is-active]': True}) + if user_id: + return self.request_ret("get", f"/users/{user_id}.json", query=query, ret=ret) + return self.request_ret("get", "/users.json", query=query, ret=ret) + + def _users(self): + self.users = {d['id']:d for d in self.get_users()} + + def export_dict(self, dirname=None, update=False, format_type=None) -> dict|str: + """ + returns dict or str if format_type=json + """ + self.__loadsave('users', dirname=dirname, update=update) + self.__loadsave('folders', dirname=dirname, update=update) + self.__loadsave('resource_types', dirname=dirname, update=update) + self.__loadsave('resources', dirname=dirname, update=update) + + d = {} + for id, resource in self.resources.items(): + r = {} + # name alias of username + r['name'] = resource['metadata']['name'] + r['username'] = resource['metadata']['username'] + try: + r['uris'] = resource['metadata']['uris'] + except: + ... + try: + r['description'] = resource['metadata']['description'] + except: + ... + try: + cfn = {cf['id']:cf for cf in resource['metadata']['custom_fields']} + for cf in resource['secrets']['custom_fields']: # merge both + cfn[cf['id']].update(cf) + r['custom_fields'] = { + cf['metadata_key']: cf['secret_value'] for cf in cfn.values()} + except: + ... + try: + r['password'] = resource['secrets']['password'] + except: + ... + try: + r['note'] = resource['secrets']['description'] + except: + ... + try: + r['totp'] = resource['secrets']['totp'] + except: + ... + r['folder'] = self.folder_path(resource['folder_parent_id']) + + r['resource_type'] = self.resource_types[resource['resource_type_id']]['slug'] + if self.resource_types[resource['resource_type_id']]['slug'] != self.resource_types[resource['metadata']['resource_type_id']]['slug']: + r['resource_type'] = [r['resource_type'], self.resource_types[resource['metadata']['resource_type_id']]['slug']] + + d[id] = r + self._export_dict = d + if format_type and format_type == 'json': + return json.dumps(d, indent=2) + return d + + def export_resources( + self, path=None, fmt='pass', cachedir=None, update=False): + """ + list of yaml format resources + pwfl: password first line + + filename: {id} + fmt: ''' + {secrets.password} + username: {metadata.username} + name: {metadta.name} + uris: {metadta.uris} + ''' + + Android app: Password Store + otpauth://totp/ACME%20Co:john@example.com?secret=HXDMVJECJJWSRB3HWIZR4IFUGFTMXBOZ&issuer=ACME%20Co&algorithm=SHA1&digits=6&period=30 + + pass-otp: searches for a line starting with otpauth:// and needs an account name + + account name is username if available otherwise name (title) + """ + r = self.export_dict(dirname=cachedir, update=update) + res = [] + if fmt == 'pass': + for k,v in r.items(): + pw = v['password'] if 'password' in v else '' + account = v['username'] if 'username' in v else v['name'] + totp = f"otpauth://totp/{urllib.parse.quote(account)}?secret={v['totp']['secret_key']}&period={v['totp']['period']}&digits={v['totp']['digits']}&algorithm={v['totp']['algorithm'].lower()}\n" if 'totp' in v else '' + e = f"{pw}\nusername: {v['username']}\ntitle: {v['name']}\n{totp}" + yaml.dump( + {vk:vv for vk,vv in v.items() if vk not in [ + 'password', 'name', 'username', 'totp']}, + default_flow_style=False, allow_unicode=True, indent=2) + res.append(e) + if path is not None: + if not os.path.isdir(os.path.expanduser(path)): + os.mkdir(os.path.expanduser(path)) + filename = os.path.expanduser(f"{path}/{get_valid_filename(v['name']+'_'+k)}.gpg") + with open(filename, 'w', encoding='utf8') as f: + f.write(self.encrypt(res[-1])) + else: + for k,v in r.items(): + e = yaml.dump( + v, default_flow_style=False, allow_unicode=True, indent=2) + res.append(e) + if path is not None: + if not os.path.isdir(os.path.expanduser(path)): + os.mkdir(os.path.expanduser(path)) + filename = os.path.expanduser(f"{path}/{get_valid_filename(v['name']+'_'+k)}.gpg") + with open(filename, 'w', encoding='utf8') as f: + f.write(self.encrypt(res[-1])) + return res + + def export_resources_pass( + self, path='~/.password-store/passbolt', cachedir=None, update=False): + return self.export_resources( + path=path, fmt='pass', cachedir=cachedir, update=update) + + def __loadsave( + self, attr, filename=None, dirname=None, func=None, para={}, update=False + ) -> dict: + """ + if filename=None and dirname=None => pb.ATTR + if filename=None and dirname => dirname/cache.ATTR + if filename => filename + """ + if filename is None: + if dirname is None: + filename = f"cache.{attr}" + else: + filename = os.path.join(dirname, f"cache.{attr}") + if not os.path.isdir(os.path.expanduser(dirname)): + os.mkdir(os.path.expanduser(dirname)) + if func is None: + func = getattr(self, f'_{attr}') + try: + getattr(self, attr) + except: + if not update and os.path.exists(filename): + with open(filename, encoding='utf8') as f: + setattr(self, attr, self.decrypt(f.read(), getdata=True, castjson=True)) + else: + func(**para) + setattr(self, attr, getattr(self, attr)) + with open(filename, 'w', encoding='utf8') as f: + f.write(self.encrypt(json.dumps(getattr(self, attr), indent=2))) + +def cli_click(): + import enum + import click + from alive_progress import alive_bar, config_handler + + # bar_titles_width + btw = max(map(len, [ + 'login', 'folders', 'gpg keys', 'groups', 'healthcheck', 'metadata keys', + 'metadata keys settings', 'metadata types settings', 'metadata rotate key', + 'metadata session keys', 'permissions', 'resources', 'resource types', + 'roles', 'secrets resource', 'settings', 'share search aros', 'tags', + 'users', 'request', 'export'])) + + config_handler.set_global( + length=40, spinner=None, unknown='waves', file=sys.stderr, stats=False, + title_length=btw) + + class FormatType(enum.Enum): + raw = enum.auto() + yaml = enum.auto() + json = enum.auto() + + from functools import update_wrapper + def pass_context(*dargs, **dkwargs): + """ + calls login + + @pass_context works + @pass_context(False) works not (an object) + @pass_context(time) works not (a module) + @pass_context(time.time) works (a function) + """ + # click.echo(f"DEBUG deco args: {dargs} {dkwargs}") + isf = False + if len(dargs): + # check if there is a first argument and if it is a callable or None (not + # a module, etc.) + assert callable(dargs[0]) or dargs[0] is None + if hasattr(dargs[0], '__qualname__') and 'cli_click..' in dargs[0].__qualname__: + isf = dargs[0] + def decorator(f): + @click.pass_context + def new_func(ctx, *args, **kwargs): + # click.echo(f'DEBUG {f.__name__} args: {ctx.params}') + ctx.obj.update(ctx.params) + if ctx.obj['debug']: + click.echo(f'DEBUG {f.__name__} obj:\n{yaml.dump(ctx.obj)}', err=True) + if f.__name__ != 'main' and 'pb' not in ctx.obj: + init(ctx, login=dkwargs['login'] if 'login' in dkwargs else True) + def echo(obj): + if ctx.obj['format_type'] == FormatType.json: + res = click.echo(json.dumps(obj)) + elif ctx.obj['format_type'] == FormatType.yaml: + res = click.echo(yaml.dump(obj)) + else: + res = click.echo(obj) + return res + ctx.echo = echo + ret = ctx.invoke(f, ctx, *args, **kwargs) + if ctx.obj['debug']: + click.echo(f'DEBUG {f.__name__} return: {ret}') + return ret + return update_wrapper(new_func, f) + return decorator(isf) if isf else decorator + + def init(ctx, login=True): + secretkey = ctx.obj['secretkey'] + passphrase = ctx.obj['passphrase'] + fingerprint = ctx.obj['fingerprint'] + urlbase = ctx.obj['urlbase'] + if secretkey and passphrase or fingerprint: + title = 'login' if login else 'init' + with alive_bar(title=title) as bar: + res = Passbolt( + urlbase=urlbase, secretkey=secretkey, passphrase=passphrase, + fingerprint=fingerprint, login=login + ) + bar() + ctx.obj['pb'] = res + else: + click.secho(f'error: define either secretkey and passphrase or fingerprint', fg='red') + sys.exit(1) + return ctx + + @click.group(context_settings=dict(help_option_names=['-h', '--help'])) + @click.option('-u', '--urlbase', help='API URL.', show_envvar=True) + @click.option('-s', '--secretkey', help='Private key.', show_envvar=True) + @click.option('-p', '--passphrase', help='Passphrase.', show_envvar=True) + @click.option('-f', '--fingerprint', help='Fingerprint.', show_envvar=True) + @click.option( + '-t', '--format', 'format_type', help='Format type.', + type=click.Choice(FormatType, case_sensitive=False), + envvar='PASSBOLT_FORMAT', show_envvar=True) + @click.option('-v', '--verbose', help='Verbose mode.', is_flag=True, show_envvar=True) + @click.option('-D', '--debug', is_flag=True, show_envvar=True, hidden=True) + @click.version_option(__version__) + @pass_context() + def main( + ctx, urlbase:str, *, + secretkey:str=None, passphrase:str=None, fingerprint:str=None, + format_type: FormatType, verbose=False, debug=False): + """Passbolt CLI. + + Define either SECRETKEY and PASSPHRASE or FINGERPRINT together with URLBASE. + URLBASE is the api base part of the url, i.e. scheme://netloc, e.g. https://passbolt.com + """ + ctx.ensure_object(dict) + + @main.command() + @pass_context + def info(ctx): + """Login info.""" + passbolt = ctx.obj['pb'] + res = { + 'fingerprint': passbolt.fingerprint, + 'secretkey': passbolt.secretkey, + # 'passphrase': passbolt.passphrase, + 'gpg': passbolt.gpg, + # 'imported_keys': passbolt.imported_keys, + # 'session': passbolt.session, + 'urlbase': passbolt.urlbase, + 'apiversion': passbolt.apiversion, + 'headers': passbolt.headers, + } + ctx.echo(res) + + @main.command() + @pass_context + def folders(ctx): + """Folders.""" + with alive_bar(title='folders') as bar: + res = ctx.obj['pb'].get_folders() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def gpgkeys(ctx): + """GPG keys.""" + with alive_bar(title='gpg-keys') as bar: + res = ctx.obj['pb'].get_gpgkeys() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def groups(ctx): + """Groups.""" + with alive_bar(title='groups') as bar: + res = ctx.obj['pb'].get_groups() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.option('-s', '--status', help='Check if passbolt is up.', is_flag=True) + @pass_context + def healthcheck(ctx, status): + """Get healthcheck information [admin].""" + with alive_bar(title='healthcheck') as bar: + if status: + res = ctx.obj['pb'].get_healthcheck_status() + else: + res = ctx.obj['pb'].get_healthcheck() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def metadata_keys(ctx): + """Metadata keys.""" + with alive_bar(title='metadata keys') as bar: + res = ctx.obj['pb'].get_metadata_keys() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def metadata_keys_settings(ctx): + """Metadata keys settings.""" + with alive_bar(title='metadata keys settings') as bar: + res = ctx.obj['pb'].get_metadata_keys_settings() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def metadata_types_settings(ctx): + """Metadata types settings.""" + with alive_bar(title='metadata types settings') as bar: + res = ctx.obj['pb'].get_metadata_types_settings() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('model', required=True) + @pass_context + def metadata_rotate_key(ctx, model): + """Metadata rotate key [admin]. + + Gather information about metadata keys that needs to be rotated. + + See the folders/resources/tags that are using an expired key that needs + to/can be rotated. + + MODEL one of: folders, resources or tags + + model=tags => status_code: 404 + """ + if model in ['folders', 'resources', 'tags']: + passbolt = ctx.obj['pb'] + with alive_bar(title='metadata rotate key') as bar: + if model == 'folders': + res = passbolt.get_metadata_rotate_key_folders() + elif model == 'resources': + res = passbolt.get_metadata_rotate_key_resources() + elif model == 'tags': + res = passbolt.get_metadata_rotate_key_tags() + else: + res = [] + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + else: + click.secho(f'model must be folders, resources or tags.', fg='red') + + @main.command() + @pass_context + def metadata_session_keys(ctx): + """Get session keys. + + Get a list of available encrypted session keys in cache for the given user. + """ + with alive_bar(title='metadata session keys') as bar: + res = ctx.obj['pb'].get_metadata_session_keys() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('resource_id', required=True) + @pass_context + def permissions(ctx, resource_id): + """Query permissions for resources. + + RESOURCE_ID ID of the resource. + """ + with alive_bar(title='permissions') as bar: + res = ctx.obj['pb'].get_permissions(resource_id) + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('resource_id', required=False) + @click.option('-c', '--contain_creator', help='Add creator to response body.', is_flag=True) + @click.option('-f', '--contain_favorite', help='Add favorite to response body.', is_flag=True) + @click.option('-m', '--contain_modifier', help='Add modifier to response body.', is_flag=True) + @click.option('-s', '--contain_secret', help='Add secrets to response body.', is_flag=True) + @click.option('-t', '--contain_resource_type', help='Add resource type to response body.', is_flag=True) + @click.option('-p', '--contain_permission', help='Add permission to response body.', is_flag=True) + @click.option('-P', '--contain_permissions', help='Add permissions to response body.', is_flag=True) + @click.option('-u', '--contain_permissions_user_profile', help='Add user profile to permissions in response body.', is_flag=True) + @click.option('-g', '--contain_permissions_group', help='Add group to permissions in response body.', is_flag=True) + @click.option('-a', '--contain_all', help='Add all in response body.', is_flag=True) + @click.option('-F', '--filter_is_favorite', help='Only return favorite elements.', is_flag=True) + @click.option('-G', '--filter_is_shared_with_group', help='Only return elements shared with group.', type=click.UUID) + @click.option('-O', '--filter_is_owned_by_me', help='Only return elements owned by yourself.', is_flag=True) + @click.option('-S', '--filter_is_shared_with_me', help='Only return elements shared to yourself and you are not owner.', is_flag=True) + @click.option('-I', '--filter_has_id', help='Only return elements filtered by their id.', type=click.UUID) + @click.option('-D', '--filter_has_parent', help='Only return elements filtered by parent folder.', type=click.UUID) + @click.option('-M', '--filter_metadata_key_type', help='Only return elements that match the metadata key type', type=click.UUID) + @pass_context + def resources(ctx, resource_id=None, contain_creator=False, contain_favorite=False, + contain_modifier=False, contain_secret=False, contain_resource_type=False, + contain_permission=False, contain_permissions=False, + contain_permissions_user_profile=False, contain_permissions_group=False, + contain_all=False, filter_is_favorite=False, + filter_is_shared_with_group:str = None, filter_is_owned_by_me=False, + filter_is_shared_with_me=False, filter_has_id:str = None, + filter_has_parent:str = None, filter_metadata_key_type:str = None): + """Resources. + + A resource holds the metadata for its secrets. + + RESOURCE_ID ID of the resource.""" + passbolt = ctx.obj['pb'] + with alive_bar(title='resources') as bar: + if resource_id: + res = passbolt.get_resource( + resource_id, + contain_creator=contain_creator, contain_favorite=contain_favorite, + contain_modifier=contain_modifier, contain_secret=contain_secret, + contain_resource_type=contain_resource_type, + contain_permission=contain_permission, + contain_permissions=contain_permissions, + contain_permissions_user_profile=contain_permissions_user_profile, + contain_permissions_group=contain_permissions_group) + bar() + else: + res = passbolt.get_resources( + contain_creator=contain_creator, contain_favorite=contain_favorite, + contain_modifier=contain_modifier, contain_secret=contain_secret, + contain_resource_type=contain_resource_type, + contain_permission=contain_permission, + contain_permissions=contain_permissions, + contain_permissions_user_profile=contain_permissions_user_profile, + contain_permissions_group=contain_permissions_group, + contain_all=contain_all, filter_is_favorite=filter_is_favorite, + filter_is_shared_with_group=filter_is_shared_with_group, + filter_is_owned_by_me=filter_is_owned_by_me, + filter_is_shared_with_me=filter_is_shared_with_me, + filter_has_id=filter_has_id, filter_has_parent=filter_has_parent, + filter_metadata_key_type=filter_metadata_key_type) + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('resource_type_id', required=False) + @pass_context + def resource_types(ctx, resource_type_id): + """Resource types. + + Resource-types are used for describing how and what data is stored for a + resource and its associated secrets. + + RESOURCE_TYPE_ID ID of the resource type. + """ + with alive_bar(title='resource types') as bar: + res = ctx.obj['pb'].get_resource_types(resource_type_id) + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def roles(ctx): + """Roles. + + Different categories of users. + """ + with alive_bar(title='roles') as bar: + res = ctx.obj['pb'].get_roles() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('resource_id', required=True) + @pass_context + def secrets_resource(ctx, resource_id): + """Secrets. + + Secrets associated to resources. + + RESOURCE_ID ID of the resource. + """ + with alive_bar(title='secrets resource') as bar: + res = ctx.obj['pb'].get_secrets_resource(resource_id) + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def settings(ctx): + """Settings. + + Retrieve the server settings. + """ + with alive_bar(title='settings') as bar: + res = ctx.obj['pb'].get_settings() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def share_search_aros(ctx): + """Shares. + + Get AROs for sharing. + """ + with alive_bar(title='share search aros') as bar: + res = ctx.obj['pb'].get_share_aros() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @pass_context + def tags(ctx): + """Tags. + + Get personal tags and shared tags. + + status_code: 404 + """ + with alive_bar(title='tags') as bar: + req = ctx.obj['pb'].get_tags(ret=None) + try: + res = req.json()['body'] + bar(len(res) if isinstance(res, list) else 1) + except requests.exceptions.JSONDecodeError as e: + res = click.style( + '\n' + yaml.dump(dict( + status_code=req.status_code, url=req.url, reason=req.reason)), + fg='red') + raise click.ClickException(res) + ctx.echo(res) + + @main.command() + @click.argument('user_id', required=False, type=click.UUID) + @click.option('-l', '--contain_last_logged_in', help='Add last logged in to response body.', is_flag=True) + @click.option('-u', '--contain_groups_users', help='Add group users to groups in response body.', is_flag=True) + @click.option('-g', '--contain_gpgkey', help='Add GPG key data to response body.', is_flag=True) + @click.option('-p', '--contain_profile', help='Add profile to response body.', is_flag=True) + @click.option('-r', '--contain_role', help='Add role to response body.', is_flag=True) + @click.option('-m', '--contain_missing_metadata_key_ids', help='Add missing metadata key IDs to response body.', is_flag=True) + @click.option('-a', '--contain_all', help='Add all to response body.', is_flag=True) + @click.option('-S', '--filter_search', help='Filter using a keyword or a string.') + @click.option('-G', '--filter_has_groups', help='Only return elements filtered by groups.', type=click.UUID, metavar='[UUID...]', multiple=True) + @click.option('-C', '--filter_has_access', help='Only return elements where specified users have access.', type=click.UUID, metavar='[UUID...]') + @click.option('-A', '--filter_is_admin', help='Only return admin users.', is_flag=True) + @click.option('-T', '--filter_is_active', help='Only return active elements.', is_flag=True) + @pass_context + def users( + ctx, contain_last_logged_in, contain_groups_users, contain_gpgkey, + contain_profile, contain_role, contain_missing_metadata_key_ids, + contain_all, filter_search, filter_has_groups, filter_has_access, + filter_is_admin, filter_is_active, user_id): + """Users. + + User are entities with the ability to interact with the application. + + USER_ID ID of the user. Can be set to me when viewing a user. + """ + with alive_bar(title='users') as bar: + res = ctx.obj['pb'].get_users( + contain_last_logged_in=contain_last_logged_in, + contain_groups_users=contain_groups_users, contain_gpgkey=contain_gpgkey, + contain_profile=contain_profile, contain_role=contain_role, + contain_missing_metadata_key_ids=contain_missing_metadata_key_ids, + contain_all=contain_all, filter_search=filter_search, + filter_has_groups=filter_has_groups, filter_has_access=filter_has_access, + filter_is_admin=filter_is_admin, filter_is_active=filter_is_active, + user_id=user_id) + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.argument('rtype', required=True) + @click.argument('path', required=True) + @click.argument('query', required=False) + @pass_context + def request(ctx, rtype, path, query): + """URL request. + + \b + RTYPE request type + PATH location + QUERY query + """ + #HEADERS headers + #DATA data + passbolt = ctx.obj['pb'] + with alive_bar(title='request') as bar: + req = passbolt.request(rtype=rtype, path=path, query=query) + bar() + res = { + 'status_code': req.status_code, + # 'headers': req.headers, + # 'raw': req.raw, + # TypeError: cannot pickle 'zlib.Decompress' objec + 'url': req.url, + 'encoding': req.encoding, + 'history': req.history, + 'reason': req.reason, + # 'cookies': req.cookies, # + 'elapsed': req.elapsed, + # 'request': req.request, # includes headers, cookies, + # 'connection': req.connection, # + } + try: + res['json'] = req.json() + except: + res['json'] = None + ctx.echo(res) + + @main.command() + @pass_context(login=False) + def secret_keys(ctx): + """List secret keys [gpg].""" + with alive_bar(title='list secret keys') as bar: + res = ctx.obj['pb'].list_secret_keys() + bar(len(res) if isinstance(res, list) else 1) + ctx.echo(res) + + @main.command() + @click.option( + '-u/-U', '--update/--no-update', is_flag=True, default=True, + help='if no-update then export only cached files') + @pass_context + def export(ctx, update): + """Export. + + Export all resources to the password-store""" + with alive_bar(title='export') as bar: + res = ctx.obj['pb'].export_resources_pass( + cachedir=click.get_app_dir('passbolt-cli'), update=update) + bar(len(res)) + click.echo('done.') + + + main(obj={}, auto_envvar_prefix='PASSBOLT') + + +if __name__ == '__main__': + cli_click()