IAM Policy Management

def get_iam_policy(
    request: GetIamPolicyRequest = None,
    *,
    resource: str = None,
    retry: OptionalRetry = DEFAULT,
    timeout: Union[float, object] = DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> Policy:
    """
    Gets the access control policy for a connection resource.
    
    Parameters:
    - request: The request object containing resource and options
    - resource: Required. Resource name of the connection in format 'projects/{project_id}/locations/{location_id}/connections/{connection_id}'
    - retry: Retry configuration for the request
    - timeout: Timeout for the request in seconds
    - metadata: Additional metadata to send with the request
    
    Returns:
    Policy: The IAM policy for the resource
    
    Raises:
    google.api_core.exceptions.NotFound: If the connection does not exist
    google.api_core.exceptions.PermissionDenied: If caller lacks permission to get the policy
    google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

from google.cloud.bigquery_connection import ConnectionServiceClient

client = ConnectionServiceClient()

# Get IAM policy for a connection
resource = "projects/my-project/locations/us-central1/connections/my-connection"
policy = client.get_iam_policy(resource=resource)

print(f"Policy version: {policy.version}")
print(f"ETag: {policy.etag}")
print("Bindings:")
for binding in policy.bindings:
    print(f"  Role: {binding.role}")
    print(f"  Members: {', '.join(binding.members)}")
    if binding.condition:
        print(f"  Condition: {binding.condition.expression}")

def set_iam_policy(
    request: SetIamPolicyRequest = None,
    *,
    resource: str = None,
    policy: Policy = None,
    retry: OptionalRetry = DEFAULT,
    timeout: Union[float, object] = DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> Policy:
    """
    Sets the access control policy for a connection resource.
    
    Parameters:
    - request: The request object containing resource and policy
    - resource: Required. Resource name of the connection
    - policy: Required. The policy to be applied to the resource
    - retry: Retry configuration for the request
    - timeout: Timeout for the request in seconds
    - metadata: Additional metadata to send with the request
    
    Returns:
    Policy: The updated IAM policy
    
    Raises:
    google.api_core.exceptions.NotFound: If the connection does not exist
    google.api_core.exceptions.PermissionDenied: If caller lacks permission to set the policy
    google.api_core.exceptions.InvalidArgument: If the policy is invalid
    google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

from google.cloud.bigquery_connection import ConnectionServiceClient
from google.iam.v1 import policy_pb2

client = ConnectionServiceClient()
resource = "projects/my-project/locations/us-central1/connections/my-connection"

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

# Create new policy with additional binding
new_policy = policy_pb2.Policy()
new_policy.version = current_policy.version
new_policy.etag = current_policy.etag

# Copy existing bindings
for binding in current_policy.bindings:
    new_binding = new_policy.bindings.add()
    new_binding.role = binding.role
    new_binding.members[:] = binding.members
    if binding.condition:
        new_binding.condition.CopyFrom(binding.condition)

# Add new binding for BigQuery Connection User role
new_binding = new_policy.bindings.add()
new_binding.role = "roles/bigquery.connectionUser"
new_binding.members.append("user:analyst@example.com")
new_binding.members.append("group:data-team@example.com")

# Set the updated policy
updated_policy = client.set_iam_policy(
    resource=resource,
    policy=new_policy
)

print(f"Updated policy version: {updated_policy.version}")

from google.iam.v1 import iam_policy_pb2, policy_pb2

# Create SetIamPolicyRequest for more control
request = iam_policy_pb2.SetIamPolicyRequest()
request.resource = resource

# Build policy with conditional binding
policy = policy_pb2.Policy()
policy.version = 3  # Required for conditions

# Add conditional binding
binding = policy.bindings.add()
binding.role = "roles/bigquery.connectionUser"
binding.members.append("user:temp-analyst@example.com")

# Add condition (access only during business hours)
condition = binding.condition
condition.title = "Business hours only"
condition.description = "Access restricted to business hours (9 AM - 5 PM UTC)"
condition.expression = '''
request.time.getHours() >= 9 && request.time.getHours() < 17
'''

request.policy.CopyFrom(policy)

# Set policy with condition
updated_policy = client.set_iam_policy(request=request)

def test_iam_permissions(
    request: TestIamPermissionsRequest = None,
    *,
    resource: str = None,
    permissions: Optional[MutableSequence[str]] = None,
    retry: OptionalRetry = DEFAULT,
    timeout: Union[float, object] = DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> TestIamPermissionsResponse:
    """
    Tests whether the caller has the specified permissions on a connection resource.
    
    Parameters:
    - request: The request object containing resource and permissions to test
    - resource: Required. Resource name of the connection
    - permissions: Required. List of permissions to test
    - retry: Retry configuration for the request
    - timeout: Timeout for the request in seconds
    - metadata: Additional metadata to send with the request
    
    Returns:
    TestIamPermissionsResponse: The permissions that the caller has
    
    Raises:
    google.api_core.exceptions.NotFound: If the connection does not exist
    google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

from google.cloud.bigquery_connection import ConnectionServiceClient

client = ConnectionServiceClient()
resource = "projects/my-project/locations/us-central1/connections/my-connection"

# Test specific permissions
permissions_to_test = [
    "bigquery.connections.get",
    "bigquery.connections.update",
    "bigquery.connections.delete",
    "bigquery.connections.use"
]

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

print("Permissions granted:")
for permission in response.permissions:
    print(f"  - {permission}")

# Check if specific permission is granted
if "bigquery.connections.use" in response.permissions:
    print("✓ Can use this connection")
else:
    print("✗ Cannot use this connection")

# Test permissions on multiple connections
connections = [
    "projects/my-project/locations/us-central1/connections/conn-1",
    "projects/my-project/locations/us-central1/connections/conn-2",
    "projects/my-project/locations/us-central1/connections/conn-3"
]

permissions_to_test = ["bigquery.connections.use", "bigquery.connections.get"]

for resource in connections:
    try:
        response = client.test_iam_permissions(
            resource=resource,
            permissions=permissions_to_test
        )
        print(f"{resource}: {len(response.permissions)}/{len(permissions_to_test)} permissions")
    except Exception as e:
        print(f"{resource}: Error - {e}")

# Core connection permissions
CONNECTION_PERMISSIONS = [
    "bigquery.connections.create",    # Create new connections
    "bigquery.connections.delete",    # Delete connections
    "bigquery.connections.get",       # Read connection metadata
    "bigquery.connections.list",      # List connections in a location
    "bigquery.connections.update",    # Modify connection configuration
    "bigquery.connections.use",       # Use connection in BigQuery queries
]

# IAM management permissions  
IAM_PERMISSIONS = [
    "bigquery.connections.getIamPolicy",    # Read IAM policy
    "bigquery.connections.setIamPolicy",    # Modify IAM policy
]

# Grant minimal permissions needed
# For analysts who only need to query data:
analyst_policy = {
    "role": "roles/bigquery.connectionUser",
    "members": ["group:analysts@example.com"]
}

# For data engineers who manage connections:
admin_policy = {
    "role": "roles/bigquery.connectionAdmin", 
    "members": ["group:data-engineers@example.com"]
}

# Time-based access restrictions
time_condition = {
    "title": "Business hours only",
    "description": "Access restricted to 9 AM - 5 PM UTC weekdays",
    "expression": '''
    request.time.getHours() >= 9 && 
    request.time.getHours() < 17 &&
    request.time.getDayOfWeek() >= 1 && 
    request.time.getDayOfWeek() <= 5
    '''
}

# IP-based access restrictions  
ip_condition = {
    "title": "Corporate network only",
    "description": "Access restricted to corporate IP ranges",
    "expression": '''
    inIpRange(origin.ip, '10.0.0.0/16') ||
    inIpRange(origin.ip, '192.168.1.0/24')
    '''
}

def audit_connection_policies(project_id: str, location: str):
    """Audit IAM policies for all connections."""
    client = ConnectionServiceClient()
    parent = f"projects/{project_id}/locations/{location}"
    
    connections = client.list_connections(parent=parent)
    
    for connection in connections:
        policy = client.get_iam_policy(resource=connection.name)
        
        print(f"Connection: {connection.friendly_name}")
        print(f"  Policy version: {policy.version}")
        print(f"  Bindings: {len(policy.bindings)}")
        
        for binding in policy.bindings:
            print(f"    Role: {binding.role}")
            print(f"    Members: {len(binding.members)}")
            if binding.condition:
                print(f"    Condition: {binding.condition.title}")

from google.iam.v1 import iam_policy_pb2

class GetIamPolicyRequest:
    """Request message for GetIamPolicy method."""
    resource: str  # Required. Resource name of the connection
    options: GetPolicyOptions  # Optional. Policy format options

class SetIamPolicyRequest:
    """Request message for SetIamPolicy method."""
    resource: str  # Required. Resource name of the connection
    policy: Policy  # Required. The policy to be applied
    update_mask: FieldMask  # Optional. Update mask for partial updates

class TestIamPermissionsRequest:
    """Request message for TestIamPermissions method."""
    resource: str  # Required. Resource name of the connection
    permissions: MutableSequence[str]  # Required. Permissions to test

from google.iam.v1 import policy_pb2, iam_policy_pb2

class Policy:
    """IAM policy for a resource."""
    version: int  # Policy format version
    bindings: MutableSequence[Binding]  # Access control bindings
    audit_configs: MutableSequence[AuditConfig]  # Audit configuration
    etag: bytes  # Version identifier for optimistic concurrency

class Binding:
    """Associates members with a role."""
    role: str  # IAM role (e.g., "roles/bigquery.connectionUser")
    members: MutableSequence[str]  # List of identities
    condition: Expr  # Optional condition for conditional binding

class TestIamPermissionsResponse:
    """Response message for TestIamPermissions method."""
    permissions: MutableSequence[str]  # Permissions that the caller has

# User accounts
"user:analyst@example.com"

# Service accounts  
"serviceAccount:bigquery-sa@my-project.iam.gserviceaccount.com"

# Google Groups
"group:data-team@example.com"

# Google Workspace domains
"domain:example.com"

# All authenticated users
"allAuthenticatedUsers"

# Public access (not recommended for connections)
"allUsers"