tessl/pypi-grpc-google-iam-v1

gRPC client library for Google Cloud Identity and Access Management (IAM) services with protocol buffer definitions.

—

Pending

Overview

Eval results

Files

Audit and Logging

Name: tessl/pypi-grpc-google-iam-v1
Author: tessl

Support for audit trails and policy change tracking through specialized message types designed for logging IAM operations and policy modifications. These types enable comprehensive audit logging of IAM policy changes and access control events.

Capabilities

Audit Data Structure

Main audit data message containing policy changes for logging purposes.

class AuditData:
    """
    Audit data containing IAM policy changes for logging.
    
    Attributes:
        policy_delta (PolicyDelta): The policy changes being audited
    """
    policy_delta: PolicyDelta

Policy Change Tracking

Comprehensive tracking of changes made to IAM policies including binding modifications and audit configuration updates.

class PolicyDelta:
    """
    Represents a complete set of changes to an IAM policy.
    
    Attributes:
        binding_deltas (List[BindingDelta]): Changes to role bindings
        audit_config_deltas (List[AuditConfigDelta]): Changes to audit configurations
    """
    binding_deltas: List[BindingDelta]
    audit_config_deltas: List[AuditConfigDelta]

Binding Change Tracking

Tracks additions and removals of members from role bindings, including conditional access changes.

class BindingDelta:
    """
    Represents a change to a role binding in an IAM policy.
    
    Attributes:
        action (Action): Whether this is an ADD or REMOVE operation
        role (str): The role being modified (e.g., "roles/viewer")
        member (str): The member being added or removed (e.g., "user:alice@example.com")
        condition (google.type.Expr): Optional conditional logic for the binding
    """
    action: Action
    role: str
    member: str
    condition: google.type.Expr

Audit Configuration Change Tracking

Tracks changes to audit logging configurations for services.

class AuditConfigDelta:
    """
    Represents a change to audit configuration in an IAM policy.
    
    Attributes:
        action (Action): Whether this is an ADD or REMOVE operation
        service (str): Service name (e.g., "storage.googleapis.com")
        exempted_member (str): Member being added/removed from exemption list
        log_type (str): Type of audit log being modified
    """
    action: Action
    service: str
    exempted_member: str
    log_type: str

Action Types

Enumeration of actions that can be performed on policy elements.

class Action:
    """
    Types of actions that can be performed on policy elements.
    """
    ACTION_UNSPECIFIED = 0  # Unspecified action
    ADD = 1                 # Add the element
    REMOVE = 2              # Remove the element

Resource Policy Members

Information about IAM policy principals for resource-specific policies.

class ResourcePolicyMember:
    """
    Contains information about IAM policy members for resource-specific policies.
    
    Attributes:
        iam_policy_name_principal (str): Principal name (output only)
        iam_policy_uid_principal (str): Principal unique identifier (output only)
    """
    iam_policy_name_principal: str  # output only
    iam_policy_uid_principal: str   # output only

Usage Examples

Creating Audit Data for Policy Changes

from google.iam.v1.logging import audit_data_pb2
from google.iam.v1 import policy_pb2

# Create audit data for a binding change
audit_data = audit_data_pb2.AuditData()

# Create policy delta
policy_delta = policy_pb2.PolicyDelta()

# Add binding delta for adding a new member
binding_delta = policy_pb2.BindingDelta()
binding_delta.action = policy_pb2.BindingDelta.Action.ADD
binding_delta.role = "roles/viewer"
binding_delta.member = "user:newuser@example.com"
policy_delta.binding_deltas.append(binding_delta)

# Add binding delta for removing a member
remove_delta = policy_pb2.BindingDelta()
remove_delta.action = policy_pb2.BindingDelta.Action.REMOVE
remove_delta.role = "roles/editor"
remove_delta.member = "user:olduser@example.com"
policy_delta.binding_deltas.append(remove_delta)

# Add audit config delta
audit_config_delta = policy_pb2.AuditConfigDelta()
audit_config_delta.action = policy_pb2.AuditConfigDelta.Action.ADD
audit_config_delta.service = "pubsub.googleapis.com"
audit_config_delta.log_type = "DATA_READ"
audit_config_delta.exempted_member = "serviceAccount:backup@project.iam.gserviceaccount.com"
policy_delta.audit_config_deltas.append(audit_config_delta)

# Set the policy delta in audit data
audit_data.policy_delta.CopyFrom(policy_delta)

print(f"Audit data created with {len(audit_data.policy_delta.binding_deltas)} binding changes")
print(f"and {len(audit_data.policy_delta.audit_config_deltas)} audit config changes")

Logging Policy Changes with Conditions

from google.iam.v1.logging import audit_data_pb2
from google.iam.v1 import policy_pb2
from google.type import expr_pb2

# Create audit data for conditional binding change
audit_data = audit_data_pb2.AuditData()
policy_delta = policy_pb2.PolicyDelta()

# Add conditional binding delta
binding_delta = policy_pb2.BindingDelta()
binding_delta.action = policy_pb2.BindingDelta.Action.ADD
binding_delta.role = "roles/storage.objectViewer"
binding_delta.member = "user:contractor@example.com"

# Add condition for temporary access
condition = expr_pb2.Expr()
condition.title = "Temporary access"
condition.description = "Access expires at end of month"
condition.expression = '''
    request.time < timestamp("2024-01-01T00:00:00Z")
'''
binding_delta.condition.CopyFrom(condition)

policy_delta.binding_deltas.append(binding_delta)
audit_data.policy_delta.CopyFrom(policy_delta)

print("Conditional binding change logged for audit")

Tracking Multiple Policy Changes

from google.iam.v1.logging import audit_data_pb2
from google.iam.v1 import policy_pb2

def create_comprehensive_audit_log(changes):
    """Create audit data for multiple policy changes."""
    
    audit_data = audit_data_pb2.AuditData()
    policy_delta = policy_pb2.PolicyDelta()
    
    # Process binding changes
    for change in changes.get('bindings', []):
        binding_delta = policy_pb2.BindingDelta()
        binding_delta.action = (
            policy_pb2.BindingDelta.Action.ADD if change['action'] == 'add' 
            else policy_pb2.BindingDelta.Action.REMOVE
        )
        binding_delta.role = change['role']
        binding_delta.member = change['member']
        policy_delta.binding_deltas.append(binding_delta)
    
    # Process audit config changes
    for change in changes.get('audit_configs', []):
        audit_delta = policy_pb2.AuditConfigDelta()
        audit_delta.action = (
            policy_pb2.AuditConfigDelta.Action.ADD if change['action'] == 'add'
            else policy_pb2.AuditConfigDelta.Action.REMOVE
        )
        audit_delta.service = change['service']
        audit_delta.log_type = change['log_type']
        if 'exempted_member' in change:
            audit_delta.exempted_member = change['exempted_member']
        policy_delta.audit_config_deltas.append(audit_delta)
    
    audit_data.policy_delta.CopyFrom(policy_delta)
    return audit_data

# Example usage
changes = {
    'bindings': [
        {
            'action': 'add',
            'role': 'roles/viewer',
            'member': 'user:alice@example.com'
        },
        {
            'action': 'remove',
            'role': 'roles/editor',
            'member': 'user:bob@example.com'
        }
    ],
    'audit_configs': [
        {
            'action': 'add',
            'service': 'storage.googleapis.com',
            'log_type': 'DATA_READ',
            'exempted_member': 'serviceAccount:reader@project.iam.gserviceaccount.com'
        }
    ]
}

audit_data = create_comprehensive_audit_log(changes)
print(f"Created audit log with {len(audit_data.policy_delta.binding_deltas)} binding changes")

Integrating with Logging Systems

import json
import logging
from google.iam.v1.logging import audit_data_pb2
from google.iam.v1 import policy_pb2
from google.protobuf.json_format import MessageToDict

# Set up logging
logging.basicConfig(level=logging.INFO)
audit_logger = logging.getLogger('iam_audit')

def log_policy_change(resource_name, audit_data):
    """Log policy changes in structured format."""
    
    # Convert protobuf to dictionary for JSON logging
    audit_dict = MessageToDict(audit_data)
    
    log_entry = {
        'timestamp': '2024-01-15T10:30:00Z',
        'resource': resource_name,
        'event_type': 'iam_policy_change',
        'audit_data': audit_dict,
        'change_count': {
            'binding_changes': len(audit_data.policy_delta.binding_deltas),
            'audit_config_changes': len(audit_data.policy_delta.audit_config_deltas)
        }
    }
    
    # Log as structured JSON
    audit_logger.info(json.dumps(log_entry, indent=2))

# Example usage
audit_data = audit_data_pb2.AuditData()
policy_delta = policy_pb2.PolicyDelta()

binding_delta = policy_pb2.BindingDelta()
binding_delta.action = policy_pb2.BindingDelta.Action.ADD
binding_delta.role = "roles/viewer"
binding_delta.member = "user:alice@example.com"
policy_delta.binding_deltas.append(binding_delta)

audit_data.policy_delta.CopyFrom(policy_delta)

log_policy_change("projects/my-project/topics/my-topic", audit_data)

Resource Policy Member Tracking

from google.iam.v1 import resource_policy_member_pb2

# Track resource policy members (typically used in responses)
def display_policy_members(members):
    """Display resource policy member information."""
    
    for member in members:
        member_info = resource_policy_member_pb2.ResourcePolicyMember()
        member_info.iam_policy_name_principal = member.get('name', '')
        member_info.iam_policy_uid_principal = member.get('uid', '')
        
        print(f"Principal: {member_info.iam_policy_name_principal}")
        print(f"UID: {member_info.iam_policy_uid_principal}")

# Example member data (would come from service response)
member_data = [
    {'name': 'user:alice@example.com', 'uid': '123456789'},
    {'name': 'serviceAccount:app@project.iam.gserviceaccount.com', 'uid': '987654321'}
]

display_policy_members(member_data)

Install with Tessl CLI

npx tessl i tessl/pypi-grpc-google-iam-v1

docs