How to Create User-to-Role Relationships

TL;DR

Link users to their roles/permissions in Veza by creating relationship mappings in the PRE_PUSH_OAA hook.

Prerequisites

  • An OAA integration with users and roles sources configured

  • Understanding of the Provider object

  • A PRE_PUSH_OAA hook in modules/hooks.py

  • Both users and roles already mapped to OAA (LocalUser, LocalRole)

Steps

Step 1: Configure Sources in config.yaml

Ensure you have both users and roles sources:

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /users
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email
      name:
        destination_field: name
       role_id:  # Important: include role reference
         destination_field: role_id
         custom_type: STRING  # Custom field (see reference for all types)

   roles:
    destination: LocalRole
    source_type: custom
    connection: my_api
    params:
      resource: /roles
    field_mapping:
      role_id:
        destination_field: unique_id
      role_name:
        destination_field: name

Step 2: Create PRE_PUSH_OAA Hook

In modules/hooks.py, add a PRE_PUSH_OAA hook:

"""Integration hooks for relationship building"""
import logging
from oaa.hooks.decorators import hook, OAAHookEvent

logger = logging.getLogger(__name__)


@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_user_to_role_relationships(sources=None, provider=None, **kwargs):
    """
    Link users to roles using the Provider object.
    
    This hook runs AFTER all OAA objects are created and loaded into memory.
    It's the right place to build relationships between users and roles.
    
    Args:
        sources: Dictionary of all sources with their records
        provider: Provider object with all OAA entities
        **kwargs: Additional arguments
    """
    
    total_users = 0
    total_links = 0
    
    # Iterate through all user records
    for user in sources['users'].records:
        try:
            total_users += 1
            user_id = user.unique_id
            
            # Get the user reference from provider
            user_ref = provider.local_users[user_id]
            
            # If user has a role, add it
            if hasattr(user, 'role_id') and user.role_id:
                try:
                    role_id = str(user.role_id).strip()
                    
                    # Verify role exists in provider
                    if role_id in provider.local_roles:
                        user_ref.add_role(role_id)
                        total_links += 1
                        logger.debug(f"✓ Linked user {user_id} to role {role_id}")
                    else:
                        logger.warning(f"Role {role_id} not found for user {user_id}")
                        
                except (ValueError, AttributeError) as e:
                    logger.error(f"Error processing role for user {user_id}: {e}")
        
        except KeyError:
            logger.warning(f"User {user_id} not found in provider")
        except Exception as e:
            logger.error(f"Error processing user {user_id}: {e}")
    
    logger.info(f"✓ Created {total_links} user-to-role relationships ({total_users} users)")

Step 3: Test the Integration

Validate and run:

# Validate configuration
uv tool run --from oaa-runner oaa app validate --config=config.yaml

# Run integration
uv tool run --from oaa-runner oaa app run --config=config.yaml --save-json

# Check output for relationships
jq '.relationships[] | select(.src_type=="LocalUser" and .dst_type=="LocalRole")' output.json

Advanced Patterns

Multiple Roles Per User

If users can have multiple roles:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_multiple_roles(sources=None, provider=None, **kwargs):
    """Link users to multiple roles"""
    
    for user in sources['users'].records:
        user_id = user.unique_id
        user_ref = provider.local_users[user_id]
        
        # If role_ids is a list or comma-separated string
        if hasattr(user, 'role_ids'):
            role_list = user.role_ids
            
            # Handle comma-separated string
            if isinstance(role_list, str):
                role_list = [r.strip() for r in role_list.split(',')]
            
            # Handle list
            if isinstance(role_list, list):
                for role_id in role_list:
                    if str(role_id) in provider.local_roles:
                        user_ref.add_role(str(role_id))
                        logger.debug(f"Added role {role_id} to {user_id}")

Conditional Roles

Add roles based on conditions:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_conditional_roles(sources=None, provider=None, **kwargs):
    """Add roles based on user attributes"""
    
    for user in sources['users'].records:
        user_ref = provider.local_users[user.unique_id]
        
        # Add admin role if user has admin flag
        if hasattr(user, 'is_admin') and user.is_admin:
            if 'admin' in provider.local_roles:
                user_ref.add_role('admin')
        
        # Add manager role if user has team
        if hasattr(user, 'team_size') and int(user.team_size or 0) > 0:
            if 'manager' in provider.local_roles:
                user_ref.add_role('manager')
        
        # Add based on department
        if hasattr(user, 'department'):
            department_lower = user.department.lower()
            if department_lower in provider.local_roles:
                user_ref.add_role(department_lower)

Derived Roles from Source Data

Create roles dynamically from user data:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_derived_roles(sources=None, provider=None, **kwargs):
    """Derive roles from user attributes"""
    
    role_mapping = {
        'superuser': ['admin', 'power_user'],
        'admin': ['admin'],
        'power_user': ['power_user'],
        'standard': ['user'],
        'readonly': ['viewer']
    }
    
    for user in sources['users'].records:
        user_ref = provider.local_users[user.unique_id]
        
        # Get role list from mapping
        user_type = user.user_type.lower() if hasattr(user, 'user_type') else 'standard'
        roles = role_mapping.get(user_type, ['user'])
        
        # Add all applicable roles
        for role_id in roles:
            if role_id in provider.local_roles:
                user_ref.add_role(role_id)

Production Examples

Pattern 1: Simple Role Assignment (Abnormal Security)

From gxo/abnormal/modules/abnormal_hooks.py - basic user-to-role mapping:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_user_role_relationships(sources=None, provider=None, **kwargs):
    """Link Abnormal users to their assigned roles"""
    
    total_links = 0
    
    for user in sources['users'].records:
        try:
            user_ref = provider.local_users[user.unique_id]
            
            # Add role if present
            if hasattr(user, 'role') and user.role:
                role_id = str(user.role).strip()
                
                if role_id in provider.local_roles:
                    user_ref.add_role(role_id)
                    total_links += 1
                    logger.debug(f"Added role {role_id} to user {user.unique_id}")
                else:
                    logger.warning(f"Role {role_id} not found for user {user.unique_id}")
        
        except (KeyError, AttributeError) as e:
            logger.warning(f"Error linking user role: {e}")
    
    logger.info(f"✓ Created {total_links} user-role links")

Use case: Direct role assignment when users have a single role field.

Pattern 2: Complex Multi-Level Relationships (ThousandEyes)

From netapp/thousandeyes/modules/thousandeyes_hooks.py - roles AND groups:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_user_permissions(sources=None, provider=None, **kwargs):
    """Link ThousandEyes users to roles and account groups"""
    
    for user in sources['users'].records:
        try:
            user_ref = provider.local_users[user.uid]
            
            # Add role (permission level)
            if hasattr(user, 'roleId') and user.roleId:
                role_id = str(user.roleId)
                
                if role_id in provider.local_roles:
                    user_ref.add_role(role_id)
                    logger.debug(f"User {user.uid} assigned to role {role_id}")
                else:
                    logger.warning(f"Role {role_id} not found")
            
            # Add to account group (team/division)
            if hasattr(user, 'accountGroup') and user.accountGroup:
                group_id = f"group_{user.accountGroup}"
                if group_id in provider.local_groups:
                    user_ref.add_group(group_id)
                    logger.debug(f"User {user.uid} added to group {group_id}")
        
        except KeyError as e:
            logger.error(f"User {user.uid} not found in provider: {e}")
    
    logger.info("✓ Completed user permission linking")

Use case: When users need multiple relationship types (roles for permissions, groups for organization).

Pattern 3: Custom Permissions (BeyondTrust)

From netapp/thousandeyes pattern - creating custom permission records:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_custom_permissions(sources=None, provider=None, **kwargs):
    """Create and assign custom permissions based on user attributes"""
    
    # Define permission mappings
    permissions = {
        'admin': {'resource_types': ['all'], 'level': 'full'},
        'power_user': {'resource_types': ['application', 'database'], 'level': 'write'},
        'viewer': {'resource_types': ['application'], 'level': 'read'},
    }
    
    for user in sources['users'].records:
        try:
            user_ref = provider.local_users[user.unique_id]
            
            # Determine user's permission level
            perm_level = 'viewer'  # Default
            if hasattr(user, 'is_admin') and user.is_admin:
                perm_level = 'admin'
            elif hasattr(user, 'is_power_user') and user.is_power_user:
                perm_level = 'power_user'
            
            # Get permission details
            perm_config = permissions.get(perm_level, permissions['viewer'])
            
            # Create custom permission record
            user_ref.add_custom_property('permission_level', perm_level, property_type='string')
            user_ref.add_custom_property('resources', 
                                        ','.join(perm_config['resource_types']), 
                                        property_type='string')
            
            logger.debug(f"User {user.unique_id} assigned {perm_level} permissions")
        
        except Exception as e:
            logger.error(f"Error assigning permissions to {user.unique_id}: {e}")
    
    logger.info("✓ Custom permissions assigned")

Use case: When you need to create calculated permission levels based on multiple user attributes.

Relationship Types

Common relationships you can add:

# Add role (permission level)
user_ref.add_role('admin')

# Add to group (team/department)
user_ref.add_group('engineering')

# Set manager
user_ref.add_manager('manager_user_id')

# Add custom relationship (if supported)
user_ref.add_resource('resource_id')

Debugging Relationships

Add logging to understand what’s happening:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def add_relationships_with_logging(sources=None, provider=None, **kwargs):
    """Debug relationships being created"""
    
    logger.info(f"Available roles: {list(provider.local_roles.keys())}")
    logger.info(f"Available users: {list(provider.local_users.keys())[:10]}")
    
    for user in sources['users'].records[:5]:  # Log first 5
        logger.info(f"Processing user: {user.unique_id}")
        
        if hasattr(user, 'role_id'):
            logger.info(f"  - Role ID: {user.role_id}")
            logger.info(f"  - Role exists: {user.role_id in provider.local_roles}")
        
        try:
            user_ref = provider.local_users[user.unique_id]
            logger.info(f"  - User found in provider: True")
        except KeyError:
            logger.warning(f"  - User found in provider: False")

Troubleshooting

Error: KeyError: user_id not in provider

  • Verify user was successfully created in LocalUser source

  • Check that user_id field is correct

  • Ensure source is properly configured

Relationships not appearing in output

  • Check logs for warnings about missing roles

  • Verify role IDs exist in provider.local_roles

  • Make sure hook is registered in logic_modules

Wrong relationships created

  • Add logging to see what role_ids are being used

  • Check data in sources for unexpected values

  • Verify role_id field mapping is correct

Hook Event Timing

PRE_PUSH_OAA runs at the right time:

1. FETCH & TRANSFORM: Get and transform data
    ↓
2. FIELD_MAPPING: Map fields
    ↓
3. Create OAA objects
    ↓
4. PRE_PUSH_OAA ← You are here!
    ├─ All users loaded
    ├─ All roles loaded
    └─ Provider has all entities
    ↓
5. Push to Veza

Next Steps