From 63d703daffa26c421cbec7f2b20deea21df4ea59 Mon Sep 17 00:00:00 2001 From: "Antonio J. Delgado" Date: Thu, 28 Nov 2024 20:02:32 +0200 Subject: [PATCH] Fix remove duplicates command --- nc_password_client/nc_password_client.py | 104 +++++++++++++++++------ 1 file changed, 77 insertions(+), 27 deletions(-) diff --git a/nc_password_client/nc_password_client.py b/nc_password_client/nc_password_client.py index 007fbd0..ff2f268 100755 --- a/nc_password_client/nc_password_client.py +++ b/nc_password_client/nc_password_client.py @@ -123,6 +123,11 @@ class NextcloudHandler: '.cache', 'nc_password_client.cache' ) + if 'comparation_fields' not in params: + self.comparation_fields = [ + 'username', 'password', 'url' + ] + self.field_replacements = {} for field_replacement in params['field_replacements']: key, value = field_replacement.split(':') @@ -944,25 +949,33 @@ class NextcloudHandler: def delete_password(self, post_obj): '''Delete a password''' + min_obj = { + 'id': post_obj['id'], + } try: r = self.session.delete( f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/delete', - data=post_obj, + data=min_obj, headers=self.headers(), auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout ) if r.status_code < 299: - for password in self.cache['cached_passwords']: - if self.cache['cached_passwords']['id'] == post_obj['id']: - self.cache['cached_passwords'].pop(password.numerator-1) - self._write_cache() + if r.json(): + counter = 0 + for password in self.cache['cached_passwords']: + if password['id'] == post_obj['id']: + self.cache['cached_passwords'].pop(counter) + self._write_cache() + counter += 1 return r.json() self.error( { "action": "delete_password", + "object": min_obj, "message": f"Nextcloud instance returned status code {r.status_code}", + "returned_content": r.content, } ) except requests.exceptions.ReadTimeout as error: @@ -1035,7 +1048,7 @@ class NextcloudHandler: return True return False - def is_same_password(self, obj1, obj2): + def is_same_password(self, obj1, obj2, ignore_empty_fields=False): '''Test if two password objects are the same or similar''' safer_obj1 = dict(obj1, **{ 'password': '***' }) safer_obj2 = dict(obj2, **{ 'password': '***' }) @@ -1060,22 +1073,51 @@ class NextcloudHandler: # } # } # ) - if self.is_same_key('username', obj1, obj2): - if self.is_same_key('password', obj1, obj2): - if self.is_same_key('url', obj1, obj2): - #if self.is_same_key('folder', obj1, obj2): - self.debug( - { - "action": "notify_match", - "object": { - "obj1": safer_obj1, - "obj2": safer_obj2 + match = True + for field in self.comparation_fields: + if ignore_empty_fields and (obj1[field] == '' or obj2[field] == ''): + continue + if not self.is_same_key(field, obj1, obj2): + if field == 'url': + url1 = self._split_url(obj1[field]) + url2 = self._split_url(obj2[field]) + if ('hostname' in url1 and 'hostname' in url2 and + url1['hostname'] == url2['hostname']): + self.debug( + { + "action": "is_same_password", + "message": f"Hostname in URL of both passwords match but URL is no match '{obj1[field]}' != '{obj2[field]}'" } - } - ) - return True + ) + continue + match = False + break + + if match: + self.debug( + { + "action": "notify_match", + "object": { + "obj1": safer_obj1, + "obj2": safer_obj2 + } + } + ) + return True return False + def _split_url(self, url): + result = {} + match = re.search('([a-z]*)://([^/]*)/(.*)', url) + if match: + result['protocol'] = match.group(1) + result['hostname'] = match.group(2) + result['path'] = match.group(3) + split_hostname = result['hostname'].split('.') + result['tld'] = split_hostname[len(split_hostname)-1] + result['domain'] = f"{split_hostname[len(split_hostname)-1]}.{split_hostname[len(split_hostname)-2]}" + return result + def get_folder_id(self, folder_name): '''Get a folder id from the name''' for folder in self.list_passwords_folders(): @@ -1253,12 +1295,13 @@ class NcPasswordClient: sys.exit(2) for password in self.nc.list_passwords(): if password[field] == name: - safer_obj = dict(password, **{ 'password': '***' }) self.debug( - { "action": "delete_password", "object": safer_obj } - ) - self.debug( - { "action": "deleted_password", "object": self.nc.delete_password(password)} + { + "action": "deleted_password", + "message": "Result of deleting password", + "object": password, + "result": self.nc.delete_password(password) + } ) return True self.warning( @@ -1360,8 +1403,9 @@ class NcPasswordClient: self.nc.delete_password(item) return True - def remove_duplicates(self, limit): + def remove_duplicates(self, limit, comparation_fields): '''Remove duplicate passwords''' + self.nc.comparation_fields = comparation_fields checked_passwords = [] count = 0 if limit == 0: @@ -1725,11 +1769,17 @@ def create_passwords_folder(ctx, name): default=-1, help='Maximun number of passwords to remove. -1 for unlimited.' ) +@click.option( + '--comparation-fields', '-c', + multiple=True, + default=['username', 'password', 'url'], + help='Field names to compare' +) @click_config_file.configuration_option() @click.pass_context -def remove_duplicates(ctx, limit): +def remove_duplicates(ctx, limit, comparation_fields): '''Remove duplicate passwords''' - ctx.obj['NcPasswordClient'].remove_duplicates(limit) + ctx.obj['NcPasswordClient'].remove_duplicates(limit, comparation_fields) @cli.command() @click_config_file.configuration_option()