KC-1151: Add custom-field fallback for no file storage

docker_ksm_utility.py

@@ -306,6 +306,10 @@ def _get_secret_by_uid_or_title(secrets_manager, record_identifier):
 def download_config(ksm_config_path, ksm_token, record_identifier, output_path):
     """
     Download config.json from KSM record.
+
+    Tries file attachments first.  If no config.json attachment exists
+    (e.g. the account has no file storage plan), falls back to the
+    ``config_json`` custom field written by the setup commands.
 
     Args:
         ksm_config_path (str): Path to KSM config file (optional)
@@ -344,15 +348,24 @@ def download_config(ksm_config_path, ksm_token, record_identifier, output_path):
         if not secret:
             return False
 
-        # Find config.json attachment
+        # Try file attachment first
         for file in secret.files:
             if file.name.lower() == 'config.json':
-                # Ensure output directory exists
                 os.makedirs(os.path.dirname(output_path), exist_ok=True)
                 file.save_file(output_path, True)
                 return True
 
-        print("ERROR: config.json attachment not found in record")
+        # Fallback: check for config_json custom field
+        config_value = secret.custom_field('config_json', single=True)
+        if config_value:
+            os.makedirs(os.path.dirname(output_path), exist_ok=True)
+            with open(output_path, 'w', encoding='utf-8') as f:
+                f.write(config_value)
+            os.chmod(output_path, 0o600)
+            print("config.json restored from custom field")
+            return True
+
+        print("ERROR: config.json not found as attachment or custom field in record")
         return False
 
     except Exception as e:

keepercommander/service/config/cli_handler.py

@@ -85,10 +85,44 @@ def get_help_output(self, params: KeeperParams) -> str:
     def download_config_from_vault(self, params: KeeperParams, title: str, config_dir: Path) -> bool:
         """Download config file from vault if it exists."""
         try:
-            if record_uid := self.find_config_record(params, title):
-                self.execute_cli_command(params, f"download-attachment {record_uid} --out-dir '{config_dir}'")
+            record_uid = self.find_config_record(params, title)
+            if not record_uid:
+                return False
+
+            self.execute_cli_command(params, f"download-attachment {record_uid} --out-dir '{config_dir}'")
+
+            json_path = config_dir / 'service_config.json'
+            yaml_path = config_dir / 'service_config.yaml'
+            if json_path.exists() or yaml_path.exists():
                 return True
-            return False
+
+            return self._restore_config_from_custom_field(params, record_uid, config_dir)
         except Exception as e:
             logger.error(f"Error downloading config from vault: {e}")
+            return False
+
+    def _restore_config_from_custom_field(self, params: KeeperParams, record_uid: str, config_dir: Path) -> bool:
+        """Write service_config content from a custom field to disk."""
+        try:
+            from ... import vault
+            record = vault.KeeperRecord.load(params, record_uid)
+            if not isinstance(record, vault.TypedRecord) or not record.custom:
+                return False
+
+            field_map = {
+                'service_config_json': config_dir / 'service_config.json',
+                'service_config_yaml': config_dir / 'service_config.yaml',
+            }
+            for field in record.custom:
+                if field.label in field_map and field.get_default_value():
+                    dest = field_map[field.label]
+                    dest.write_text(field.get_default_value())
+                    from ... import utils
+                    utils.set_file_permissions(str(dest))
+                    logger.debug(f"Restored {dest.name} from custom field")
+                    return True
+
+            return False
+        except Exception as e:
+            logger.error(f"Error restoring config from custom field: {e}")
             return False

keepercommander/service/config/record_handler.py

@@ -25,6 +25,21 @@ def __init__(self):
         self.validator = ConfigValidator()
         self.cli_handler = CommandHandler()
 
+    @staticmethod
+    def has_file_storage(params) -> bool:
+        """Check whether the current user can upload file attachments."""
+        if not params.license or 'bytes_total' not in params.license:
+            return False
+        if int(params.license['bytes_total']) <= 0:
+            return False
+        if params.enforcements and 'booleans' in params.enforcements:
+            restricted = next(
+                (x['value'] for x in params.enforcements['booleans']
+                 if x['key'] == 'restrict_file_upload'), False)
+            if restricted:
+                return False
+        return True
+
     @debug_decorator
     def create_record(self, is_advanced_security_enabled: str, commands: str, token_expiration: str = None, record_uid: str = None) -> Dict[str, Any]:
         """Create a new configuration record."""
@@ -50,6 +65,13 @@ def create_record(self, is_advanced_security_enabled: str, commands: str, token_
 
     def update_or_add_record(self, params: KeeperParams, title: str, config_path: Path) -> None:
         """Update existing record or add new one."""
+        if self.has_file_storage(params):
+            self._update_or_add_record_attachment(params, title, config_path)
+        else:
+            self._update_or_add_record_custom_field(params, title, config_path)
+
+    def _update_or_add_record_attachment(self, params: KeeperParams, title: str, config_path: Path) -> None:
+        """Upload service_config as a file attachment (original behaviour)."""
         try:
             record_uid = self.cli_handler.find_config_record(params, title)
 
@@ -74,6 +96,51 @@ def update_or_add_record(self, params: KeeperParams, title: str, config_path: Pa
         except Exception as e:
             print(f"Error updating/adding record: {e}")
 
+    def _update_or_add_record_custom_field(self, params: KeeperParams, title: str, config_path: Path) -> None:
+        """Store service_config content as a custom field (no file storage plan)."""
+        try:
+            from ... import api, vault, record_management
+
+            config_content = config_path.read_text()
+            field_label = f'service_config_{config_path.suffix.lstrip(".")}'
+
+            record_uid = self.cli_handler.find_config_record(params, title)
+
+            if record_uid:
+                record = vault.KeeperRecord.load(params, record_uid)
+            else:
+                record = vault.KeeperRecord.create(params, 'login')
+                record.record_uid = utils.generate_uid()
+                record.record_key = utils.generate_aes_key()
+                record.title = title
+                record.type_name = 'login'
+                record_management.add_record_to_folder(params, record)
+                api.sync_down(params)
+
+            if not isinstance(record, vault.TypedRecord):
+                print("Error: Invalid record type for custom field storage")
+                return
+
+            if record.custom is None:
+                record.custom = []
+            record.custom = [
+                f for f in record.custom
+                if f.label not in ('service_config_json', 'service_config_yaml')
+            ]
+            record.custom.append(vault.TypedField.new_field('secret', config_content, field_label))
+
+            record_management.update_record(params, record)
+            params.sync_data = True
+            api.sync_down(params)
+
+            if not record_uid:
+                self.record_uid = record.record_uid
+
+            logger.debug(f"Service config stored as custom field '{field_label}' (no file storage plan)")
+
+        except Exception as e:
+            print(f"Error storing service config as custom field: {e}")
+
     def update_or_add_cert_record(self, params: KeeperParams, title: str) -> None:
         """Update existing certificate record or add a new one in Keeper Vault."""
         try:

keepercommander/service/docker/setup_base.py

@@ -33,6 +33,7 @@
 from .models import SetupResult, SetupStep, DockerSetupConstants
 from .printer import DockerSetupPrinter
 from ..config.config_validation import ConfigValidator, ValidationError
+from ..config.record_handler import RecordHandler
 
 
 class DockerSetupBase:
@@ -62,8 +63,8 @@ def run_setup_steps(self, params, folder_name: str, app_name: str, record_name:
         DockerSetupPrinter.print_step(SetupStep.CREATE_RECORD.value, total_steps, f"Creating record '{record_name}'...")
         record_uid = self._create_config_record(params, record_name, folder_uid)
 
-        # Step 4: Upload config file
-        DockerSetupPrinter.print_step(SetupStep.UPLOAD_CONFIG.value, total_steps, "Uploading config.json attachment...")
+        # Step 4: Store config file (attachment or custom field)
+        DockerSetupPrinter.print_step(SetupStep.UPLOAD_CONFIG.value, total_steps, "Storing config.json...")
         self._upload_config_file(params, record_uid, config_path)
 
         # Step 5: Create KSM app
@@ -180,41 +181,65 @@ def _create_config_record(self, params, record_name: str, folder_uid: str) -> st
             raise CommandError('docker-setup', f'Failed to create record: {str(e)}')
 
     def _upload_config_file(self, params, record_uid: str, config_path: str) -> None:
-        """Upload config.json as attachment to the record"""
+        """Upload config.json as attachment, or store as custom field if no file storage plan."""
         temp_config_path = None
         try:
-            # Clean the config first
             cleaned_config_path = self._clean_config_json(config_path)
             if cleaned_config_path != config_path:
                 temp_config_path = cleaned_config_path
-
-            record = vault.KeeperRecord.load(params, record_uid)
-            if not isinstance(record, vault.TypedRecord):
-                raise CommandError('docker-setup', 'Invalid record type for attachments')
-            # Delete existing config.json attachments to prevent duplicates
-            self._delete_existing_config_attachments(record, params)
-
-            # Upload attachment
-            upload_task = attachment.FileUploadTask(cleaned_config_path)
-            upload_task.title = 'config.json'
-
-            attachment.upload_attachments(params, record, [upload_task])
-            record_management.update_record(params, record)
-            params.sync_data = True
-            api.sync_down(params)
-
-            DockerSetupPrinter.print_success("Config file uploaded successfully")
+
+            if RecordHandler.has_file_storage(params):
+                self._upload_config_as_attachment(params, record_uid, cleaned_config_path)
+            else:
+                self._store_config_as_custom_field(params, record_uid, cleaned_config_path)
+        except CommandError:
+            raise
         except Exception as e:
-            raise CommandError('docker-setup', f'Failed to upload config file: {str(e)}')
+            raise CommandError('docker-setup', f'Failed to store config file: {str(e)}')
         finally:
             if temp_config_path and os.path.exists(temp_config_path):
                 try:
                     os.unlink(temp_config_path)
-                except OSError as e:
-                    # Log or handle specifically
-                    print(f"Warning: Could not delete temporary config file: {e}")
+                except OSError:
                     pass
 
+    def _upload_config_as_attachment(self, params, record_uid: str, config_path: str) -> None:
+        """Upload config.json as a file attachment (requires file storage plan)."""
+        record = vault.KeeperRecord.load(params, record_uid)
+        if not isinstance(record, vault.TypedRecord):
+            raise CommandError('docker-setup', 'Invalid record type for attachments')
+        self._delete_existing_config_attachments(record, params)
+
+        upload_task = attachment.FileUploadTask(config_path)
+        upload_task.title = 'config.json'
+
+        attachment.upload_attachments(params, record, [upload_task])
+        record_management.update_record(params, record)
+        params.sync_data = True
+        api.sync_down(params)
+
+        DockerSetupPrinter.print_success("Config file uploaded as attachment")
+
+    def _store_config_as_custom_field(self, params, record_uid: str, config_path: str) -> None:
+        """Store config.json content as a custom field (fallback when no file storage plan)."""
+        with open(config_path, 'r') as f:
+            config_content = f.read()
+
+        record = vault.KeeperRecord.load(params, record_uid)
+        if not isinstance(record, vault.TypedRecord):
+            raise CommandError('docker-setup', 'Invalid record type')
+
+        if record.custom is None:
+            record.custom = []
+        record.custom = [f for f in record.custom if f.label != 'config_json']
+        record.custom.append(vault.TypedField.new_field('secret', config_content, 'config_json'))
+
+        record_management.update_record(params, record)
+        params.sync_data = True
+        api.sync_down(params)
+
+        DockerSetupPrinter.print_success("Config stored as custom field (no file storage plan)")
+
     def _delete_existing_config_attachments(self, record, params) -> None:
         """Delete any existing config.json attachments to prevent duplicates"""
         # Modern records use TypedRecord with fileRef system