1052 lines
39 KiB
Python
1052 lines
39 KiB
Python
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.<locals>.' 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, <urllib3.response.HTTPResponse object at 0x7fa236e0cdc0>
|
|
# TypeError: cannot pickle 'zlib.Decompress' objec
|
|
'url': req.url,
|
|
'encoding': req.encoding,
|
|
'history': req.history,
|
|
'reason': req.reason,
|
|
# 'cookies': req.cookies, # <RequestsCookieJar[]>
|
|
'elapsed': req.elapsed,
|
|
# 'request': req.request, # includes headers, cookies, <PreparedRequest [GET]>
|
|
# 'connection': req.connection, # <requests.adapters.HTTPAdapter object at 0x7f333ee1f620>
|
|
}
|
|
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()
|