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()