Handle error when no passwords

nc_password_client/nc_password_client.py

@@ -112,6 +112,7 @@ class NextcloudHandler:
             self._log = params['logger']
         else:
             self._init_log(params)
+        self.params = params
         self.http = 'https'
         self.timeout = params.get('timeout', 3)
         self.ssl = True
@@ -314,10 +315,12 @@ class NextcloudHandler:
             self.reset_cache()
 
     def reset_cache(self):
+        '''Reset cached password'''
         self.cache = {
             "last_update": -1,
             "cached_passwords": []
         }
+
     def _write_cache(self):
         self.debug(
                 {
@@ -346,14 +349,34 @@ class NextcloudHandler:
         return obj
 
     def _output(self, obj, unsafe=False):
+        if 'timestamp' not in obj:
+            obj['timestamp'] = time.time()
         if unsafe:
             out_obj = obj
         else:
             out_obj = self._safer_obj(obj)
         if self.output_format == 'json':
-            return json.dumps(out_obj, indent=2)
+            try:
+                return json.dumps(out_obj, indent=2)
+            except json.decoder.JSONDecodeError as error:
+                self.error(
+                    {
+                        'action': '_outpout',
+                        'message': 'Error decoding JSON',
+                        'error': error,
+                    }
+                )
         if self.output_format == 'yaml':
-            return dump(out_obj, Dumper=Dumper)
+            try:
+                return dump(out_obj, Dumper=Dumper)
+            except Exception as error:
+                self.error(
+                    {
+                        'action': '_outpout',
+                        'message': 'Error decoding YAML',
+                        'error': error,
+                    }
+                )
         output = ''
         if isinstance(out_obj, list):
             output += '-' * os.get_terminal_size(0)[0]
@@ -369,6 +392,10 @@ class NextcloudHandler:
 
     def debug(self, obj, unsafe=False):
         '''Show debug information'''
+        obj['connection'] = {
+            "host": self.params['host'],
+            "user": self.params['user']
+        }
         self._log.debug(
             self._output(obj, unsafe=unsafe)
         )
@@ -422,7 +449,7 @@ class NextcloudHandler:
 
     def get(self, path):
         '''Do a GET request'''
-        self.debug({ "action": "get", "message": f"Requesting {path}" })
+        self.debug({ "action": "get", "message": f"Requesting '{self.http}://{self.host}/{path}'" })
         try:
             r = self.session.get(
                 f'{self.http}://{self.host}/{path}',
@@ -434,6 +461,8 @@ class NextcloudHandler:
                 self.debug(
                     {"action": "get", "status_code": r.status_code, "size": len(r.content)}
                 )
+                if r.headers['Content-Type'] == 'text/html; charset=UTF-8':
+                    return None
                 return r.json()
             if r.status_code == 404:
                 self.error(
@@ -751,7 +780,7 @@ class NextcloudHandler:
                     "folder_name": name
                 }
             )
-        return False
+        return None
 
     def exists_passwords_folder(self, name):
         '''Test if a passwords folder exists'''
@@ -760,6 +789,13 @@ class NextcloudHandler:
                 return folder
         return False
 
+    def _is_uuid(self, string):
+        '''Test if a string is a UUID'''
+        match = re.match(r'^[0-9a-f\-]*$', string)
+        if match:
+            return True
+        return False
+
     def create_passwords_folder(self, name):
         '''Create passwords folder'''
         if not self.exists_passwords_folder(name):
@@ -879,9 +915,14 @@ class NextcloudHandler:
             }
         )
         if new_obj.get('folder', '') != '':
-            folder_id = self.get_folder_id(new_obj['folder'])
+            if self._is_uuid(new_obj['folder']):
+                folder_id = new_obj['folder']
+            else:
+                folder_id = self.get_folder_id(new_obj['folder'])
             if not folder_id:
-                folder_id = self.create_passwords_folder(new_obj['folder'])['id']
+                created_folder = self.create_passwords_folder(new_obj['folder'])
+                if created_folder:
+                    folder_id = created_folder['id']
             new_obj['folder'] = folder_id
         else:
             new_obj.pop('folder', None)
@@ -930,9 +971,17 @@ class NextcloudHandler:
                     }
                 )
         else:
-            if update:
+            if update and 'id' in exists_password:
                 new_obj['id'] = exists_password['id']
                 return self.update_password(new_obj)
+            if 'id' not in exists_password:
+                self.debug(
+                    {
+                        "action": "create_password",
+                        "message": "Found an existing password without an 'id'",
+                        "existing_password": exists_password
+                    }
+                )
             self.warning(
                 {
                     "action": "create_password",
@@ -1150,9 +1199,11 @@ class NextcloudHandler:
 
     def get_folder_id(self, folder_name):
         '''Get a folder id from the name'''
-        for folder in self.list_passwords_folders():
+        passwords_folders = self.list_passwords()
-            if folder['label'] == folder_name:
+        if passwords_folders:
-                return folder['id']
+            for folder in passwords_folders:
+                if folder['label'] == folder_name:
+                    return folder['id']
         return False
 
 class NcPasswordClient:
@@ -1178,12 +1229,17 @@ class NcPasswordClient:
                 'nc_password_client.log'
             )
         self._init_log()
+        self.timeout = timeout
+        self.cache_duration = cache_duration
+        self.https_proxy = https_proxy
+        self.output_format = output_format
+        self.field_replacements = field_replacements
         params = {
             "host": host,
             "user": user,
             "api_token": api_token,
             "cse_password": cse_password,
-            "timeout": timeout,
+            "timeout": self.timeout,
             "cache_duration": cache_duration,
             "https_proxy": https_proxy,
             "logger": self._log,
@@ -1191,6 +1247,7 @@ class NcPasswordClient:
             "field_replacements": field_replacements,
         }
         self.nc = NextcloudHandler(params)
+        self.destination_nc = None
 
     def _safer_obj(self, obj, fields=None):
         if fields is None:
@@ -1573,6 +1630,35 @@ class NcPasswordClient:
         )
         return True
 
+    def migrate_passwords(
+        self,
+        destination_host,
+        destination_user,
+        destination_api_token,
+        destination_cse_password
+    ):
+        '''Migrate passwords from one NextCloud instance to another'''
+        params = {
+            "host": destination_host,
+            "user": destination_user,
+            "api_token": destination_api_token,
+            "cse_password": destination_cse_password,
+            "timeout": self.timeout,
+            "cache_duration": self.cache_duration,
+            "https_proxy": self.https_proxy,
+            "logger": self._log,
+            "output_format": self.output_format,
+            "field_replacements": self.field_replacements,
+        }
+        self.destination_nc = NextcloudHandler(params)
+        all_passwords = self.nc.list_passwords()
+        if all_passwords:
+            for password in all_passwords:
+                self.destination_nc.create_password(password, update=True)
+        else:
+            return False
+        return True
+
     def _init_log(self):
         ''' Initialize log object '''
         self._log = logging.getLogger("nc_password_client")
@@ -1671,6 +1757,37 @@ def cli(
         timeout, cache_duration, https_proxy, output_format, field_replacements
     )
 
+@cli.command()
+@click.option(
+    '--destination-host', '-D',
+    required=True,
+    help='Destination host'
+)
+@click.option(
+    '--destination-user', '-U',
+    required=True,
+    help='Destination host user name'
+)
+@click.option(
+    '--destination-api-token', '-A',
+    required=True,
+    help="Destination host user's API token "
+)
+@click.option(
+    '--destination-cse-password', '-C',
+    help='Destination Nextcloud host user\'s end-to-end encryption password'
+)
+@click_config_file.configuration_option()
+@click.pass_context
+def migrate_passwords(ctx, destination_host, destination_user, destination_api_token, destination_cse_password):
+    '''Migrate passwords between two NextCloud Password instances'''
+    ctx.obj['NcPasswordClient'].migrate_passwords(
+        destination_host,
+        destination_user,
+        destination_api_token,
+        destination_cse_password
+    )
+
 @cli.command()
 @click.option('--name', '-n', required=True, help='Name of the password to show')
 @click.option(

requirements.txt

@@ -5,3 +5,4 @@ pysodium
 passpy
 secretstorage
 pyyaml
+libsodium