File size: 3,828 Bytes
fea414e e79a06d fea414e 2008dd3 fea414e e79a06d fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e 503d4ac fea414e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
import logging
import os
import sys
# Add parent directory to path to allow importing from google_src
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from google.cloud import storage
from google.iam.v1 import policy_pb2
from google_src.gcloud_wrapper import get_default_wrapper
# Use the project's configured logger to avoid duplicate log output
try:
from utils import logger
except ImportError:
logger = logging.getLogger(__name__)
def setup_bucket_permissions(bucket_name: str, members: list, role: str = "roles/storage.objectViewer", storage_client=None):
"""
Creates a GCS bucket (if it doesn't exist) and grants the specified role to the given members.
Args:
bucket_name (str): The name of the GCS bucket.
members (list): A list of members to grant access to (e.g., ["user:jebin.einstein@elvoro.com", "serviceAccount:elvoro@elvoro-final-videos.iam.gserviceaccount.com"]).
role (str): The IAM role to grant (default: roles/storage.objectViewer).
storage_client (google.cloud.storage.Client, optional): Existing storage client to reuse.
"""
try:
# Initialize the wrapper to get the client with 'final_data' account credentials if not provided
if not storage_client:
wrapper = get_default_wrapper()
storage_client = wrapper.get_storage_client("final_data")
# 1. Create or get the bucket
try:
bucket = storage_client.get_bucket(bucket_name)
logger.debug(f"✅ Bucket '{bucket_name}' already exists.")
except Exception:
try:
logger.debug(f"📦 Bucket '{bucket_name}' not found. Attempting to create...")
bucket = storage_client.create_bucket(bucket_name, location="us-central1")
logger.debug(f"✅ Bucket '{bucket_name}' created successfully.")
except Exception as e:
logger.error(f"❌ Failed to create bucket '{bucket_name}': {e}")
return
# 2. Update IAM Policy
logger.debug(f"🔒 Updating IAM policy for bucket '{bucket_name}'...")
policy = bucket.get_iam_policy(requested_policy_version=3)
# Check if binding already exists for this role
binding = next((b for b in policy.bindings if b['role'] == role), None)
if binding:
logger.debug(f"Found existing binding for role '{role}'. Adding new members...")
# specific binding found, add members if not present
existing_members = set(binding['members'])
new_members = set(members)
updated_members = existing_members.union(new_members)
binding['members'] = list(updated_members)
else:
logger.debug(f"No existing binding for role '{role}'. Creating new binding...")
# Create a new binding
binding = {"role": role, "members": members}
policy.bindings.append(binding)
# Set the updated policy
bucket.set_iam_policy(policy)
logger.debug(f"✅ IAM policy updated successfully for bucket '{bucket_name}'.")
logger.debug(f" Granted '{role}' to:")
for member in members:
logger.debug(f" - {member}")
except Exception as e:
logger.error(f"❌ An error occurred during permission setup: {e}")
if __name__ == "__main__":
# Configuration from user request
BUCKET_NAME = "globe_air"
TARGET_MEMBERS = [
"user:jebin.einstein@elvoro.com",
"serviceAccount:elvoro@elvoro-final-videos.iam.gserviceaccount.com"
]
TARGET_ROLE = "roles/storage.objectViewer"
logger.debug("🚀 Starting GCS Bucket Permission Setup...")
setup_bucket_permissions(BUCKET_NAME, TARGET_MEMBERS, TARGET_ROLE)
|