IAM and Security

def get_iam_policy(
    self,
    request: Union[iam_policy_pb2.GetIamPolicyRequest, dict] = None,
    *,
    resource: str = None,
    retry: OptionalRetry = DEFAULT,
    timeout: float = DEFAULT,
    metadata: Sequence[Tuple[str, str]] = ()
) -> policy_pb2.Policy:
    """Get the IAM policy for a queue.
    
    Args:
        request: The request object or dictionary.
        resource: Required. The resource for which the policy is being requested.
        retry: Designation of what errors should be retried.
        timeout: The timeout for this request.
        metadata: Strings which should be sent along with the request as metadata.
        
    Returns:
        The IAM policy for the resource.
    """

def set_iam_policy(
    self,
    request: Union[iam_policy_pb2.SetIamPolicyRequest, dict] = None,
    *,
    resource: str = None,
    retry: OptionalRetry = DEFAULT,
    timeout: float = DEFAULT,
    metadata: Sequence[Tuple[str, str]] = ()
) -> policy_pb2.Policy:
    """Set the IAM policy for a queue.
    
    Args:
        request: The request object or dictionary.
        resource: Required. The resource for which the policy is being specified.
        retry: Designation of what errors should be retried.
        timeout: The timeout for this request.
        metadata: Strings which should be sent along with the request as metadata.
        
    Returns:
        The updated IAM policy.
    """

def test_iam_permissions(
    self,
    request: Union[iam_policy_pb2.TestIamPermissionsRequest, dict] = None,
    *,
    resource: str = None,
    permissions: MutableSequence[str] = None,
    retry: OptionalRetry = DEFAULT,
    timeout: float = DEFAULT,
    metadata: Sequence[Tuple[str, str]] = ()
) -> iam_policy_pb2.TestIamPermissionsResponse:
    """Test IAM permissions on a queue.
    
    Args:
        request: The request object or dictionary.
        resource: Required. The resource for which the policy detail is being requested.
        permissions: Required. The set of permissions to check for the resource.
        retry: Designation of what errors should be retried.
        timeout: The timeout for this request.
        metadata: Strings which should be sent along with the request as metadata.
        
    Returns:
        The permissions that the caller has on the resource.
    """

def get_location(
    self,
    request: Union[locations_pb2.GetLocationRequest, dict] = None,
    *,
    retry: OptionalRetry = DEFAULT,
    timeout: float = DEFAULT,
    metadata: Sequence[Tuple[str, str]] = ()
) -> locations_pb2.Location:
    """Get information about a location.
    
    Args:
        request: The request object or dictionary.
        retry: Designation of what errors should be retried.
        timeout: The timeout for this request.
        metadata: Strings which should be sent along with the request as metadata.
        
    Returns:
        The location information.
    """

def list_locations(
    self,
    request: Union[locations_pb2.ListLocationsRequest, dict] = None,
    *,
    retry: OptionalRetry = DEFAULT,
    timeout: float = DEFAULT,
    metadata: Sequence[Tuple[str, str]] = ()
) -> locations_pb2.ListLocationsResponse:
    """List information about supported locations.
    
    Args:
        request: The request object or dictionary.
        retry: Designation of what errors should be retried.
        timeout: The timeout for this request.
        metadata: Strings which should be sent along with the request as metadata.
        
    Returns:
        The list of locations.
    """

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'secure-queue')

# Get current IAM policy for a queue
policy = client.get_iam_policy(resource=queue_path)

print(f'IAM Policy for {queue_path}:')
print(f'Version: {policy.version}')
print(f'ETag: {policy.etag}')

for binding in policy.bindings:
    print(f'Role: {binding.role}')
    print(f'Members: {list(binding.members)}')
    if binding.condition:
        print(f'Condition: {binding.condition.expression}')

from google.cloud import tasks
from google.iam.v1 import policy_pb2

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'secure-queue')

# Get current policy
current_policy = client.get_iam_policy(resource=queue_path)

# Add a new binding for task enqueuers
new_binding = policy_pb2.Binding(
    role='roles/cloudtasks.enqueuer',
    members=[
        'serviceAccount:task-creator@my-project.iam.gserviceaccount.com',
        'user:developer@mycompany.com'
    ]
)

# Add the binding to the policy
current_policy.bindings.append(new_binding)

# Set the updated policy
updated_policy = client.set_iam_policy(resource=queue_path, policy=current_policy)
print(f'Updated IAM policy with {len(updated_policy.bindings)} bindings')

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'test-queue')

# Test if current credentials have specific permissions
permissions_to_test = [
    'cloudtasks.tasks.create',
    'cloudtasks.tasks.delete',
    'cloudtasks.queues.get',
    'cloudtasks.queues.update'
]

response = client.test_iam_permissions(
    resource=queue_path,
    permissions=permissions_to_test
)

print('Permissions test results:')
for permission in permissions_to_test:
    has_permission = permission in response.permissions
    print(f'- {permission}: {"✓" if has_permission else "✗"}')

# Only proceed with operations if permissions are available
if 'cloudtasks.tasks.create' in response.permissions:
    print('Can create tasks - proceeding with task creation')
    # Create task logic here
else:
    print('Cannot create tasks - insufficient permissions')

from google.cloud import tasks
from google.iam.v1 import policy_pb2
from google.type import expr_pb2

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'conditional-queue')

# Get current policy
policy = client.get_iam_policy(resource=queue_path)

# Create a conditional binding - only allow task creation during business hours
condition = expr_pb2.Expr(
    title='Business hours only',
    description='Allow task creation only during business hours (9 AM to 5 PM UTC)',
    expression='request.time.getHours() >= 9 && request.time.getHours() < 17'
)

conditional_binding = policy_pb2.Binding(
    role='roles/cloudtasks.enqueuer',
    members=['group:developers@mycompany.com'],
    condition=condition
)

policy.bindings.append(conditional_binding)

# Set the updated policy with conditional access
updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
print('Set conditional IAM policy for business hours access')

from google.cloud import tasks

client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'service-queue')

# Grant a service account minimal permissions for task operations
policy = client.get_iam_policy(resource=queue_path)

# Add service account with enqueuer role
service_account_binding = {
    'role': 'roles/cloudtasks.enqueuer',
    'members': [
        'serviceAccount:task-producer@my-project.iam.gserviceaccount.com'
    ]
}

# Add admin role for queue management service account
admin_binding = {
    'role': 'roles/cloudtasks.queueAdmin',
    'members': [
        'serviceAccount:queue-manager@my-project.iam.gserviceaccount.com'
    ]
}

# Add both bindings
policy.bindings.extend([
    policy_pb2.Binding(**service_account_binding),
    policy_pb2.Binding(**admin_binding)
])

updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
print('Configured service account permissions')

from google.cloud import tasks

client = tasks.CloudTasksClient()

# List all available locations for Cloud Tasks
locations_response = client.list_locations(
    name='projects/my-project-id'
)

print('Available Cloud Tasks locations:')
for location in locations_response.locations:
    print(f'- {location.location_id}: {location.display_name}')
    if location.metadata:
        print(f'  Metadata: {location.metadata}')

# Get specific location information
location_name = 'projects/my-project-id/locations/us-central1'
location_info = client.get_location(name=location_name)
print(f'Location details for us-central1: {location_info.display_name}')

from google.cloud import tasks
from google.iam.v1 import policy_pb2

def setup_secure_queue_permissions(client, queue_path):
    """Set up secure IAM permissions for a production queue."""
    
    # Define role-based access
    bindings = [
        # Production service accounts for task creation
        {
            'role': 'roles/cloudtasks.enqueuer',
            'members': [
                'serviceAccount:prod-app@my-project.iam.gserviceaccount.com',
                'serviceAccount:batch-processor@my-project.iam.gserviceaccount.com'
            ]
        },
        
        # Operations team for queue management
        {
            'role': 'roles/cloudtasks.queueAdmin',
            'members': [
                'group:ops-team@mycompany.com',
                'serviceAccount:automation@my-project.iam.gserviceaccount.com'
            ]
        },
        
        # Read-only access for monitoring
        {
            'role': 'roles/cloudtasks.viewer',
            'members': [
                'group:sre-team@mycompany.com',
                'serviceAccount:monitoring@my-project.iam.gserviceaccount.com'
            ]
        }
    ]
    
    # Create policy with defined bindings
    policy = policy_pb2.Policy(
        bindings=[policy_pb2.Binding(**binding) for binding in bindings]
    )
    
    # Set the secure policy
    updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
    
    # Verify permissions were set correctly
    test_permissions = [
        'cloudtasks.tasks.create',
        'cloudtasks.queues.get',
        'cloudtasks.queues.update'
    ]
    
    response = client.test_iam_permissions(
        resource=queue_path,
        permissions=test_permissions
    )
    
    print(f'Secure permissions configured for {queue_path}')
    print(f'Verified permissions: {response.permissions}')
    
    return updated_policy

# Usage
client = tasks.CloudTasksClient()
queue_path = client.queue_path('my-project-id', 'us-central1', 'production-queue')
setup_secure_queue_permissions(client, queue_path)