Add bypass governance

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 * Add `get_file_info_by_name` to the B2Api class
+* 'bypass_governance' flag to delete_file_version
 
 ### Fixed
 * Require `typing_extensions` on Python 3.11 (already required on earlier versinons) for better compatibility with pydantic v2

b2sdk/api.py

@@ -460,12 +460,15 @@ def cancel_large_file(self, file_id: str) -> FileIdAndName:
         """
         return self.services.large_file.cancel_large_file(file_id)
 
-    def delete_file_version(self, file_id: str, file_name: str) -> FileIdAndName:
+    def delete_file_version(
+        self, file_id: str, file_name: str, bypass_governance: bool = False
+    ) -> FileIdAndName:
         """
-        Permanently and irrevocably delete one version of a file.
+        Permanently and irrevocably delete one version of a file. bypass_governance must be set to true if deleting a
+        file version protected by Object Lock governance mode retention settings (unless its retention period expired)
         """
         # filename argument is not first, because one day it may become optional
-        response = self.session.delete_file_version(file_id, file_name)
+        response = self.session.delete_file_version(file_id, file_name, bypass_governance)
         return FileIdAndName.from_cancel_or_delete_response(response)
 
     # download

b2sdk/bucket.py

@@ -1207,15 +1207,17 @@ def copy(
             max_part_size=max_part_size,
         )
 
-    def delete_file_version(self, file_id, file_name):
+    def delete_file_version(self, file_id: str, file_name: str, bypass_governance: bool = False):
         """
         Delete a file version.
 
-        :param str file_id: a file ID
-        :param str file_name: a file name
+        :param file_id: a file ID
+        :param file_name: a file name
+        :param bypass_governance: Must be set to true if deleting a file version protected by Object Lock governance
+                                  mode retention settings (unless its retention period expired)
         """
         # filename argument is not first, because one day it may become optional
-        return self.api.delete_file_version(file_id, file_name)
+        return self.api.delete_file_version(file_id, file_name, bypass_governance)
 
     @disable_trace
     def as_dict(self):

b2sdk/file_version.py

@@ -177,8 +177,10 @@ def _all_slots(self):
             all_slots.extend(getattr(klass, '__slots__', []))
         return all_slots
 
-    def delete(self) -> FileIdAndName:
-        return self.api.delete_file_version(self.id_, self.file_name)
+    def delete(self, bypass_governance: bool = False) -> FileIdAndName:
+        """Delete this file version. bypass_governance must be set to true if deleting a file version protected by
+        Object Lock governance mode retention settings (unless its retention period expired)"""
+        return self.api.delete_file_version(self.id_, self.file_name, bypass_governance)
 
     def update_legal_hold(self, legal_hold: LegalHold) -> BaseFileVersion:
         legal_hold = self.api.update_file_legal_hold(self.id_, self.file_name, legal_hold)

b2sdk/raw_api.py

@@ -191,7 +191,9 @@ def delete_bucket(self, api_url, account_auth_token, account_id, bucket_id):
         pass
 
     @abstractmethod
-    def delete_file_version(self, api_url, account_auth_token, file_id, file_name):
+    def delete_file_version(
+        self, api_url, account_auth_token, file_id, file_name, bypass_governance: bool = False
+    ):
         pass
 
     @abstractmethod
@@ -528,13 +530,16 @@ def delete_bucket(self, api_url, account_auth_token, account_id, bucket_id):
             bucketId=bucket_id
         )
 
-    def delete_file_version(self, api_url, account_auth_token, file_id, file_name):
+    def delete_file_version(
+        self, api_url, account_auth_token, file_id, file_name, bypass_governance: bool = False
+    ):
         return self._post_json(
             api_url,
             'b2_delete_file_version',
             account_auth_token,
             fileId=file_id,
-            fileName=file_name
+            fileName=file_name,
+            bypassGovernance=bypass_governance,
         )
 
     def delete_key(self, api_url, account_auth_token, application_key_id):

b2sdk/raw_simulator.py

@@ -26,6 +26,7 @@
 from .b2http import ResponseContextManager
 from .encryption.setting import EncryptionMode, EncryptionSetting
 from .exception import (
+    AccessDenied,
     BadJson,
     BadRequest,
     BadUploadUrl,
@@ -51,6 +52,7 @@
     BucketRetentionSetting,
     FileRetentionSetting,
     LegalHold,
+    RetentionMode,
 )
 from .file_version import UNVERIFIED_CHECKSUM_PREFIX
 from .raw_api import ALL_CAPABILITIES, AbstractRawApi, LifecycleRule, MetadataDirectiveMode
@@ -524,9 +526,8 @@ def __init__(
         # File IDs count down, so that the most recent will come first when they are sorted.
         self.file_id_counter = iter(range(self.FIRST_FILE_NUMBER, 0, -1))
         self.upload_timestamp_counter = iter(range(5000, 9999))
-        self.file_id_to_file = dict()
-        # It would be nice to use an OrderedDict for this, but 2.6 doesn't have it.
-        self.file_name_and_id_to_file = dict()
+        self.file_id_to_file: dict[str, FileSimulator] = dict()
+        self.file_name_and_id_to_file: dict[tuple[str, str], FileSimulator] = dict()
         if default_server_side_encryption is None:
             default_server_side_encryption = EncryptionSetting(mode=EncryptionMode.NONE)
         self.default_server_side_encryption = default_server_side_encryption
@@ -537,6 +538,12 @@ def __init__(
             assert self.replication.asReplicationSource is None or self.replication.asReplicationSource.rules
             assert self.replication.asReplicationDestination is None or self.replication.asReplicationDestination.sourceToDestinationKeyMapping
 
+    def get_file(self, file_id, file_name) -> FileSimulator:
+        try:
+            return self.file_name_and_id_to_file[(file_name, file_id)]
+        except KeyError:
+            raise FileNotPresent(file_id_or_name=file_id)
+
     def is_allowed_to_read_bucket_encryption_setting(self, account_auth_token):
         return self._check_capability(account_auth_token, 'readBucketEncryption')
 
@@ -612,9 +619,23 @@ def cancel_large_file(self, file_id):
             fileName=file_sim.name
         )  # yapf: disable
 
-    def delete_file_version(self, file_id, file_name):
+    def delete_file_version(
+        self, account_auth_token, file_id, file_name, bypass_governance: bool = False
+    ):
         key = (file_name, file_id)
-        file_sim = self.file_name_and_id_to_file[key]
+        file_sim = self.get_file(file_id, file_name)
+        if file_sim.file_retention:
+            if file_sim.file_retention.retain_until and file_sim.file_retention.retain_until > int(
+                time.time()
+            ):
+                if file_sim.file_retention.mode == RetentionMode.COMPLIANCE:
+                    raise AccessDenied()
+                elif file_sim.file_retention.mode == RetentionMode.GOVERNANCE:
+                    if not bypass_governance:
+                        raise AccessDenied()
+                    if not self._check_capability(account_auth_token, 'bypassGovernance'):
+                        raise AccessDenied()
+
         del self.file_name_and_id_to_file[key]
         del self.file_id_to_file[file_id]
         return dict(fileId=file_id, fileName=file_name, uploadTimestamp=file_sim.upload_timestamp)
@@ -1180,10 +1201,10 @@ def __init__(self, b2_http=None):
         # Counter for generating account IDs an their matching master application keys.
         self.account_counter = 0
 
-        self.bucket_name_to_bucket = dict()
-        self.bucket_id_to_bucket = dict()
+        self.bucket_name_to_bucket: dict[str, BucketSimulator] = dict()
+        self.bucket_id_to_bucket: dict[str, BucketSimulator] = dict()
         self.bucket_id_counter = iter(range(100))
-        self.file_id_to_bucket_id = {}
+        self.file_id_to_bucket_id: dict[str, str] = {}
         self.all_application_keys = []
         self.app_key_counter = 0
         self.upload_errors = []
@@ -1354,11 +1375,15 @@ def create_key(
         self.all_application_keys.append(key_sim)
         return key_sim.as_created_key()
 
-    def delete_file_version(self, api_url, account_auth_token, file_id, file_name):
-        bucket_id = self.file_id_to_bucket_id[file_id]
+    def delete_file_version(
+        self, api_url, account_auth_token, file_id, file_name, bypass_governance: bool = False
+    ):
+        bucket_id = self.file_id_to_bucket_id.get(file_id)
+        if not bucket_id:
+            raise FileNotPresent(file_id_or_name=file_id)
         bucket = self._get_bucket_by_id(bucket_id)
         self._assert_account_auth(api_url, account_auth_token, bucket.account_id, 'deleteFiles')
-        return bucket.delete_file_version(file_id, file_name)
+        return bucket.delete_file_version(account_auth_token, file_id, file_name, bypass_governance)
 
     def update_file_retention(
         self,
@@ -1934,7 +1959,7 @@ def _assert_account_auth(
             if file_name is not None and not file_name.startswith(key_sim.name_prefix_or_none):
                 raise Unauthorized('', 'unauthorized')
 
-    def _get_bucket_by_id(self, bucket_id):
+    def _get_bucket_by_id(self, bucket_id) -> BucketSimulator:
         if bucket_id not in self.bucket_id_to_bucket:
             raise NonExistentBucket(bucket_id)
         return self.bucket_id_to_bucket[bucket_id]

b2sdk/session.py

@@ -182,8 +182,10 @@ def delete_key(self, application_key_id):
     def delete_bucket(self, account_id, bucket_id):
         return self._wrap_default_token(self.raw_api.delete_bucket, account_id, bucket_id)
 
-    def delete_file_version(self, file_id, file_name):
-        return self._wrap_default_token(self.raw_api.delete_file_version, file_id, file_name)
+    def delete_file_version(self, file_id, file_name, bypass_governance: bool = False):
+        return self._wrap_default_token(
+            self.raw_api.delete_file_version, file_id, file_name, bypass_governance
+        )
 
     def download_file_from_url(self, url, range_=None, encryption: EncryptionSetting | None = None):
         return self._wrap_token(

test/unit/api/test_api.py

@@ -34,7 +34,7 @@
     RawSimulator,
     RetentionMode,
 )
-from apiver_deps_exception import FileNotPresent, InvalidArgument, RestrictedBucket
+from apiver_deps_exception import AccessDenied, FileNotPresent, InvalidArgument, RestrictedBucket
 
 from ..test_base import create_key
 
@@ -632,3 +632,20 @@ def test_get_key(self):
 
         assert self.api.get_key(key_id) is None
         assert self.api.get_key('non-existent') is None
+
+    def test_delete_file_version_bypass_governance(self):
+        self._authorize_account()
+        bucket = self.api.create_bucket('bucket1', 'allPrivate')
+        created_file = bucket.upload_bytes(
+            b'hello world',
+            'file',
+            file_retention=FileRetentionSetting(RetentionMode.GOVERNANCE,
+                                                int(time.time()) + 100),
+        )
+
+        with pytest.raises(AccessDenied):
+            self.api.delete_file_version(created_file.id_, 'file')
+
+        self.api.delete_file_version(created_file.id_, 'file', bypass_governance=True)
+        with pytest.raises(FileNotPresent):
+            bucket.get_file_info_by_name(created_file.file_name)

test/unit/bucket/test_bucket.py

@@ -14,13 +14,15 @@
 import os
 import pathlib
 import platform
+import time
 import unittest.mock as mock
 from contextlib import suppress
 from io import BytesIO
 
 import apiver_deps
 import pytest
 from apiver_deps_exception import (
+    AccessDenied,
     AlreadyFailed,
     B2ConnectionError,
     B2Error,
@@ -567,6 +569,22 @@ def test_delete_file_version(self):
         expected = [('hello.txt', 15, 'upload', None)]
         self.assertBucketContents(expected, '', show_versions=True)
 
+    def test_delete_file_version_bypass_governance(self):
+        data = b'hello world'
+
+        file_id = self.bucket.upload_bytes(
+            data,
+            'hello.txt',
+            file_retention=FileRetentionSetting(RetentionMode.GOVERNANCE,
+                                                int(time.time()) + 100),
+        ).id_
+
+        with pytest.raises(AccessDenied):
+            self.bucket.delete_file_version(file_id, 'hello.txt')
+
+        self.bucket.delete_file_version(file_id, 'hello.txt', bypass_governance=True)
+        self.assertBucketContents([], '', show_versions=True)
+
     def test_non_recursive_returns_folder_names(self):
         data = b'hello world'
         self.bucket.upload_bytes(data, 'a')

test/unit/file_version/test_file_version.py

@@ -9,6 +9,8 @@
 ######################################################################
 from __future__ import annotations
 
+import time
+
 import apiver_deps
 import pytest
 from apiver_deps import (
@@ -27,7 +29,7 @@
     RawSimulator,
     RetentionMode,
 )
-from apiver_deps_exception import FileNotPresent
+from apiver_deps_exception import AccessDenied, FileNotPresent
 
 if apiver_deps.V <= 1:
     from apiver_deps import FileVersionInfo as VFileVersion
@@ -138,6 +140,21 @@ def test_delete_file_version(self):
         with pytest.raises(FileNotPresent):
             self.bucket.get_file_info_by_name(self.file_version.file_name)
 
+    def test_delete_bypass_governance(self):
+        locked_file_version = self.bucket.upload_bytes(
+            b'nothing',
+            'test_file_with_governance',
+            file_retention=FileRetentionSetting(RetentionMode.GOVERNANCE,
+                                                int(time.time()) + 100),
+        )
+
+        with pytest.raises(AccessDenied):
+            locked_file_version.delete()
+
+        locked_file_version.delete(bypass_governance=True)
+        with pytest.raises(FileNotPresent):
+            self.bucket.get_file_info_by_name(locked_file_version.file_name)
+
     def test_delete_download_version(self):
         download_version = self.api.download_file_by_id(self.file_version.id_).download_version
         ret = download_version.delete()