IAM and Security

def set_iam_policy(self, request: SetIamPolicyRequest = None, **kwargs) -> Policy:
    """
    Sets the access control policy on the specified secret.

    Args:
        request (SetIamPolicyRequest): The request object containing resource and policy.
        resource (str): Required. The resource for which the policy is being specified.
            Format: projects/*/secrets/*.
        policy (Policy): Required. The policy to apply to the resource.

    Returns:
        Policy: The updated IAM policy.

    Raises:
        google.api_core.exceptions.NotFound: If the secret does not exist.
        google.api_core.exceptions.InvalidArgument: If policy is malformed.
        google.api_core.exceptions.PermissionDenied: If insufficient permissions.
    """

from google.iam.v1 import iam_policy_pb2
from google.iam.v1 import policy_pb2

# Create a policy granting secret accessor role
policy = policy_pb2.Policy()

# Add binding for service account
binding = policy_pb2.Binding()
binding.role = "roles/secretmanager.secretAccessor"
binding.members.append("serviceAccount:app@my-project.iam.gserviceaccount.com")
policy.bindings.append(binding)

# Add binding for user
user_binding = policy_pb2.Binding()
user_binding.role = "roles/secretmanager.viewer"
user_binding.members.append("user:developer@company.com")
policy.bindings.append(user_binding)

# Set the policy
request = iam_policy_pb2.SetIamPolicyRequest()
request.resource = "projects/my-project/secrets/api-key"
request.policy = policy

updated_policy = client.set_iam_policy(request=request)
print(f"Policy updated with {len(updated_policy.bindings)} bindings")

def get_iam_policy(self, request: GetIamPolicyRequest = None, **kwargs) -> Policy:
    """
    Gets the access control policy for a secret.

    Args:
        request (GetIamPolicyRequest): The request object.
        resource (str): Required. The resource for which the policy is being requested.
            Format: projects/*/secrets/*.
        options (GetPolicyOptions): Optional. Options for getting the policy.

    Returns:
        Policy: The current IAM policy for the resource.

    Raises:
        google.api_core.exceptions.NotFound: If the secret does not exist.
        google.api_core.exceptions.PermissionDenied: If insufficient permissions.
    """

from google.iam.v1 import iam_policy_pb2

# Get current policy
request = iam_policy_pb2.GetIamPolicyRequest()
request.resource = "projects/my-project/secrets/api-key"

policy = client.get_iam_policy(request=request)
print(f"Current policy has {len(policy.bindings)} bindings")

for binding in policy.bindings:
    print(f"Role: {binding.role}")
    for member in binding.members:
        print(f"  Member: {member}")
    if binding.condition:
        print(f"  Condition: {binding.condition.expression}")

def test_iam_permissions(self, request: TestIamPermissionsRequest = None, **kwargs) -> TestIamPermissionsResponse:
    """
    Returns permissions that a caller has on the specified secret.

    Args:
        request (TestIamPermissionsRequest): The request object.
        resource (str): Required. The resource for which the policy detail is being requested.
            Format: projects/*/secrets/*.
        permissions (Sequence[str]): Required. The set of permissions to check for the resource.

    Returns:
        TestIamPermissionsResponse: Response containing permissions the caller has.

    Raises:
        google.api_core.exceptions.NotFound: If the secret does not exist.
        google.api_core.exceptions.InvalidArgument: If permissions list is invalid.
        google.api_core.exceptions.PermissionDenied: If insufficient permissions to test.
    """

from google.iam.v1 import iam_policy_pb2

# Test multiple permissions
request = iam_policy_pb2.TestIamPermissionsRequest()
request.resource = "projects/my-project/secrets/api-key"
request.permissions.extend([
    "secretmanager.secrets.get",
    "secretmanager.versions.access",
    "secretmanager.secrets.update",
    "secretmanager.secrets.delete"
])

response = client.test_iam_permissions(request=request)
print("Granted permissions:")
for permission in response.permissions:
    print(f"  {permission}")

# Check specific permission
has_access = "secretmanager.versions.access" in response.permissions
print(f"Can access secret data: {has_access}")

# Common Secret Manager roles
ROLES = {
    "roles/secretmanager.admin": "Full access to secrets and versions",
    "roles/secretmanager.secretAccessor": "Access to secret data",  
    "roles/secretmanager.viewer": "Read secret and version metadata",
    "roles/secretmanager.secretVersionAdder": "Add new secret versions",
    "roles/secretmanager.secretVersionManager": "Manage secret version lifecycle"
}

# Common permissions for testing
PERMISSIONS = [
    "secretmanager.secrets.create",      # Create secrets
    "secretmanager.secrets.delete",      # Delete secrets  
    "secretmanager.secrets.get",         # Get secret metadata
    "secretmanager.secrets.list",        # List secrets
    "secretmanager.secrets.update",      # Update secret metadata
    "secretmanager.versions.access",     # Access secret data
    "secretmanager.versions.add",        # Add secret versions
    "secretmanager.versions.destroy",    # Destroy versions
    "secretmanager.versions.disable",    # Disable versions
    "secretmanager.versions.enable",     # Enable versions
    "secretmanager.versions.get",        # Get version metadata
    "secretmanager.versions.list"        # List versions
]

from google.type import expr_pb2

# Create conditional binding
binding = policy_pb2.Binding()
binding.role = "roles/secretmanager.secretAccessor"
binding.members.append("serviceAccount:app@my-project.iam.gserviceaccount.com")

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

policy.bindings.append(binding)

def add_member_to_policy(policy, role, member):
    """Add a member to a specific role in the policy."""
    for binding in policy.bindings:
        if binding.role == role:
            if member not in binding.members:
                binding.members.append(member)
            return
    
    # Create new binding if role doesn't exist
    new_binding = policy_pb2.Binding()
    new_binding.role = role
    new_binding.members.append(member)
    policy.bindings.append(new_binding)

def remove_member_from_policy(policy, role, member):
    """Remove a member from a specific role in the policy."""
    for binding in policy.bindings:
        if binding.role == role and member in binding.members:
            binding.members.remove(member)
            break

# Usage example
policy = client.get_iam_policy(request=get_request)
add_member_to_policy(policy, "roles/secretmanager.secretAccessor", 
                    "user:newuser@company.com")

# Update the policy
set_request = iam_policy_pb2.SetIamPolicyRequest()
set_request.resource = "projects/my-project/secrets/api-key"
set_request.policy = policy
client.set_iam_policy(request=set_request)

# Grant minimal necessary permissions
def setup_app_access(secret_resource, service_account):
    """Set up minimal access for an application service account."""
    policy = client.get_iam_policy(resource=secret_resource)
    
    # Only grant secretAccessor, not admin
    add_member_to_policy(policy, 
                        "roles/secretmanager.secretAccessor",
                        f"serviceAccount:{service_account}")
    
    client.set_iam_policy(resource=secret_resource, policy=policy)

def audit_secret_access(secret_resource):
    """Audit who has access to a secret."""
    policy = client.get_iam_policy(resource=secret_resource)
    
    access_report = {
        "secret": secret_resource,
        "bindings": []
    }
    
    for binding in policy.bindings:
        binding_info = {
            "role": binding.role,
            "members": list(binding.members),
            "has_condition": bool(binding.condition)
        }
        if binding.condition:
            binding_info["condition"] = binding.condition.expression
        
        access_report["bindings"].append(binding_info)
    
    return access_report

class SetIamPolicyRequest:
    """
    Request message for SetIamPolicy method.
    
    Attributes:
        resource (str): Required. The resource for which the policy is being specified.
        policy (Policy): Required. The policy to apply.
        update_mask (FieldMask): Optional. Fields to update.
    """
    resource: str
    policy: Policy
    update_mask: field_mask_pb2.FieldMask

class GetIamPolicyRequest:
    """
    Request message for GetIamPolicy method.
    
    Attributes:
        resource (str): Required. The resource for which the policy is being requested.
        options (GetPolicyOptions): Optional. Options for getting the policy.
    """
    resource: str
    options: GetPolicyOptions

class TestIamPermissionsRequest:
    """
    Request message for TestIamPermissions method.
    
    Attributes:
        resource (str): Required. The resource for which the policy detail is being requested.
        permissions (Sequence[str]): Required. The set of permissions to check.
    """
    resource: str
    permissions: Sequence[str]

class TestIamPermissionsResponse:
    """
    Response message for TestIamPermissions method.
    
    Attributes:
        permissions (Sequence[str]): The permissions that the caller has on the resource.
    """
    permissions: Sequence[str]