Added an email duplication detection feature in the userprincipalname…

… and proxyaddresses mail fields.

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## [2024-11-11]
+- Added an email duplication detection feature in the userprincipalname and proxyaddresses mail fields.
+
 ## [2024-11-09]
 - Add params verify, Useful when you have a firewall that performs SSL inspection, you can mention the firewall certificate in configuration file.
 

azure.conf.exemple

@@ -30,3 +30,6 @@ dry_run=True
 ; db file, the last data sent is stored there.
 dbpath = /root/last_send_azuread.db
 
+;The script includes error detection in the event of duplicate mail on two different objects.
+warning_duplicate_mail_value=True
+

libsync.py

@@ -146,6 +146,7 @@ def __init__(self, smbconf="/etc/samba/smb.conf",url='/var/lib/samba/private/sam
         self.callback_calculated_hashnt = None
         self.callback_calculated_group  = None
         self.callback_calculated_device = None
+        self.warning_duplicate_mail_value = True
 
         # SAMDB
         lp = param.LoadParm()
@@ -295,6 +296,7 @@ def generate_all_dict(self):
         for bdn_user in self.basedn_user:
                 result_user.extend(self.samdb_loc.search(base=bdn_user, expression=r"(&(objectClass=user)(!(objectClass=computer))%s)" % self.custom_filter_user))
 
+        dict_mail_dn={}
         for user in result_user:
 
             # Update if password different in dict mail pwdlastset
@@ -313,7 +315,7 @@ def generate_all_dict(self):
                 hashnt = password[passwordattr][0].hex().upper()
 
             SourceAnchor = self.return_source_anchor(user)
-
+            dn = user['distinguishedName'][0].decode('utf-8')
 
             if int(user["userAccountControl"][0]) & UF_ACCOUNTDISABLE:
                 enabled = False
@@ -348,7 +350,21 @@ def generate_all_dict(self):
                        "proxyAddresses"             : [p.decode('utf-8') for p in user.get("proxyAddresses",[])],
                        "usertype"                   : "User"
                    }
-
+
+            if self.warning_duplicate_mail_value:
+                for testmail in [self.alternate_login_id_attr,'mail','proxyAddresses']:
+                    if not testmail in user:
+                        continue
+                    for v in user[testmail]:
+                        m = v.decode('utf-8').split(':')[-1].strip()
+                        if not m :
+                            continue
+                        if not m in dict_mail_dn:
+                            dict_mail_dn[m] = {dn:[]}
+                        if not dn in dict_mail_dn[m]:
+                            dict_mail_dn[m][dn] = []
+                        dict_mail_dn[m][dn].append(testmail)
+
             if self.callback_calculated_user != None:
                 data = self.callback_calculated_user(sambaobj=self.samdb_loc,entry=user,result=data)
                 SourceAnchor = data['SourceAnchor']
@@ -367,6 +383,9 @@ def generate_all_dict(self):
             self.all_dn[str(user["dn"])]=SourceAnchor
             self.dict_all_users_samba[SourceAnchor] = data
 
+        for t in dict_mail_dn:
+            if len(dict_mail_dn[t]) > 1:
+                write_log_json_data('warning_duplicate_mail_value',{'mail':t,'list_conflicting_objects':dict_mail_dn[t]})
 
         if self.add_device:
             self.dict_all_device_samba={}

run_sync.py

@@ -204,6 +204,9 @@ def run_sync(force=False,from_db=False):
     smb.dry_run = dry_run
     smb.add_device = sync_device
 
+    if config.has_option('common', 'warning_duplicate_mail_value'):
+        smb.warning_duplicate_mail_value = config.getboolean('common', 'warning_duplicate_mail_value')
+
     if not AzureObject.table_exists():
         db.create_tables([AzureObject])