<s> import json <EOL> import requests <EOL> from hvac import exceptions <EOL> class Client ( object ) : <EOL> def __init__ ( self , url = '<STR_LIT>' , token = None , <EOL> cert = None , verify = True , timeout = <NUM_LIT:30> , proxies = None ) : <EOL> self . token = token <EOL> self . _url = url <EOL> self . _kwargs = { <EOL> '<STR_LIT>' : cert , <EOL> '<STR_LIT>' : verify , <EOL> '<STR_LIT>' : timeout , <EOL> '<STR_LIT>' : proxies , <EOL> } <EOL> def read ( self , path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . _get ( '<STR_LIT>' . format ( path ) ) . json ( ) <EOL> except exceptions . InvalidPath : <EOL> return None <EOL> def list ( self , path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> payload = { <EOL> '<STR_LIT:list>' : True <EOL> } <EOL> return self . _get ( '<STR_LIT>' . format ( path ) , params = payload ) . json ( ) <EOL> except exceptions . InvalidPath : <EOL> return None <EOL> def write ( self , path , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response = self . _put ( '<STR_LIT>' . format ( path ) , json = kwargs ) <EOL> if response . status_code == <NUM_LIT:200> : <EOL> return response . json ( ) <EOL> def delete ( self , path ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' . format ( path ) ) <EOL> def is_initialized ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) [ '<STR_LIT>' ] <EOL> def initialize ( self , secret_shares = <NUM_LIT:5> , secret_threshold = <NUM_LIT:3> , pgp_keys = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : secret_shares , <EOL> '<STR_LIT>' : secret_threshold , <EOL> } <EOL> if pgp_keys : <EOL> if len ( pgp_keys ) != secret_shares : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> params [ '<STR_LIT>' ] = pgp_keys <EOL> return self . _put ( '<STR_LIT>' , json = params ) . json ( ) <EOL> @ property <EOL> def seal_status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def is_sealed ( self ) : <EOL> return self . seal_status [ '<STR_LIT>' ] <EOL> def seal ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _put ( '<STR_LIT>' ) <EOL> def unseal ( self , key ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:key>' : key , <EOL> } <EOL> return self . _put ( '<STR_LIT>' , json = params ) . json ( ) <EOL> def unseal_multi ( self , keys ) : <EOL> result = None <EOL> for key in keys : <EOL> result = self . unseal ( key ) <EOL> if not result [ '<STR_LIT>' ] : <EOL> break <EOL> return result <EOL> @ property <EOL> def key_status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def rotate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _put ( '<STR_LIT>' ) <EOL> @ property <EOL> def rekey_status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def start_rekey ( self , secret_shares = <NUM_LIT:5> , secret_threshold = <NUM_LIT:3> , pgp_keys = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : secret_shares , <EOL> '<STR_LIT>' : secret_threshold , <EOL> } <EOL> if pgp_keys : <EOL> if len ( pgp_keys ) != secret_shares : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> params [ '<STR_LIT>' ] = pgp_keys <EOL> resp = self . _put ( '<STR_LIT>' , json = params ) <EOL> if resp . text : <EOL> return resp . json ( ) <EOL> def cancel_rekey ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' ) <EOL> def rekey ( self , key , nonce = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:key>' : key , <EOL> } <EOL> if nonce : <EOL> params [ '<STR_LIT>' ] = nonce <EOL> return self . _put ( '<STR_LIT>' , json = params ) . json ( ) <EOL> def rekey_multi ( self , keys , nonce = None ) : <EOL> result = None <EOL> for key in keys : <EOL> result = self . rekey ( key , nonce = nonce ) <EOL> if result [ '<STR_LIT>' ] : <EOL> break <EOL> return result <EOL> @ property <EOL> def ha_status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def renew_secret ( self , lease_id , increment = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : increment , <EOL> } <EOL> return self . _post ( '<STR_LIT>' . format ( lease_id ) , json = params ) . json ( ) <EOL> def revoke_secret ( self , lease_id ) : <EOL> """<STR_LIT>""" <EOL> self . _put ( '<STR_LIT>' . format ( lease_id ) ) <EOL> def revoke_secret_prefix ( self , path_prefix ) : <EOL> """<STR_LIT>""" <EOL> self . _put ( '<STR_LIT>' . format ( path_prefix ) ) <EOL> def revoke_self_token ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _put ( '<STR_LIT>' ) <EOL> def list_secret_backends ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def enable_secret_backend ( self , backend_type , description = None , mount_point = None , config = None ) : <EOL> """<STR_LIT>""" <EOL> if not mount_point : <EOL> mount_point = backend_type <EOL> params = { <EOL> '<STR_LIT:type>' : backend_type , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : config , <EOL> } <EOL> self . _post ( '<STR_LIT>' . format ( mount_point ) , json = params ) <EOL> def disable_secret_backend ( self , mount_point ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' . format ( mount_point ) ) <EOL> def remount_secret_backend ( self , from_mount_point , to_mount_point ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : from_mount_point , <EOL> '<STR_LIT:to>' : to_mount_point , <EOL> } <EOL> self . _post ( '<STR_LIT>' , json = params ) <EOL> def list_policies ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) [ '<STR_LIT>' ] <EOL> def get_policy ( self , name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . _get ( '<STR_LIT>' . format ( name ) ) . json ( ) [ '<STR_LIT>' ] <EOL> except exceptions . InvalidPath : <EOL> return None <EOL> def set_policy ( self , name , rules ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( rules , dict ) : <EOL> rules = json . dumps ( rules ) <EOL> params = { <EOL> '<STR_LIT>' : rules , <EOL> } <EOL> self . _put ( '<STR_LIT>' . format ( name ) , json = params ) <EOL> def delete_policy ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' . format ( name ) ) <EOL> def list_audit_backends ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def enable_audit_backend ( self , backend_type , description = None , options = None , name = None ) : <EOL> """<STR_LIT>""" <EOL> if not name : <EOL> name = backend_type <EOL> params = { <EOL> '<STR_LIT:type>' : backend_type , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : options , <EOL> } <EOL> self . _post ( '<STR_LIT>' . format ( name ) , json = params ) <EOL> def disable_audit_backend ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' . format ( name ) ) <EOL> def audit_hash ( self , name , input ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:input>' : input , <EOL> } <EOL> return self . _post ( '<STR_LIT>' . format ( name ) , json = params ) . json ( ) <EOL> def create_token ( self , id = None , policies = None , meta = None , <EOL> no_parent = False , lease = None , display_name = None , <EOL> num_uses = None , no_default_profile = False , <EOL> ttl = None , orphan = False ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:id>' : id , <EOL> '<STR_LIT>' : policies , <EOL> '<STR_LIT>' : meta , <EOL> '<STR_LIT>' : no_parent , <EOL> '<STR_LIT>' : display_name , <EOL> '<STR_LIT>' : num_uses , <EOL> '<STR_LIT>' : no_default_profile , <EOL> } <EOL> if lease : <EOL> params [ '<STR_LIT>' ] = lease <EOL> else : <EOL> params [ '<STR_LIT>' ] = ttl <EOL> if orphan : <EOL> return self . _post ( '<STR_LIT>' , json = params ) . json ( ) <EOL> else : <EOL> return self . _post ( '<STR_LIT>' , json = params ) . json ( ) <EOL> def lookup_token ( self , token = None ) : <EOL> """<STR_LIT>""" <EOL> if token : <EOL> return self . _get ( '<STR_LIT>' . format ( token ) ) . json ( ) <EOL> else : <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def revoke_token ( self , token , orphan = False ) : <EOL> """<STR_LIT>""" <EOL> if orphan : <EOL> self . _post ( '<STR_LIT>' . format ( token ) ) <EOL> else : <EOL> self . _post ( '<STR_LIT>' . format ( token ) ) <EOL> def revoke_token_prefix ( self , prefix ) : <EOL> """<STR_LIT>""" <EOL> self . _post ( '<STR_LIT>' . format ( prefix ) ) <EOL> def renew_token ( self , token = None , increment = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : increment , <EOL> } <EOL> if token : <EOL> return self . _post ( '<STR_LIT>' . format ( token ) , json = params ) . json ( ) <EOL> else : <EOL> return self . _post ( '<STR_LIT>' , json = params ) . json ( ) <EOL> def logout ( self , revoke_token = False ) : <EOL> """<STR_LIT>""" <EOL> if revoke_token : <EOL> self . revoke_self_token ( ) <EOL> self . token = None <EOL> def is_authenticated ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . token : <EOL> return False <EOL> try : <EOL> self . lookup_token ( ) <EOL> return True <EOL> except exceptions . Forbidden : <EOL> return False <EOL> except exceptions . InvalidPath : <EOL> return False <EOL> def auth_app_id ( self , app_id , user_id , mount_point = '<STR_LIT>' , use_token = True ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : app_id , <EOL> '<STR_LIT>' : user_id , <EOL> } <EOL> return self . auth ( '<STR_LIT>' . format ( mount_point ) , json = params , use_token = use_token ) <EOL> def auth_tls ( self , mount_point = '<STR_LIT>' , use_token = True ) : <EOL> """<STR_LIT>""" <EOL> return self . auth ( '<STR_LIT>' . format ( mount_point ) , use_token = use_token ) <EOL> def auth_userpass ( self , username , password , mount_point = '<STR_LIT>' , use_token = True , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:password>' : password , <EOL> } <EOL> params . update ( kwargs ) <EOL> return self . auth ( '<STR_LIT>' . format ( mount_point , username ) , json = params , use_token = use_token ) <EOL> def create_userpass ( self , username , password , policies , mount_point = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( policies , ( list , set , tuple ) ) : <EOL> policies = '<STR_LIT:U+002C>' . join ( policies ) <EOL> params = { <EOL> '<STR_LIT:password>' : password , <EOL> '<STR_LIT>' : policies <EOL> } <EOL> return self . _post ( '<STR_LIT>' . format ( mount_point , username ) , json = params ) <EOL> def delete_userpass ( self , username , mount_point = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return self . _delete ( '<STR_LIT>' . format ( mount_point , username ) ) <EOL> def create_app_id ( self , app_id , policies , display_name = None , mount_point = '<STR_LIT>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( policies , ( list , set , tuple ) ) : <EOL> policies = '<STR_LIT:U+002C>' . join ( policies ) <EOL> params = { <EOL> '<STR_LIT:value>' : policies <EOL> } <EOL> if display_name : <EOL> params [ '<STR_LIT>' ] = display_name <EOL> params . update ( kwargs ) <EOL> return self . _post ( '<STR_LIT>' . format ( mount_point , app_id ) , json = params ) <EOL> def create_user_id ( self , user_id , app_id , cidr_block = None , mount_point = '<STR_LIT>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( app_id , ( list , set , tuple ) ) : <EOL> app_id = '<STR_LIT:U+002C>' . join ( app_id ) <EOL> params = { <EOL> '<STR_LIT:value>' : app_id <EOL> } <EOL> if cidr_block : <EOL> params [ '<STR_LIT>' ] = cidr_block <EOL> params . update ( kwargs ) <EOL> return self . _post ( '<STR_LIT>' . format ( mount_point , user_id ) , json = params ) <EOL> def auth_ldap ( self , username , password , mount_point = '<STR_LIT>' , use_token = True , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:password>' : password , <EOL> } <EOL> params . update ( kwargs ) <EOL> return self . auth ( '<STR_LIT>' . format ( mount_point , username ) , json = params , use_token = use_token ) <EOL> def auth_github ( self , token , mount_point = '<STR_LIT>' , use_token = True ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : token , <EOL> } <EOL> return self . auth ( '<STR_LIT>' . format ( mount_point ) , json = params , use_token = use_token ) <EOL> def auth ( self , url , use_token = True , ** kwargs ) : <EOL> response = self . _post ( url , ** kwargs ) . json ( ) <EOL> if use_token : <EOL> self . token = response [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return response <EOL> def list_auth_backends ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) . json ( ) <EOL> def enable_auth_backend ( self , backend_type , description = None , mount_point = None ) : <EOL> """<STR_LIT>""" <EOL> if not mount_point : <EOL> mount_point = backend_type <EOL> params = { <EOL> '<STR_LIT:type>' : backend_type , <EOL> '<STR_LIT:description>' : description , <EOL> } <EOL> self . _post ( '<STR_LIT>' . format ( mount_point ) , json = params ) <EOL> def disable_auth_backend ( self , mount_point ) : <EOL> """<STR_LIT>""" <EOL> self . _delete ( '<STR_LIT>' . format ( mount_point ) ) <EOL> def _get ( self , url , ** kwargs ) : <EOL> return self . __request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def _post ( self , url , ** kwargs ) : <EOL> return self . __request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def _put ( self , url , ** kwargs ) : <EOL> return self . __request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def _delete ( self , url , ** kwargs ) : <EOL> return self . __request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def __request ( self , method , url , headers = None , ** kwargs ) : <EOL> url = self . _url + url <EOL> if not headers : <EOL> headers = { } <EOL> if self . token : <EOL> headers [ '<STR_LIT>' ] = self . token <EOL> _kwargs = self . _kwargs . copy ( ) <EOL> _kwargs . update ( kwargs ) <EOL> response = requests . request ( method , <EOL> url , <EOL> headers = headers , <EOL> ** _kwargs ) <EOL> if response . status_code >= <NUM_LIT> and response . status_code < <NUM_LIT> : <EOL> errors = response . json ( ) . get ( '<STR_LIT>' ) <EOL> if response . status_code == <NUM_LIT> : <EOL> raise exceptions . InvalidRequest ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . Unauthorized ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . Forbidden ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . InvalidPath ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . RateLimitExceeded ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . InternalServerError ( errors = errors ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise exceptions . VaultDown ( errors = errors ) <EOL> else : <EOL> raise exceptions . UnexpectedError ( ) <EOL> return response </s>
