Блокировка объектов

AWS CLI

Если AWS CLI еще не установлен, установите и настройте его. Для работы с объектным хранилищем используйте команды s3api и s3.

AWS CLI

Проверить конфигурацию объектов бакета

В терминале выполните команду:

aws s3api get-object-lock-configuration --bucket <bucket_name>

Укажите:

<bucket_name> — имя бакета.

Создание бакета с блокировкой объектов

В терминале выполните команду:

aws s3api s3api create-bucket bucket <bucket_name> object-lock-enabled-for-bucket

Укажите:

<bucket_name> — имя бакета.

Настройка конфигурации блокировки объектов

В терминале выполните команду:

aws s3api put-object-lock-configuration \
  --bucket <bucket_name> \
  --object-lock-configuration '{
    "ObjectLockEnabled": "Enabled",
    "Rule": {
      "DefaultRetention": {
        "Mode": "<mode>",
        "Days": <days>
      }
    }
  }'

Укажите:

<bucket_name> — имя бакета; <mode> — COMPLIANCE или GOVERNANCE; <days> — количество дней в формате DD.

Проверка текущей блокировки объекта

В терминале выполните команду:

aws s3api get-object-retention --bucket <bucket_name> --key <object_name>

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Снятие блокировки в режиме GOVERNANCE

Непосредственно снять блокировку нельзя, пока RetainUntilDate не истекла. Но вы можете, установить RetainUntilDate в максимально близкую, но все же будущую дату (например, завтра), чтобы блокировка скоро закончилась:

В терминале выполните команду:

aws s3api put-object-retention \
    --bucket <bucket_name> \
    --key <object_name> \
    --retention "Mode=GOVERNANCE, RetainUntilDate=<date>" \
    --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <date> — на дату в будущем в формате YYYY-MM-DD.

Продлить срок блокировки (оба режима)

В терминале выполните команду:

aws s3api put-object-retention \
    --bucket <bucket_name> \
    --key <object_name> \
    --retention "Mode=<mode>, RetainUntilDate=<date>" \
    --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <mode> — COMPLIANCE или GOVERNANCE; <date> — на дату в будущем в формате YYYY-MM-DD.

Удалить объект с блокировкой GOVERNANCE

В терминале выполните команду:

aws s3api delete-object \
  --bucket <bucket_name> \
  --key <object_name> \
  --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Получить информацию о блокировке конкретной версии

В терминале выполните команду:

aws s3api get-object-retention \
    --bucket <bucket_name> \
    --key <object_name> \
    --version-id <versionid> \

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии.

Снятие блокировки объекта конкретной версии в режиме GOVERNANCE

Непосредственно снять блокировку нельзя, пока RetainUntilDate не истекла. Но вы можете, установить RetainUntilDate в максимально близкую, но все же будущую дату (например, завтра), чтобы блокировка скоро закончилась:

В терминале выполните команду:

aws s3api put-object-retention \
    --bucket <bucket_name> \
    --key <object_name> \
    --version-id <versionid> \
    --retention "Mode=GOVERNANCE, RetainUntilDate=<date>" \
    --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии; <date> — на дату в будущем в формате YYYY-MM-DD.

Продлить срок блокировки объекта конкретной версии (оба режима)

В терминале выполните команду:

aws s3api put-object-retention \
    --bucket <bucket_name> \
    --key <object_name> \
    --version-id <versionid> \
    --retention "Mode=<mode>, RetainUntilDate=<date>" \
    --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии; <mode> — COMPLIANCE или GOVERNANCE; <date> — на дату в будущем в формате YYYY-MM-DD.

Удалить объект конкретной версии с блокировкой GOVERNANCE

В терминале выполните команду:

aws s3api delete-object \
  --bucket <bucket_name> \
  --key <object_name> \
  --version-id <versionid> \
  --bypass-governance-retention

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии.

Работа с Legal Hold

Получение статуса Legal Hold

В терминале выполните команду:

aws s3api get-object-legal-hold \
    --bucket <bucket_name> \
    --key <object_name> \

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Включить Legal Hold

В терминале выполните команду:

aws s3api put-object-legal-hold \
    --bucket <bucket_name> \
    --key <object_name> \
    --legal-hold Status=ON

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Снять Legal Hold

В терминале выполните команду:

aws s3api put-object-legal-hold \
--bucket <bucket_name> \
--key <object_name> \
--legal-hold Status=OFF

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Подробнее о блокировке объектов см. в статье Блокировка объектов. Подробнее об объектах см. в статье Об объектах.

Python API

Boto3 — это библиотека Python, предоставляющая программный интерфейс (SDK) для взаимодействия с объектным хранилищем S3. Подробности см. в документации AWS Boto3.

Python API

Проверить конфигурацию объектов бакета

import boto3
from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def get_object_lock_config(bucket_name):
    try:
        response = s3.get_object_lock_configuration(Bucket=bucket_name)
        print("Object Lock Configuration:", response.get('ObjectLockConfiguration', 'не настроено'))
    except ClientError as e:
        print(f"Ошибка: {e}")
 
get_object_lock_config('<bucket_name>')

Укажите:

<bucket_name> — имя бакета.

Создание бакета с блокировки объектов

import boto3
from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def create_bucket_with_object_lock(bucket_name, region):
    try:
        s3.create_bucket(
            Bucket=bucket_name,
            CreateBucketConfiguration={'LocationConstraint': region},
            ObjectLockEnabledForBucket=True
        )
        print(f"Бакет '{bucket_name}' создан с поддержкой Object Lock.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
create_bucket_with_object_lock('<bucket_name>', '<region>')

Укажите:

<bucket_name> — имя бакета; <region> — регион, в котором будет создан бакет со включенной блокировкой объектов.

Настройка конфигурации блокировки объектов

import boto3
from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def set_object_lock_configuration(bucket_name, mode, days):
    try:
        config = {
            "ObjectLockEnabled": "Enabled",
            "Rule": {
                "DefaultRetention": {
                    "Mode": mode,
                    "Days": days
                }
            }
        }
        s3.put_object_lock_configuration(
            Bucket=bucket_name,
            ObjectLockConfiguration=config
        )
        print(f"Object Lock '{mode}' на {days} дней установлен для бакета '{bucket_name}'.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
set_object_lock_configuration('<bucket_name>', mode="<mode>", days=<days>)

Укажите:

<bucket_name> — имя бакета; <mode> — COMPLIANCE или GOVERNANCE; <days> — количество дней в формате DD.

Проверка текущей блокировки объекта

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def get_object_retention(bucket_name, object_key):
    try:
        response = s3.get_object_retention(
            Bucket=bucket_name,
            Key=object_key
        )
        retention = response.get('Retention')
        if retention:
            mode = retention.get('Mode', 'N/A')
            date = retention.get('RetainUntilDate')
            if isinstance(date, datetime):
                date_str = date.strftime('%Y-%m-%d %H:%M:%S')
            else:
                date_str = str(date)
            print(f"Блокировка объекта '{object_key}':")
            print(f"  Режим: {mode}")
            print(f"  Действует до: {date_str}")
        else:
            print(f"Для объекта '{object_key}' не установлена блокировка.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
get_object_retention('<bucket_name>', '<object_name>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Снятие блокировки в режиме GOVERNANCE

Непосредственно снять блокировку нельзя, пока RetainUntilDate не истекла. Но вы можете, установить RetainUntilDate в максимально близкую, но все же будущую дату (например, завтра), чтобы блокировка скоро закончилась:

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def set_retention_governance(bucket, key, days_ahead):
    retain_until = (datetime.utcnow() + timedelta(days=days_ahead)).replace(microsecond=0).isoformat() + 'Z'
    try:
        s3.put_object_retention(
            Bucket=bucket,
            Key=key,
            Retention={
                'Mode': 'GOVERNANCE',
                'RetainUntilDate': retain_until
            },
            BypassGovernanceRetention=True
        )
        print(f"Установлена блокировка GOVERNANCE до {retain_until} для объекта {key}")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
set_retention_governance('<bucket_name>', '<object_name>', days_ahead=<date>)

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <date> — на дату в будущем в формате DD.

Продлить срок блокировки (оба режима)

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def extend_retention(bucket, key, mode, until_date):
    try:
        s3.put_object_retention(
            Bucket=bucket,
            Key=key,
            Retention={
                'Mode': mode.upper(),
                'RetainUntilDate': until_date + 'T00:00:00Z'
            },
            BypassGovernanceRetention=True
        )
        print(f"Продлена блокировка в режиме {mode} до {until_date} для объекта {key}")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
 
extend_retention('<bucket_name>', '<object_name>', '<mode>', '<date>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <mode> — COMPLIANCE или GOVERNANCE; <date> — на дату в будущем в формате YYYY-MM-DD.

Удалить объект с блокировкой GOVERNANCE

import boto3 from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def delete_object_governance(bucket, key):
    try:
        s3.delete_object(
            Bucket=bucket,
            Key=key,
            BypassGovernanceRetention=True
        )
        print(f"Объект {key} удалён с обходом GOVERNANCE блокировки.")
    except ClientError as e:
        print(f"Ошибка удаления: {e}")
 
 
delete_object_governance('<bucket_name>', '<object_name>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Получить информацию о блокировке конкретной версии

В терминале выполните команду:

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def get_retention_version(bucket, key, version_id):
    try:
        response = s3.get_object_retention(
            Bucket=bucket,
            Key=key,
            VersionId=version_id
        )
        retention = response.get('Retention')
        if retention:
            mode = retention.get('Mode', 'N/A')
            date = retention.get('RetainUntilDate')
            if isinstance(date, datetime):
                date_str = date.strftime('%Y-%m-%d %H:%M:%S')
            else:
                date_str = str(date)
            print(f"Блокировка объекта '{key}':")
            print(f"  Режим: {mode}")
            print(f"  Действует до: {date_str}")
        else:
            print(f"Для объекта '{key}' не установлена блокировка.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
get_retention_version('<bucket_name>', '<object_name>', '<versionid>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии.

Снятие блокировки объекта конкретной версии в режиме GOVERNANCE

Непосредственно снять блокировку нельзя, пока RetainUntilDate не истекла. Но вы можете, установить RetainUntilDate в максимально близкую, но все же будущую дату (например, завтра), чтобы блокировка скоро закончилась:

import boto3 from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials. s3 = session.client('s3', endpoint_url='$endpoint')
 
def set_retention_governance(bucket, key, version_id, days_ahead):
    retain_until = (datetime.utcnow() + timedelta(days=days_ahead)).replace(microsecond=0).isoformat() + 'Z'
    try:
        s3.put_object_retention(
            Bucket=bucket,
            Key=key,
            VersionId=version_id,
            Retention={
                'Mode': 'GOVERNANCE',
                'RetainUntilDate': retain_until
            },
            BypassGovernanceRetention=True
        )
        print(f"Установлена блокировка GOVERNANCE до {retain_until} для объекта {key}")
    except ClientError as e:
        print(f"Ошибка: {e}")
  
set_retention_governance('<bucket_name>', '<object_name>','<versionid>', days_ahead=<date>)

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии; <date> — на дату в будущем в формате DD.

Продлить срок блокировки объекта конкретной версии (оба режима)

import boto3 from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials. s3 = session.client('s3', endpoint_url='$endpoint')
 
def extend_retention(bucket, key, mode, version_id, until_date):
    try:
        s3.put_object_retention(
            Bucket=bucket,
            Key=key,
            VersionId=version_id,
            Retention={
                'Mode': mode.upper(),
                'RetainUntilDate': until_date + 'T00:00:00Z'
            },
            BypassGovernanceRetention=True
        )
        print(f"Продлена блокировка в режиме {mode} до {until_date} для объекта {key}")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
extend_retention('<bucket_name>', '<object_name>', '<mode>', '<versionid>', '<date>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <mode> — COMPLIANCE или GOVERNANCE; <versionid> — уникальный идентификатор версии; <date> — на дату в будущем в формате YYYY-MM-DD.

Удалить объект конкретной версии с блокировкой GOVERNANCE

import boto3 from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials. s3 = session.client('s3', endpoint_url='$endpoint')
 
def delete_object_governance(bucket, key, version_id):
    try:
        s3.delete_object(
            Bucket=bucket,
            Key=key,
            VersionId=version_id,
            BypassGovernanceRetention=True
        )
        print(f"Объект {key} удалён с обходом GOVERNANCE блокировки.")
    except ClientError as e:
        print(f"Ошибка удаления: {e}")
  
  
delete_object_governance('<bucket_name>', '<object_name>','<versionid>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта; <versionid> — уникальный идентификатор версии.

Работа с Legal Hold

Получение статуса Legal Hold

import boto3 from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials. s3 = session.client('s3', endpoint_url='$endpoint')
 
def get_legal_hold_status(bucket_name, object_key):
    try:
        response = s3.get_object_legal_hold(Bucket=bucket_name, Key=object_key)
        status = response.get('LegalHold', {}).get('Status', 'NOT SET')
        print(f"Legal Hold статус объекта '{object_key}' в бакете '{bucket_name}': {status}")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
get_legal_hold_status('<bucket_name>', '<object_name>')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Включить Legal Hold

import boto3
from botocore.exceptions import ClientError
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials. s3 = session.client('s3', endpoint_url='$endpoint')
 
def set_legal_hold(bucket_name, object_key, status):
    assert status in ['ON']
    try:
        s3.put_object_legal_hold(
            Bucket=bucket_name,
            Key=object_key,
            LegalHold={'Status': status}
        )
        print(f"Legal Hold '{status}' установлен для объекта '{object_key}' в бакете '{bucket_name}'.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
# Пример вызовов
set_legal_hold('<bucket_name>', '<object_name>', 'ON')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Снять Legal Hold

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
session = boto3.session.Session(profile_name='default') #Создает сессию с использованием профиля default. Профиль находится в файле ~/.aws/credentials.
s3 = session.client('s3', endpoint_url='$endpoint')
 
def set_legal_hold(bucket_name, object_key, status):
    assert status in ['OFF']
    try:
        s3.put_object_legal_hold(
            Bucket=bucket_name,
            Key=object_key,
            LegalHold={'Status': status}
        )
        print(f"Legal Hold '{status}' установлен для объекта '{object_key}' в бакете '{bucket_name}'.")
    except ClientError as e:
        print(f"Ошибка: {e}")
 
# Пример вызовов
set_legal_hold('<bucket_name>', '<object_name>', 'OFF')

Укажите:

<bucket_name> — имя бакета; <object_name> — имя объекта.

Подробнее о блокировке объектов см. в статье Блокировка объектов. Подробнее об объектах см. в статье Об объектах.