feat(backend): ensure restore of backups only works in correct enviroments (#11372)

* [FR] ensure restore of backups only works in correct enviroments
Fixes #11214

* update PR nbr

* fix wrong ty detection

* fix link

* ensure tracing does not enagage while running backup ops

* fix import

* remove debugging string

* add error codes

* add tests for backup and restore

* complete test for restore

* we do not need e2e on every matrix entry
there is no realy db dep here

* fix changelog format

* add flag to allow bypass

src/backend/InvenTree/InvenTree/backup.py

@@ -5,7 +5,16 @@ We use the django-dbbackup library to handle backup and restore operations.
 Ref: https://archmonger.github.io/django-dbbackup/latest/configuration/
 """
 
+from datetime import datetime, timedelta
+
+from django.conf import settings
+
+import structlog
+
 import InvenTree.config
+import InvenTree.version
+
+logger = structlog.get_logger('inventree')
 
 
 def get_backup_connector_options() -> dict:
@@ -131,3 +140,131 @@ def backup_media_filename_template() -> str:
         default_value='InvenTree-media-{datetime}.{extension}',
         typecast=str,
     )
+
+
+# schema for backup metadata
+InvenTreeBackupMetadata = dict[str, str | int | bool | None]
+
+
+def _gather_environment_metadata() -> InvenTreeBackupMetadata:
+    """Gather metadata about the current environment to be stored with the backup."""
+    import plugin.installer
+
+    new_data: InvenTreeBackupMetadata = {}
+
+    new_data['ivt_1_debug'] = settings.DEBUG
+    new_data['ivt_1_version'] = InvenTree.version.inventreeVersion()
+    new_data['ivt_1_version_api'] = InvenTree.version.inventreeApiVersion()
+    new_data['ivt_1_plugins_enabled'] = settings.PLUGINS_ENABLED
+    new_data['ivt_1_plugins_file_hash'] = plugin.installer.plugins_file_hash()
+    new_data['ivt_1_installer'] = InvenTree.config.inventreeInstaller()
+    new_data['ivt_1_backup_time'] = datetime.now().isoformat()
+
+    return new_data
+
+
+def _parse_environment_metadata(metadata: InvenTreeBackupMetadata) -> dict[str, str]:
+    """Parse backup metadata to extract environment information."""
+    data = {}
+
+    data['debug'] = metadata.get('ivt_1_debug', False)
+    data['version'] = metadata.get('ivt_1_version', 'unknown')
+    data['version_api'] = metadata.get('ivt_1_version_api', 'unknown')
+    data['plugins_enabled'] = metadata.get('ivt_1_plugins_enabled', False)
+    data['plugins_file_hash'] = metadata.get('ivt_1_plugins_file_hash', 'unknown')
+    data['installer'] = metadata.get('ivt_1_installer', 'unknown')
+    data['backup_time'] = metadata.get('ivt_1_backup_time', 'unknown')
+
+    return data
+
+
+def metadata_set(metadata) -> InvenTreeBackupMetadata:
+    """Set backup metadata for the current backup operation."""
+    return _gather_environment_metadata()
+
+
+def validate_restore(metadata: InvenTreeBackupMetadata) -> bool | None:
+    """Validate whether a backup restore operation should proceed, based on the provided metadata."""
+    if metadata.get('ivt_1_version') is None:
+        logger.warning(
+            'INVE-W13: Backup metadata does not contain version information',
+            error_code='INVE-W13',
+        )
+        return True
+
+    current_environment = _parse_environment_metadata(_gather_environment_metadata())
+    backup_environment = _parse_environment_metadata(metadata)
+
+    # Version mismatch
+    if backup_environment['version'] != current_environment['version']:
+        logger.warning(
+            f'INVE-W13: Backup being restored was created with InvenTree version {backup_environment["version"]}, but current version is {current_environment["version"]}',
+            error_code='INVE-W13',
+        )
+
+        # Backup is from newer version - fail
+        try:
+            if int(backup_environment['version_api']) > int(
+                str(current_environment['version_api'])
+            ):
+                logger.error(
+                    'INVE-E16: Backup being restored was created with a newer version of InvenTree - restore cannot proceed. If you are using the invoke task for your restore this warning might be overridden once with `--restore-allow-newer-version`',
+                    error_code='INVE-E16',
+                )
+                # Check for pass flag to allow restore
+                if not settings.BACKUP_RESTORE_ALLOW_NEWER_VERSION:  # defaults to False
+                    return False
+                else:
+                    logger.warning(
+                        'INVE-W13: Backup restore is allowing a restore from a newer version of InvenTree - this can lead to data loss or corruption',
+                        error_code='INVE-W13',
+                    )
+        except ValueError:  # pragma: no cover
+            logger.warning(
+                'INVE-W13: Could not parse API version from backup metadata - cannot determine if backup is from newer version',
+                error_code='INVE-W13',
+            )
+
+    # Plugins enabled on backup but not restore environment - warn
+    if (
+        backup_environment['plugins_enabled']
+        and not current_environment['plugins_enabled']
+    ):
+        logger.warning(
+            'INVE-W13: Backup being restored was created with plugins enabled, but current environment has plugins disabled - this can lead to data loss',
+            error_code='INVE-W13',
+        )
+
+    # Plugins file hash mismatch - warn
+    if pg_hash := backup_environment['plugins_file_hash']:
+        if pg_hash != current_environment['plugins_file_hash']:
+            logger.warning(
+                'INVE-W13: Backup being restored has a different plugins file hash to the current environment - this can lead to data loss or corruption',
+                error_code='INVE-W13',
+            )
+
+    # Installer mismatch - warn
+    if installer := backup_environment['installer']:
+        if installer != current_environment['installer']:
+            logger.warning(
+                f"INVE-W13: Backup being restored was created with installer '{installer}', but current environment has installer '{current_environment['installer']}'",
+                error_code='INVE-W13',
+            )
+
+    # Age of backup
+    last_backup_time = backup_environment.get('backup_time')
+    if datetime.now() - datetime.fromisoformat(last_backup_time) > timedelta(days=120):
+        logger.warning(
+            f'INVE-W13: Backup being restored is over 120 days old (last backup time: {last_backup_time})',
+            error_code='INVE-W13',
+        )
+
+    if settings.DEBUG:  # pragma: no cover
+        logger.info(
+            f'INVE-I3: Backup environment: {backup_environment}', error_code='INVE-I3'
+        )
+        logger.info(
+            f'INVE-I3: Current environment: {current_environment}', error_code='INVE-I3'
+        )
+
+    return True

src/backend/InvenTree/InvenTree/settings.py

@@ -27,7 +27,7 @@ from corsheaders.defaults import default_headers as default_cors_headers
 import InvenTree.backup
 from InvenTree.cache import get_cache_config, is_global_cache_enabled
 from InvenTree.config import get_boolean_setting, get_oidc_private_key, get_setting
-from InvenTree.ready import isInMainThread
+from InvenTree.ready import isInMainThread, isRunningBackup
 from InvenTree.sentry import default_sentry_dsn, init_sentry
 from InvenTree.version import checkMinPythonVersion, inventreeCommitHash
 from users.oauth2_scopes import oauth2_scopes
@@ -258,12 +258,22 @@ DBBACKUP_EMAIL_SUBJECT_PREFIX = InvenTree.backup.backup_email_prefix()
 
 DBBACKUP_CONNECTORS = {'default': InvenTree.backup.get_backup_connector_options()}
 
+DBBACKUP_BACKUP_METADATA_SETTER = InvenTree.backup.metadata_set
+DBBACKUP_RESTORE_METADATA_VALIDATOR = InvenTree.backup.validate_restore
+
 # Data storage options
 DBBACKUP_STORAGE_CONFIG = {
     'BACKEND': InvenTree.backup.get_backup_storage_backend(),
     'OPTIONS': InvenTree.backup.get_backup_storage_options(),
 }
 
+# This can also be overridden with a command line flag --restore-allow-newer-version when running the restore command
+BACKUP_RESTORE_ALLOW_NEWER_VERSION = get_boolean_setting(
+    'INVENTREE_BACKUP_RESTORE_ALLOW_NEWER_VERSION',
+    'backup_restore_allow_newer_version',
+    False,
+)
+
 # Enable django admin interface?
 INVENTREE_ADMIN_ENABLED = get_boolean_setting(
     'INVENTREE_ADMIN_ENABLED', config_key='admin_enabled', default_value=True
@@ -862,7 +872,7 @@ TRACING_ENABLED = get_boolean_setting(
 )
 TRACING_DETAILS: Optional[dict] = None
 
-if TRACING_ENABLED:  # pragma: no cover
+if TRACING_ENABLED and not isRunningBackup():  # pragma: no cover
     from InvenTree.tracing import setup_instruments, setup_tracing
 
     _t_endpoint = get_setting('INVENTREE_TRACING_ENDPOINT', 'tracing.endpoint', None)

src/backend/InvenTree/InvenTree/test_commands.py

@@ -2,10 +2,15 @@
 
 from pathlib import Path
 
+from django.conf import settings
 from django.contrib.auth.models import User
 from django.core.management import call_command
 from django.test import TestCase
 
+from opentelemetry.instrumentation.sqlite3 import SQLite3Instrumentor
+
+from InvenTree.config import get_testfolder_dir
+
 
 class CommandTestCase(TestCase):
     """Test case for custom management commands."""
@@ -66,3 +71,144 @@ class CommandTestCase(TestCase):
         output = call_command('remove_mfa', username=my_admin3.username, verbosity=0)
         self.assertEqual(output, 'done')
         self.assertEqual(my_admin3.authenticator_set.all().count(), 0)
+
+    def test_backup_metadata(self):
+        """Test the backup metadata functions."""
+        from InvenTree.backup import (
+            _gather_environment_metadata,
+            _parse_environment_metadata,
+        )
+
+        metadata = _gather_environment_metadata()
+        self.assertIn('ivt_1_version', metadata)
+        self.assertIn('ivt_1_plugins_enabled', metadata)
+
+        parsed = _parse_environment_metadata(metadata)
+        self.assertIn('version', parsed)
+        self.assertIn('plugins_enabled', parsed)
+
+    def test_restore_validation(self):
+        """Test the restore validation functions."""
+        from InvenTree.backup import _gather_environment_metadata, validate_restore
+
+        metadata = _gather_environment_metadata()
+
+        self.assertTrue(validate_restore(metadata))
+
+        # Version
+        with self.assertLogs() as cm:
+            self.assertTrue(validate_restore({}))
+        self.assertIn(
+            'INVE-W13: Backup metadata does not contain version information', str(cm[1])
+        )
+        with self.assertLogs() as cm:
+            self.assertTrue(validate_restore({**metadata, 'ivt_1_version': '123xx'}))
+        self.assertIn(
+            'INVE-W13: Backup being restored was created with InvenTree version',
+            str(cm[1]),
+        )
+        with self.assertLogs() as cm:
+            self.assertFalse(
+                validate_restore({
+                    **metadata,
+                    'ivt_1_version': '9999',
+                    'ivt_1_version_api': '9999',
+                })
+            )
+        self.assertIn(
+            'INVE-E16: Backup being restored was created with a newer version',
+            str(cm[1]),
+        )
+        # not with allow flag
+        with self.settings(BACKUP_RESTORE_ALLOW_NEWER_VERSION=True):
+            with self.assertLogs() as cm:
+                self.assertTrue(
+                    validate_restore({
+                        **metadata,
+                        'ivt_1_version': '9999',
+                        'ivt_1_version_api': '9999',
+                    })
+                )
+            self.assertIn(
+                'INVE-W13: Backup restore is allowing a restore from a newer version of InvenTree',
+                str(cm[1]),
+            )
+
+        # Plugins enabled
+        with self.settings(PLUGINS_ENABLED=False):
+            with self.assertLogs() as cm:
+                self.assertTrue(validate_restore(metadata))
+        self.assertIn(
+            'INVE-W13: Backup being restored was created with plugins enabled',
+            str(cm[1]),
+        )
+
+        # Plugin hash
+        with self.assertLogs() as cm:
+            self.assertTrue(
+                validate_restore({**metadata, 'ivt_1_plugins_file_hash': '123xx'})
+            )
+        self.assertIn(
+            'INVE-W13: Backup being restored has a different plugins file hash',
+            str(cm[1]),
+        )
+
+        # installer
+        with self.assertLogs() as cm:
+            self.assertTrue(validate_restore({**metadata, 'ivt_1_installer': '123xx'}))
+        self.assertIn(
+            'INVE-W13: Backup being restored was created with installer', str(cm[1])
+        )
+
+        # Age
+        with self.assertLogs() as cm:
+            self.assertTrue(
+                validate_restore({
+                    **metadata,
+                    'ivt_1_backup_time': '2020-02-02T00:00:00',
+                })
+            )
+        self.assertIn(
+            'INVE-W13: Backup being restored is over 120 days old', str(cm[1])
+        )
+
+    def test_backup_command_e2e(self):
+        """Test the backup command."""
+        # we only test on sqlite, the environment in which we also run coverage
+        if settings.DB_ENGINE != 'django.db.backends.sqlite3':
+            self.skipTest('Backup command test only runs on sqlite database')
+
+        # disable tracing for now
+        if settings.TRACING_ENABLED:  # pragma: no cover
+            print('Disabling tracing for backup command test')
+            SQLite3Instrumentor().uninstrument()
+
+        output_path = get_testfolder_dir().joinpath('backup.zip').resolve()
+
+        # Backup
+        with self.assertLogs() as cm:
+            output = call_command(
+                'dbbackup', noinput=True, verbosity=2, output_path=str(output_path)
+            )
+            self.assertIsNone(output)
+        self.assertIn(f'Writing metadata file to {output_path}', str(cm[1]))
+
+        # Restore
+        with self.assertLogs() as cm:
+            output = call_command(
+                'dbrestore',
+                noinput=True,
+                interactive=False,
+                verbosity=2,
+                input_path=str(output_path),
+            )
+            self.assertIsNone(output)
+        self.assertIn('Using connector from metadata', str(cm[1]))
+
+        # Cleanup the generated backup file and metadata file
+        output_path.unlink()
+        Path(str(output_path) + '.metadata').unlink()
+
+        if settings.TRACING_ENABLED:  # pragma: no cover
+            print('Re-enabling tracing for backup command test')
+            SQLite3Instrumentor().instrument()