diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e73442e..c147755e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,8 @@ Don't forget to remove deprecated code on each major release! ## [Unreleased] -- Nothing (yet)! +### Added +- Added support for custom metadata writing and validation during operations via `DBBACKUP_BACKUP_METADATA_SETTER` and `DBBACKUP_RESTORE_METADATA_VALIDATOR` settings. ## [5.1.2] - 2026-01-14 diff --git a/dbbackup/management/commands/dbbackup.py b/dbbackup/management/commands/dbbackup.py index 3f83bff5..443eebbc 100644 --- a/dbbackup/management/commands/dbbackup.py +++ b/dbbackup/management/commands/dbbackup.py @@ -101,6 +101,11 @@ def _save_metadata(self, filename, local=False): "connector": f"{self.connector.__module__}.{self.connector.__class__.__name__}", } metadata_filename = f"{filename}.metadata" + + # Load custom metadata if configured + user_metadata = utils.get_user_metadata(metadata) + metadata.update(user_metadata) + metadata_content = json.dumps(metadata) if local: diff --git a/dbbackup/management/commands/dbrestore.py b/dbbackup/management/commands/dbrestore.py index 5ba4f1ad..cdd80110 100644 --- a/dbbackup/management/commands/dbrestore.py +++ b/dbbackup/management/commands/dbrestore.py @@ -153,6 +153,12 @@ def _check_metadata(self, filename): ) raise CommandError(msg) + # Check if we custom metadata validation is configured + user_metadata_valid = utils.validate_user_metadata(metadata) + if not user_metadata_valid: + msg = f"Custom metadata validation failed for backup file '{filename}'." + raise CommandError(msg) + return metadata def _restore_backup(self): diff --git a/dbbackup/settings.py b/dbbackup/settings.py index 2689863b..c97f3f10 100644 --- a/dbbackup/settings.py +++ b/dbbackup/settings.py @@ -42,3 +42,5 @@ SERVER_EMAIL = getattr(settings, "DBBACKUP_SERVER_EMAIL", settings.SERVER_EMAIL) ADMINS = getattr(settings, "DBBACKUP_ADMIN", settings.ADMINS) EMAIL_SUBJECT_PREFIX = getattr(settings, "DBBACKUP_EMAIL_SUBJECT_PREFIX", "[dbbackup] ") +BACKUP_METADATA_SETTER = getattr(settings, "DBBACKUP_BACKUP_METADATA_SETTER", None) +RESTORE_METADATA_VALIDATOR = getattr(settings, "DBBACKUP_RESTORE_METADATA_VALIDATOR", None) diff --git a/dbbackup/utils.py b/dbbackup/utils.py index d6c5f761..238e1c18 100644 --- a/dbbackup/utils.py +++ b/dbbackup/utils.py @@ -2,7 +2,11 @@ Utility functions for dbbackup. """ +from __future__ import annotations + +import copy import gzip +import json import logging import os import re @@ -12,6 +16,7 @@ from datetime import datetime from functools import wraps from getpass import getpass +from importlib import import_module from shutil import copyfileobj from django.core.mail import EmailMultiAlternatives @@ -431,3 +436,93 @@ def filename_generate(extension, database_name="", servername=None, content_type filename = REG_FILENAME_CLEAN.sub("-", filename) filename = filename.removeprefix("-") return filename + + +def _get_function_from_path(func_or_path): + """ + Load a callable from a dotted path or return the callable itself. + + :param func_or_path: Dotted path to callable or callable itself + :type func_or_path: str | callable + + :returns: Callable object + :rtype: callable + """ + if callable(func_or_path): + return func_or_path + + path = func_or_path + module_path, func_name = path.rsplit(".", 1) + try: + module = import_module(module_path) + except ImportError as e: + msg = f"Could not import module '{module_path}': {e}" + raise ImportError(msg) from e + func = getattr(module, func_name) + if not callable(func): + msg = f"The object at '{path}' is not callable." + raise TypeError(msg) + return func + + +def get_user_metadata(metadata: dict | None = None) -> dict: + """ + Get user generated metadata from the user's custom metadata setter. + + :returns: Custom metadata dictionary + :rtype: dict + """ + user_metadata = {} + setter_setting = settings.BACKUP_METADATA_SETTER + if setter_setting: + setter_function = _get_function_from_path(setter_setting) + try: + # We pass a copy to avoid side effects + user_metadata = setter_function(copy.deepcopy(metadata)) + except Exception: + logger = logging.getLogger("dbbackup") + logger.exception("Error loading custom metadata: %s") + + if user_metadata is None: + user_metadata = {} + + if not isinstance(user_metadata, dict): + msg = "DBBACKUP_BACKUP_METADATA_SETTER must return a dictionary." + raise ValueError(msg) + + # Validate that we can serialize the provided data + try: + json.dumps(user_metadata) + except Exception as e: + msg = f"Custom metadata is not JSON serializable: {e}" + raise ValueError(msg) from e + return user_metadata + + +def validate_user_metadata(metadata) -> bool | None: + """ + Validate custom metadata using a callable defined in `settings.py`. + Raise a CommandError to provide custom feedback if validation fails. + + :param metadata: Metadata dictionary to validate + :type metadata: dict + + :returns: True if validation passes (or None), False otherwise + :rtype: Optional[bool] + """ + validator_setting = settings.RESTORE_METADATA_VALIDATOR + if validator_setting: + validator_function = _get_function_from_path(validator_setting) + try: + # We pass a copy to avoid side effects + user_metadata = validator_function(copy.deepcopy(metadata)) + except Exception as e: + msg = f"Error during custom metadata validation: {e}" + raise ValueError(msg) from e + if user_metadata is None: + return None + if not isinstance(user_metadata, bool): + msg = "DBBACKUP_RESTORE_METADATA_VALIDATOR must return a boolean or None." + raise TypeError(msg) + return user_metadata + return True diff --git a/docs/src/configuration.md b/docs/src/configuration.md index b2e6af55..4d405a41 100644 --- a/docs/src/configuration.md +++ b/docs/src/configuration.md @@ -123,9 +123,9 @@ python manage.py dbrestore --decrypt Requirements: -- Install the python package python-gnupg: `pip install python-gnupg>=0.5.0`. -- You need a GPG key. ([GPG manual](https://www.gnupg.org/gph/en/manual/c14.html)) -- Set the setting `DBBACKUP_GPG_RECIPIENT` to the name of the GPG key. +- Install the python package python-gnupg: `pip install python-gnupg>=0.5.0`. +- You need a GPG key. ([GPG manual](https://www.gnupg.org/gph/en/manual/c14.html)) +- Set the setting `DBBACKUP_GPG_RECIPIENT` to the name of the GPG key. Note (Windows): The `gpg` executable must be installed and on your PATH for encryption/decryption. If it is absent, django-dbbackup still works; only encryption-related features are unavailable. The test suite will automatically skip encryption tests when `gpg` is not found. @@ -204,3 +204,53 @@ By default DBBackup uses values from `settings.DATABASES`. Use You must configure a storage backend (`STORAGES['dbbackup']`) to persist backups. See [Storage settings](storage.md) for supported options. + +## Custom metadata + +### DBBACKUP_BACKUP_METADATA_SETTER + +Function or dotted path (string) to a callable that returns a dictionary of extra metadata. + +The function must accept a single argument: a metadata dictionary containing information about the current backup operation. If a dictionary is returned, it will be merged into the current backup's metadata file. If `None` is returned, it is treated as an empty dictionary. + +Default: `None` + +```python +def metadata_set(metadata): + last_backup_time = datetime.now().isoformat() + if metadata and metadata.get('engine') == 'my.custom.Engine': + region = os.environ.get('AWS_REGION') + return {'environment': 'AWS', 'region': region, 'backup_time': last_backup_time} + else: + return {'backup_time': last_backup_time} + +DBBACKUP_BACKUP_METADATA_SETTER = metadata_set +``` + +### DBBACKUP_RESTORE_METADATA_VALIDATOR + +Function or dotted path (string) to a callable that performs additional validation on the backup's metadata during restore operations. This validator runs **after** the built-in validation (e.g. database engine checks). + +The callable should accept a single argument: the metadata dictionary loaded from the current restore operation's metadata file. If this function returns `True`, the metadata is valid, `False` means the restore operation will be aborted. `None` can be returned to do nothing and allow `dbbackup` to decide. It may raise a `CommandError` with a descriptive message if validation fails. + +Default: `None` + +```python +import os + +def validate_restore(metadata): + region = os.environ.get('AWS_REGION') + if not metadata.get('region') == region: + raise CommandError(f"Backup region does not match current region {region}; cross region restores are not allowed due to SMP-0123.") + + last_backup_time = metadata.get('backup_time') + if datetime.now() - datetime.fromisoformat(last_backup_time) > timedelta(days=120): + return False # Skip restore since it is stale (>120 days) + + if os.environ.get('AWS_REGION') == None: + return None # Do nothing if performing a backup outside AWS_REGION + + return True + +DBBACKUP_RESTORE_METADATA_VALIDATOR = validate_restore +``` diff --git a/docs/src/dictionary.txt b/docs/src/dictionary.txt index 08e7ba8d..a0a6bb38 100644 --- a/docs/src/dictionary.txt +++ b/docs/src/dictionary.txt @@ -48,3 +48,5 @@ postgis postgresql gzip tarfile +validator +no-op diff --git a/pyproject.toml b/pyproject.toml index fa0cc1ce..c2fc7eb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -119,6 +119,8 @@ matrix.django.dependencies = [ # >>> Documentation Scripts <<< [tool.hatch.envs.docs] +installer = "uv" +python = "3.11" # `editdistpy` (v0.1.6) is a dependency of `mkdocs-spellcheck[all]`, and is incompatible with Python 3.12+. template = "docs" detached = true dependencies = [ diff --git a/tests/commands/test_dbrestore_metadata.py b/tests/commands/test_dbrestore_metadata.py index a8804078..d0a22935 100644 --- a/tests/commands/test_dbrestore_metadata.py +++ b/tests/commands/test_dbrestore_metadata.py @@ -1,3 +1,4 @@ +import importlib import json from unittest.mock import Mock, patch @@ -5,7 +6,9 @@ from django.conf import settings from django.core.management.base import CommandError from django.test import TestCase +from django.test.utils import override_settings +from dbbackup.management.commands import dbbackup from dbbackup.management.commands.dbrestore import Command as DbrestoreCommand @@ -83,6 +86,55 @@ def test_django_connector_mismatch_allowed(self): # Should not raise self.command._check_metadata("backup.dump") + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator") + def test_metadata_match_custom(self): + importlib.reload(dbbackup.settings) + # Setup metadata + metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "aabbcc-1122-3344_eu-west"} + self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata)) + + # Should not raise + self.command._check_metadata("backup.dump") + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator") + def test_metadata_match_custom_fail(self): + importlib.reload(dbbackup.settings) + # Setup metadata + metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "aabbcc-1122-3344_eu-ttt"} + self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata)) + + # Should raise CommandError due to validation failure + with pytest.raises(CommandError): + self.command._check_metadata("backup.dump") + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator") + def test_metadata_match_custom_failwithcustomerror(self): + importlib.reload(dbbackup.settings) + # Setup metadata + metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "xx-1122-3344_eu-west"} + self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata)) + + # Should raise CommandError due to validation failure + with pytest.raises(ValueError, match="CSMT_VAL must start with 'aabbcc'"): + self.command._check_metadata("backup.dump") + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator") + def test_metadata_mismatch_custom_valid(self): + """Test that built-in validation runs before custom validation.""" + importlib.reload(dbbackup.settings) + # Setup metadata with different engine (primary fail) but valid custom data + metadata = { + "engine": "django.db.backends.postgresql", + "CSMT_VAL": "aabbcc-1122-3344_eu-west" + } + self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata)) + + # Should raise CommandError due to engine mismatch (primary), not custom validation + with pytest.raises(CommandError) as cm: + self.command._check_metadata("backup.dump") + + assert "Restoring to a different database engine is not supported" in str(cm.value) + class DbrestoreConnectorOverrideTest(TestCase): def setUp(self): diff --git a/tests/test_user_metadata.py b/tests/test_user_metadata.py new file mode 100644 index 00000000..7fa84e65 --- /dev/null +++ b/tests/test_user_metadata.py @@ -0,0 +1,186 @@ +"""Tests for custom metadata.""" + +import importlib + +import pytest +from django.test import TestCase, override_settings + +import dbbackup.settings +import dbbackup.utils + +# Helpers +TEST_VAL = "aabbcc-1122-3344_eu-west" + + +def dummy_setter(metadata): + return {"CSMT_VAL": TEST_VAL} + + +def broken_setter(metadata): + return ["not", "a", "dict"] + + +def setter_returning_none(metadata): + return None + + +def mutation_setter(context): + context["mutated"] = True + return {} + + +def mutation_validator(metadata): + metadata["mutated"] = True + return True + + +class CSTM: + abc = "123" + + +def anotherbroken_setter(metadata): + return {"CSMT_VAL": TEST_VAL, "CSTM_OBJ": CSTM()} + + +def dummy_validator(metadata): + """Dummy validator for testing purposes that assues we only allow one type of customer and region. + + A real validator would probably do more complex checks. + """ + if metadata.get("NO-OP", False): + return None + if val := metadata.get("CSMT_VAL", ""): + if not val.startswith("aabbcc"): + raise ValueError("CSMT_VAL must start with 'aabbcc'") + if not val.endswith("eu-west"): + return False + return True + + +def broken_validator(metadata): + print(1 / 0) # Always raises ZeroDivisionError + return True + +def non_bool_validator(metadata): + return "not a boolean" + +DEFAULT_META = {'ENGINE': 'django.db.backends.sqlite3'} + +# Actual tests +class CustomMetadataTest(TestCase): + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.dummy_setter") + def test_metadata_setter_valid(self): + """Test that setting DBBACKUP_BACKUP_METADATA_SETTER works.""" + importlib.reload(dbbackup.settings) + + assert dbbackup.settings.BACKUP_METADATA_SETTER == "tests.test_user_metadata.dummy_setter" + assert dbbackup.utils.get_user_metadata(DEFAULT_META) == {"CSMT_VAL": TEST_VAL} + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER=dummy_setter) + def test_metadata_setter_valid_callable(self): + """Test that setting DBBACKUP_BACKUP_METADATA_SETTER works with a callable.""" + importlib.reload(dbbackup.settings) + + assert dbbackup.settings.BACKUP_METADATA_SETTER == dummy_setter + assert dbbackup.utils.get_user_metadata(DEFAULT_META) == {"CSMT_VAL": TEST_VAL} + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.setter_returning_none") + def test_metadata_setter_none(self): + """Test that setting DBBACKUP_BACKUP_METADATA_SETTER works when returning None.""" + importlib.reload(dbbackup.settings) + + assert dbbackup.utils.get_user_metadata(DEFAULT_META) == {} + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.mutation_setter") + def test_metadata_setter_no_side_effects(self): + """Test that the setter cannot mutate the original context.""" + importlib.reload(dbbackup.settings) + context = {"original": True} + dbbackup.utils.get_user_metadata(context) + assert context == {"original": True} + assert "mutated" not in context + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="non.existent.loader") + def test_metadata_setter_invalid_1(self): + """Test that various setter missconfigurations do not work - Non-existent module.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(ImportError, match="Could not import module 'non.existent': No module named 'non'"): + dbbackup.utils.get_user_metadata(DEFAULT_META) + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.TEST_VAL") + def test_metadata_setter_invalid_2(self): + """Test that various setter missconfigurations do not work - Non-callable object.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(TypeError, match="The object at 'tests.test_user_metadata.TEST_VAL' is not callable."): + dbbackup.utils.get_user_metadata(DEFAULT_META) + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.broken_setter") + def test_metadata_setter_invalid_3(self): + """Test that various setter missconfigurations do not work - Wrong return type.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(ValueError, match="DBBACKUP_BACKUP_METADATA_SETTER must return a dictionary."): + dbbackup.utils.get_user_metadata(DEFAULT_META) + + @override_settings(DBBACKUP_BACKUP_METADATA_SETTER="tests.test_user_metadata.anotherbroken_setter") + def test_metadata_setter_invalid_4(self): + """Test that various setter missconfigurations do not work - Not JSON serializable.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(ValueError, match="Custom metadata is not JSON serializable"): + dbbackup.utils.get_user_metadata(DEFAULT_META) + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator") + def test_metadata_validator_valid(self): + """Test that setting DBBACKUP_RESTORE_METADATA_VALIDATOR works.""" + importlib.reload(dbbackup.settings) + + assert dbbackup.settings.RESTORE_METADATA_VALIDATOR == "tests.test_user_metadata.dummy_validator" + assert dbbackup.utils.validate_user_metadata({"CSMT_VAL": TEST_VAL}) is True + + # Test that a validator can raise an exception + with pytest.raises(ValueError, match="CSMT_VAL must start with 'aabbcc'"): + dbbackup.utils.validate_user_metadata({"CSMT_VAL": "123"}) + + # Test that a wrong also raises a ValueError + assert dbbackup.utils.validate_user_metadata({"CSMT_VAL": "aabbcc-xyz_eu-central"}) is False + + # Test that no-op works + assert dbbackup.utils.validate_user_metadata({"NO-OP": True}) is None + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.mutation_validator") + def test_metadata_validator_no_side_effects(self): + """Test that the validator cannot mutate the original metadata.""" + importlib.reload(dbbackup.settings) + metadata = {"original": True} + assert dbbackup.utils.validate_user_metadata(metadata) is True + assert metadata == {"original": True} + assert "mutated" not in metadata + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="non.existent.validator") + def test_metadata_validator_invalid_1(self): + """Test that various validator missconfigurations do not work - Non-existent module.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(ImportError, match="Could not import module 'non.existent': No module named 'non'"): + dbbackup.utils.validate_user_metadata({"CSMT_VAL": TEST_VAL}) + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.broken_validator") + def test_metadata_validator_invalid_2(self): + """Test that various validator missconfigurations do not work - Exception during validation.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(ValueError, match="Error during custom metadata validation:"): + assert dbbackup.utils.validate_user_metadata({"CSMT_VAL": TEST_VAL}) is False + assert False, "Should not reach this point" + + @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.non_bool_validator") + def test_metadata_validator_invalid_3(self): + """Test that various validator missconfigurations do not work - Exception during validation.""" + importlib.reload(dbbackup.settings) + + with pytest.raises(TypeError, match="DBBACKUP_RESTORE_METADATA_VALIDATOR must return a boolean or None"): + assert dbbackup.utils.validate_user_metadata({"CSMT_VAL": TEST_VAL}) is False + assert False, "Should not reach this point"