Add support for AWS EBS Snapshot Lock mode

barman/clients/cloud_backup.py

@@ -45,7 +45,7 @@
     UnrecoverableHookScriptError,
 )
 from barman.postgres import PostgreSQLConnection
-from barman.utils import check_backup_name, check_positive, check_size, force_str
+from barman.utils import check_aws_snapshot_lock_duration_range, check_aws_snapshot_lock_cool_off_period_range, check_backup_name, check_positive, check_size, check_timestamp, force_str
 
 _find_space = re.compile(r"[\s]").search
 
@@ -419,6 +419,26 @@ def parse_arguments(args=None):
         "timing out (default: 3600 seconds)",
         type=check_positive,
     )
+    s3_arguments.add_argument(
+        "--aws-snapshot-lock-mode",
+        help="The lock mode to apply to the snapshot. Allowed values: 'governance'|'compliance'.",
+        choices=["governance", "compliance"],
+    )
+    s3_arguments.add_argument(
+        "--aws-snapshot-lock-duration",
+        help="The duration (in days) for which the snapshot should be locked. Must be between 1 and 36500. To lock a snapshopt, you must specify either this argument or --aws-snapshot-lock-expiration-date, but not both.",
+        type=check_aws_snapshot_lock_duration_range,
+    )
+    s3_arguments.add_argument(
+        "--aws-snapshot-lock-cool-off-period",
+        help="Specifies the cool-off period (in hours) for a snapshot locked in 'compliance' mode, allowing you to unlock or modify lock settings after it is locked. Must be between 1 and 72 hours. To lock the snapshot immediately without a cool-off period, leave this option unset.",
+        type=check_aws_snapshot_lock_cool_off_period_range,
+    )
+    s3_arguments.add_argument(
+        "--aws-snapshot-lock-expiration-date",
+        help="The expiration date for a locked snapshot in the format YYYY-MM-DDThh:mm:ss.sssZ. To lock a snapshot, you must specify either this argument or --aws-snapshot-lock-duration, but not both.",
+        type=check_timestamp
+    )
     azure_arguments.add_argument(
         "--encryption-scope",
         help="The name of an encryption scope defined in the Azure Blob Storage "
@@ -434,7 +454,14 @@ def parse_arguments(args=None):
         help="The name of the Azure resource group to which the compute instance and "
         "disks defined by the --snapshot-instance and --snapshot-disk arguments belong.",
     )
-    return parser.parse_args(args=args)
+
+    parsed_args = parser.parse_args(args=args)
+
+    # Perform mutual exclusivity check
+    if parsed_args.aws_snapshot_lock_duration is not None and parsed_args.aws_snapshot_lock_expiration_date is not None:
+        parser.error("You must specify either --aws-snapshot-lock-duration or --aws-snapshot-lock-expiration-date, but not both.")
+
+    return parsed_args
 
 
 if __name__ == "__main__":

barman/cloud_providers/__init__.py

@@ -201,6 +201,10 @@ def get_snapshot_interface(config):
             config.aws_profile,
             config.aws_region,
             config.aws_await_snapshots_timeout,
+            config.aws_snapshot_lock_mode,
+            config.aws_snapshot_lock_duration,
+            config.aws_snapshot_lock_cool_off_period,
+            config.aws_snapshot_lock_expiration_date,
         ]
         return AwsCloudSnapshotInterface(*args)
     else:
@@ -253,6 +257,10 @@ def get_snapshot_interface_from_server_config(server_config):
             server_config.aws_profile,
             server_config.aws_region,
             server_config.aws_await_snapshots_timeout,
+            server_config.aws_snapshot_lock_mode,
+            server_config.aws_snapshot_lock_duration,
+            server_config.aws_snapshot_lock_cool_off_period,
+            server_config.aws_snapshot_lock_expiration_date,
         )
     else:
         raise CloudProviderUnsupported(

barman/cloud_providers/aws_s3.py

@@ -463,21 +463,41 @@ class AwsCloudSnapshotInterface(CloudSnapshotInterface):
         https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ebs-creating-snapshot.html
     """
 
-    def __init__(self, profile_name=None, region=None, await_snapshots_timeout=3600):
+    def __init__(
+        self,
+        profile_name=None,
+        region=None,
+        await_snapshots_timeout=3600,
+        lock_mode=None,
+        lock_duration=None,
+        lock_cool_off_period=None,
+        lock_expiration_date=None
+    ):
         """
         Creates the client necessary for creating and managing snapshots.
 
         :param str profile_name: AWS auth profile identifier.
         :param str region: The AWS region in which snapshot resources are located.
         :param int await_snapshots_timeout: The maximum time in seconds to wait for
             snapshots to complete.
+        :param str lock_mode: The lock mode to apply to the snapshot.
+        :param int lock_duration: The duration (in days) for which the snapshot
+            should be locked.
+        :param int lock_cool_off_period: The cool-off period (in hours) for the snapshot.
+        :param str lock_expiration_date: The expiration date for the snapshot in the format
+            YYYY-MM-DDThh:mm:ss.sssZ.
         """
+
         self.session = boto3.Session(profile_name=profile_name)
         # If a specific region was provided then this overrides any region which may be
         # defined in the profile
         self.region = region or self.session.region_name
         self.ec2_client = self.session.client("ec2", region_name=self.region)
         self.await_snapshots_timeout = await_snapshots_timeout
+        self.lock_mode = lock_mode
+        self.lock_duration = lock_duration
+        self.lock_cool_off_period = lock_cool_off_period
+        self.lock_expiration_date = lock_expiration_date
 
     def _get_waiter_config(self):
         delay = 15
@@ -761,10 +781,12 @@ def take_snapshot_backup(self, backup_info, instance_identifier, volumes):
             snapshot_name, snapshot_resp = self._create_snapshot(
                 backup_info, volume_identifier, volume_metadata.id
             )
+
             snapshots.append(
                 AwsSnapshotMetadata(
                     snapshot_id=snapshot_resp["SnapshotId"],
                     snapshot_name=snapshot_name,
+                    snapshot_lock_mode=self.lock_mode,
                     device_name=attached_volumes[0]["DeviceName"],
                     mount_options=volume_metadata.mount_options,
                     mount_point=volume_metadata.mount_point,
@@ -783,14 +805,57 @@ def take_snapshot_backup(self, backup_info, instance_identifier, volumes):
             WaiterConfig=self._get_waiter_config(),
         )
 
+        # Apply lock on snapshots if lock mode is specified
+        if self.lock_mode:
+            self._lock_snapshots(
+                snapshots,
+                self.lock_mode,
+                self.lock_duration,
+                self.lock_cool_off_period,
+                self.lock_expiration_date
+            )
+
         backup_info.snapshots_info = AwsSnapshotsInfo(
             snapshots=snapshots,
             region=self.region,
             # All snapshots will have the same OwnerId so we get it from the last
             # snapshot response.
             account_id=snapshot_resp["OwnerId"],
+
         )
 
+    def _lock_snapshots(self, snapshots, lock_mode, lock_duration, lock_cool_off_period, lock_expiration_date):
+        lock_snapshot_default_args = {
+            "LockMode": lock_mode
+        }
+
+        if lock_duration:
+            lock_snapshot_default_args["LockDuration"] = lock_duration
+
+        if lock_cool_off_period:
+            lock_snapshot_default_args["CoolOffPeriod"] = lock_cool_off_period
+
+        if lock_expiration_date:
+            lock_snapshot_default_args["ExpirationDate"] = lock_expiration_date
+
+        for snapshot in snapshots:
+            lock_snapshot_args = lock_snapshot_default_args.copy()
+            lock_snapshot_args["SnapshotId"] = snapshot.identifier
+
+            resp = self.ec2_client.lock_snapshot(**lock_snapshot_args)
+
+            logging.info(
+                "Snapshot %s locked in state '%s' (lock duration: %s days, cool-off period: %s hours, "
+                "cool-off period expires on: %s, lock expires on: %s, lock duration time: %s)",
+                snapshot.identifier,
+                resp["LockState"],
+                resp["LockDuration"],
+                resp["CoolOffPeriod"],
+                resp["CoolOffPeriodExpiresOn"],
+                resp["LockExpiresOn"],
+                resp["LockDurationStartTime"]
+            )
+
     def _delete_snapshot(self, snapshot_id):
         """
         Delete the specified snapshot.
@@ -824,6 +889,16 @@ def delete_snapshot_backup(self, backup_info):
                 snapshot.identifier,
                 backup_info.backup_id,
             )
+
+            if snapshot.snapshot_lock_mode is not None:
+                resp = self.ec2_client.describe_locked_snapshots(
+                    SnapshotIds=[snapshot.identifier],
+                )
+
+                if resp["Snapshots"] and resp["Snapshots"][0]["LockState"] != "expired":
+                    logging.warning("Skipping deletion of snapshot %s as it not expired yet", snapshot.identifier)
+                    continue
+
             self._delete_snapshot(snapshot.identifier)
 
     def get_attached_volumes(
@@ -1003,11 +1078,11 @@ class AwsSnapshotMetadata(SnapshotMetadata):
     """
     Specialization of SnapshotMetadata for AWS EBS snapshots.
 
-    Stores the device_name, snapshot_id and snapshot_name in the provider-specific
+    Stores the device_name, snapshot_id, snapshot_name and snapshot_lock_mode in the provider-specific
     field.
     """
 
-    _provider_fields = ("device_name", "snapshot_id", "snapshot_name")
+    _provider_fields = ("device_name", "snapshot_id", "snapshot_name", "snapshot_lock_mode")
 
     def __init__(
         self,
@@ -1016,6 +1091,7 @@ def __init__(
         device_name=None,
         snapshot_id=None,
         snapshot_name=None,
+        snapshot_lock_mode=None,
     ):
         """
         Constructor saves additional metadata for AWS snapshots.
@@ -1027,12 +1103,14 @@ def __init__(
         :param str device_name: The device name used in the AWS API.
         :param str snapshot_id: The snapshot ID used in the AWS API.
         :param str snapshot_name: The snapshot name stored in the `Name` tag.
+        :param str snapshot_lock_mode: The mode with which the snapshot has been locked, if set.
         :param str project: The AWS project name.
         """
         super(AwsSnapshotMetadata, self).__init__(mount_options, mount_point)
         self.device_name = device_name
         self.snapshot_id = snapshot_id
         self.snapshot_name = snapshot_name
+        self.snapshot_lock_mode = snapshot_lock_mode
 
     @property
     def identifier(self):

barman/config.py

@@ -232,6 +232,17 @@ def parse_time_interval(value):
     return time_delta
 
 
+def parse_datetime(value):
+    """
+    Parse a string, transforming it in a datetime object.
+    Accepted format: YYYY-MM-DDThh:mm:ss.sssZ
+
+    :param str value: the string to evaluate
+    """
+
+    return datetime.datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
+
+
 def parse_si_suffix(value):
     """
     Parse a string, transforming it into integer and multiplying by
@@ -489,6 +500,10 @@ class ServerConfig(BaseConfig):
         "archiver_batch_size",
         "autogenerate_manifest",
         "aws_await_snapshots_timeout",
+        "aws_snapshot_lock_mode",
+        "aws_snapshot_lock_duration",
+        "aws_snapshot_lock_cool_off_period",
+        "aws_snapshot_lock_expiration_date",
         "aws_profile",
         "aws_region",
         "azure_credential",
@@ -587,6 +602,10 @@ class ServerConfig(BaseConfig):
         "archiver_batch_size",
         "autogenerate_manifest",
         "aws_await_snapshots_timeout",
+        "aws_snapshot_lock_mode",
+        "aws_snapshot_lock_duration",
+        "aws_snapshot_lock_cool_off_period",
+        "aws_snapshot_lock_expiration_date",
         "aws_profile",
         "aws_region",
         "azure_credential",
@@ -710,6 +729,9 @@ class ServerConfig(BaseConfig):
         "archiver_batch_size": int,
         "autogenerate_manifest": parse_boolean,
         "aws_await_snapshots_timeout": int,
+        "aws_snapshot_lock_duration": int,
+        "aws_snapshot_lock_cool_off_period": int,
+        "aws_snapshot_lock_expiration_date": parse_datetime,
         "backup_compression": parse_backup_compression,
         "backup_compression_format": parse_backup_compression_format,
         "backup_compression_level": int,

barman/utils.py

@@ -726,6 +726,44 @@ def check_positive(value):
     return int_value
 
 
+def check_aws_snapshot_lock_duration_range(value):
+    """
+    Check for AWS Snapshot Lock duration range option
+
+    :param value: str containing the value to check
+    """
+    if value is None:
+        return None
+    try:
+        int_value = int(value)
+    except Exception:
+        raise ArgumentTypeError("'%s' is not a valid input" % value)
+
+    if int_value < 1 or int_value > 36500:
+        raise ArgumentTypeError("'%s' is outside supported range of 1-36500 days" % value)
+
+    return int_value
+
+
+def check_aws_snapshot_lock_cool_off_period_range(value):
+    """
+    Check for AWS Snapshot Lock cool-off period range option
+
+    :param value: str containing the value to check
+    """
+    if value is None:
+        return None
+    try:
+        int_value = int(value)
+    except Exception:
+        raise ArgumentTypeError("'%s' is not a valid input" % value)
+
+    if int_value < 1 or int_value > 72:
+        raise ArgumentTypeError("'%s' is outside supported range of 1-72 hours" % value)
+
+    return int_value
+
+
 def check_tli(value):
     """
     Check for a positive integer option, and also make "current" and "latest" acceptable values
@@ -783,6 +821,16 @@ def check_size(value):
     return int_value
 
 
+def check_timestamp(value):
+    try:
+        # Attempt to parse the input date string into a datetime object
+        return datetime.datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
+    except ValueError:
+        raise ArgumentTypeError(
+            "Invalid expiration date: '%s'. Expected format is 'YYYY-MM-DDThh:mm:ss.sssZ'." % value
+        )
+
+
 def check_backup_name(backup_name):
     """
     Verify that a backup name is not a backup ID or reserved identifier.

tests/test_cli.py

@@ -1761,6 +1761,7 @@ def test_help_output(self, minimal_parser, capsys):
         if sys.version_info < (3, 10):
             options_label = "optional arguments"
         expected_output = self._expected_help_output.format(options_label=options_label)
+
         assert expected_output == out