This guide provides a Python script to sync users from a FreeIPA server to a Mautic instance. FreeIPA 4.13.1 includes enhanced API capabilities and security features that improve user management workflows, making it ideal for integration with marketing automation platforms like Mautic.
python-freeipa and requests librariesInstall required Python packages:
pip install python-freeipa requests
Here’s an improved Python script that syncs users from a FreeIPA server to a Mautic instance:
#!/usr/bin/env python3
"""
FreeIPA to Mautic User Sync Script
Sync users from a FreeIPA server to a Mautic instance
"""
import os
import sys
import logging
from typing import List, Dict, Any
from python_freeipa import Client, exceptions
import requests
import argparse
from datetime import datetime
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class FreeIPAMauticSync:
def __init__(self):
# FreeIPA configuration from environment variables
self.ipa_server = os.getenv('IPA_SERVER', 'ipa.example.com')
self.ipa_username = os.getenv('IPA_USERNAME', 'admin')
self.ipa_password = os.getenv('IPA_PASSWORD', '')
# Mautic configuration from environment variables
self.mautic_url = os.getenv('MAUTIC_URL', 'https://mautic.example.com')
self.mautic_username = os.getenv('MAUTIC_USERNAME', '')
self.mautic_password = os.getenv('MAUTIC_PASSWORD', '')
self.mautic_api_key = os.getenv('MAUTIC_API_KEY', '') # Preferred over username/password
# Initialize clients
self.ipa_client = None
self.mautic_session = None
def connect_to_freeipa(self):
"""Establish connection to FreeIPA server"""
try:
self.ipa_client = Client(self.ipa_server)
self.ipa_client.login(self.ipa_username, self.ipa_password)
logger.info("Successfully connected to FreeIPA server")
return True
except exceptions.AuthenticationError as e:
logger.error(f"FreeIPA authentication failed: {e}")
return False
except Exception as e:
logger.error(f"FreeIPA connection failed: {e}")
return False
def connect_to_mautic(self):
"""Establish connection to Mautic API"""
try:
self.mautic_session = requests.Session()
# Set up authentication headers
if self.mautic_api_key:
# Use API key authentication (preferred)
self.mautic_session.headers.update({
'Authorization': f'Basic {self.mautic_api_key}'
})
else:
# Use username/password authentication
self.mautic_session.auth = (self.mautic_username, self.mautic_password)
# Test connection
test_url = f"{self.mautic_url.rstrip('/')}/api/segments/list"
response = self.mautic_session.get(test_url)
if response.status_code == 200:
logger.info("Successfully connected to Mautic API")
return True
else:
logger.error(f"Mautic API connection failed: {response.status_code} - {response.text}")
return False
except Exception as e:
logger.error(f"Mautic connection failed: {e}")
return False
def get_users_from_freeipa(self) -> List[Dict[str, Any]]:
"""Retrieve users from FreeIPA server"""
try:
# Get all users from FreeIPA
result = self.ipa_client.user_find(sizelimit=0)
users = []
for user in result['result']:
# Extract user information
user_info = {
'username': user.get('uid', [None])[0],
'first_name': user.get('givenname', [None])[0],
'last_name': user.get('sn', [None])[0],
'email': user.get('mail', [None])[0],
'full_name': user.get('cn', [None])[0],
'display_name': user.get('displayname', [None])[0],
'title': user.get('title', [None])[0],
'department': user.get('departmentnumber', [None])[0],
'telephone': user.get('telephonenumber', [None])[0],
'mobile': user.get('mobile', [None])[0],
'employee_number': user.get('employeenumber', [None])[0],
'status': 'active' if user.get('nsaccountlock', [False])[0] == False else 'inactive'
}
# Only add users with email addresses
if user_info['email']:
users.append(user_info)
logger.info(f"Retrieved {len(users)} users from FreeIPA")
return users
except Exception as e:
logger.error(f"Error retrieving users from FreeIPA: {e}")
return []
def contact_exists_in_mautic(self, email: str) -> Dict[str, Any]:
"""Check if a contact exists in Mautic by email"""
try:
# Search for contact by email
search_url = f"{self.mautic_url.rstrip('/')}/api/contacts"
params = {
'search': f'email:{email}'
}
response = self.mautic_session.get(search_url, params=params)
if response.status_code == 200:
data = response.json()
if 'contacts' in data and data['contacts']:
# Return the first contact found
contact = next(iter(data['contacts'].values()))
return {'exists': True, 'id': contact['id'], 'contact': contact}
else:
return {'exists': False, 'id': None, 'contact': None}
else:
logger.error(f"Error searching for contact {email} in Mautic: {response.status_code}")
return {'exists': False, 'id': None, 'contact': None}
except Exception as e:
logger.error(f"Error checking if contact {email} exists in Mautic: {e}")
return {'exists': False, 'id': None, 'contact': None}
def create_contact_in_mautic(self, user_data: Dict[str, Any]) -> bool:
"""Create a contact in Mautic"""
try:
contact_url = f"{self.mautic_url.rstrip('/')}/api/contacts/new"
# Prepare contact data
contact_data = {
'email': user_data['email'],
'firstname': user_data['first_name'],
'lastname': user_data['last_name'],
'position': user_data['title'],
'company': user_data['department'],
'phone': user_data['telephone'],
'mobile': user_data['mobile'],
'tags': ['freeipa-sync', 'employee'],
'custom_fields': {
'username': user_data['username'],
'employee_number': user_data['employee_number'],
'status': user_data['status']
}
}
# Remove None values
contact_data = {k: v for k, v in contact_data.items() if v is not None}
response = self.mautic_session.post(contact_url, json=contact_data)
if response.status_code in [200, 201]:
logger.info(f"Successfully created contact {user_data['email']} in Mautic")
return True
else:
logger.error(f"Failed to create contact {user_data['email']} in Mautic: {response.status_code} - {response.text}")
return False
except Exception as e:
logger.error(f"Error creating contact {user_data['email']} in Mautic: {e}")
return False
def update_contact_in_mautic(self, contact_id: int, user_data: Dict[str, Any]) -> bool:
"""Update an existing contact in Mautic"""
try:
contact_url = f"{self.mautic_url.rstrip('/')}/api/contacts/{contact_id}/edit"
# Prepare update data
update_data = {
'firstname': user_data['first_name'],
'lastname': user_data['last_name'],
'position': user_data['title'],
'company': user_data['department'],
'phone': user_data['telephone'],
'mobile': user_data['mobile'],
'tags': ['freeipa-sync', 'employee'],
'custom_fields': {
'username': user_data['username'],
'employee_number': user_data['employee_number'],
'status': user_data['status']
}
}
# Remove None values
update_data = {k: v for k, v in update_data.items() if v is not None}
response = self.mautic_session.put(contact_url, json=update_data)
if response.status_code == 200:
logger.info(f"Successfully updated contact {user_data['email']} (ID: {contact_id}) in Mautic")
return True
else:
logger.error(f"Failed to update contact {user_data['email']} (ID: {contact_id}) in Mautic: {response.status_code} - {response.text}")
return False
except Exception as e:
logger.error(f"Error updating contact {user_data['email']} (ID: {contact_id}) in Mautic: {e}")
return False
def sync_users(self, update_existing: bool = True, tag_segment: str = None) -> int:
"""Main synchronization function"""
# Connect to FreeIPA
if not self.connect_to_freeipa():
logger.error("Cannot proceed without FreeIPA connection")
return 0
# Connect to Mautic
if not self.connect_to_mautic():
logger.error("Cannot proceed without Mautic connection")
return 0
# Get users from FreeIPA
freeipa_users = self.get_users_from_freeipa()
if not freeipa_users:
logger.error("No users retrieved from FreeIPA, aborting sync")
return 0
success_count = 0
# Process each user from FreeIPA
for user_data in freeipa_users:
email = user_data['email']
if not email:
logger.warning("Skipping user with no email address")
continue
# Check if contact exists in Mautic
contact_info = self.contact_exists_in_mautic(email)
if contact_info['exists']:
if update_existing:
if self.update_contact_in_mautic(contact_info['id'], user_data):
success_count += 1
else:
logger.info(f"Contact {email} exists in Mautic, skipping (use update mode to modify)")
success_count += 1 # Count as success since contact exists
else:
if self.create_contact_in_mautic(user_data):
success_count += 1
# Add contacts to segment if specified
if tag_segment:
self.add_contacts_to_segment(tag_segment)
return success_count
def add_contacts_to_segment(self, segment_name: str):
"""Add synced contacts to a specific Mautic segment"""
try:
# This would require additional logic to identify recently synced contacts
# For now, we'll just log that this functionality exists
logger.info(f"Segment tagging functionality would add contacts to '{segment_name}' segment")
except Exception as e:
logger.error(f"Error adding contacts to segment {segment_name}: {e}")
def main():
# Initialize the sync tool
sync_tool = FreeIPAMauticSync()
# Get command-line arguments or use environment variables
parser = argparse.ArgumentParser(description='Sync users from FreeIPA to Mautic')
parser.add_argument('--update', action='store_true', help='Update existing contacts in Mautic')
parser.add_argument('--segment', type=str, help='Add contacts to specified Mautic segment')
parser.add_argument('--dry-run', action='store_true', help='Show what would be done without making changes')
args = parser.parse_args()
if args.dry_run:
logger.info("DRY RUN MODE: No changes will be made to Mautic")
# In dry run mode, we'll just show what would happen
if sync_tool.connect_to_freeipa():
freeipa_users = sync_tool.get_users_from_freeipa()
print(f"Would process {len(freeipa_users)} users from FreeIPA")
for user in freeipa_users[:5]: # Show first 5 users
print(f" - {user['email']}: {user.get('first_name', '')} {user.get('last_name', '')}")
if len(freeipa_users) > 5:
print(f" ... and {len(freeipa_users) - 5} more users")
return 0
# Perform the actual sync
logger.info("Starting FreeIPA to Mautic user synchronization")
success_count = sync_tool.sync_users(update_existing=args.update, tag_segment=args.segment)
logger.info(f"Synchronization completed. Successfully processed {success_count} contacts.")
return success_count
if __name__ == "__main__":
sys.exit(main())
Set the following environment variables to configure the script:
export IPA_SERVER="ipa.example.com"
export IPA_USERNAME="admin"
export IPA_PASSWORD="your_admin_password"
export MAUTIC_URL="https://mautic.example.com"
export MAUTIC_API_KEY="your_mautic_api_key" # Preferred
# OR
export MAUTIC_USERNAME="your_mautic_username"
export MAUTIC_PASSWORD="your_mautic_password"
# Basic sync (create new contacts, skip existing)
python freeipa_mautic_sync.py
# Update existing contacts as well
python freeipa_mautic_sync.py --update
# Add contacts to a specific segment
python freeipa_mautic_sync.py --update --segment="FreeIPA Users"
# Dry run to see what would be done
python freeipa_mautic_sync.py --dry-run
With FreeIPA 4.13.x, consider these additional capabilities:
FreeIPA 4.13.x may include additional user attributes that can be synchronized:
# Additional attributes available in newer versions
additional_attrs = {
'loginshell': user.get('loginshell', [None])[0],
'homedirectory': user.get('homedirectory', [None])[0],
'gecos': user.get('gecos', [None])[0]
}
If you need to sync system accounts as well (4.13.x+ feature):
def get_system_accounts_from_freeipa(self) -> List[Dict[str, Any]]:
"""Retrieve system accounts from FreeIPA server (4.13.x+)"""
try:
result = self.ipa_client._request('sysaccount_find', sizelimit=0)
# Process system accounts similarly to regular users
# ...
except Exception as e:
logger.error(f"Error retrieving system accounts from FreeIPA: {e}")
return []
# Test FreeIPA connectivity
ipa user-find --sizelimit=1
# Test Mautic API connectivity
curl -H "Authorization: Basic $(echo -n 'username:password' | base64)" \
https://mautic.example.com/api/contacts
# Check specific user in FreeIPA
ipa user-show username
If you are configuring Mautic for integration with FreeIPA, please note that we provide a dedicated FreeIPA to Mautic synchronization script that automates the process of exporting users from FreeIPA to Mautic. This integration enables seamless user management between your identity provider and marketing automation platform. For more information about this integration, see this page and the related resources above.