Merge pull request #11 from oncase/feat/fix-ldap-service

Feat/fix ldap service

.github/workflows/publish.yml

@@ -35,8 +35,8 @@ jobs:
         uses: marvinpinto/[email protected]
         with:
           repo_token: ${{ secrets.GITHUB_TOKEN }}
-          title: 'v1.1.4 Ldap authentication bug fixed'
+          title: 'v1.1.5 LDAP set_users and set_groups fixed'
           prerelease: false
-          automatic_release_tag: v1.1.4
+          automatic_release_tag: v1.1.5
           files: |
             LICENSE

.gitignore

@@ -159,5 +159,6 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
-# Local files
-test.py
+# Local test files
+test.py
+docker-compose.yml

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyscora-wrangler"
-version = "1.1.4"
+version = "1.1.5"
 description = "Python lib for DE"
 authors = ["Oncase <[email protected]>"]
 maintainers = ["Guilherme Morone <[email protected]>"]

pyscora_wrangler/aws/athena/__init__.py

@@ -176,10 +176,10 @@ def _key_gen(meta: Type[table_meta]) -> str:
             boto3_session=session,
         )
 
-        return {"table": table}
+        return {'table': table}
 
     async def _process_async(tables: Type[tables]):
-        logger.info("[athena_refresh_process_async] {0:<30} {1:>20}".format("File", "Completed at"))
+        logger.info('[athena_refresh_process_async] {0:<30} {1:>20}'.format('File', 'Completed at'))
         with ThreadPoolExecutor(max_workers=15) as executor:
             loop = asyncio.get_event_loop()
             tasks = [loop.run_in_executor(executor, _process, *table_meta) for table_meta in tables]

pyscora_wrangler/aws/cognito/__init__.py

@@ -325,7 +325,7 @@ def set_user_password(
     except client.exceptions.UserNotFoundException:
         logger.warning(f'[set_user_password] User {username} does not exists. Skipping...')
     except Exception as err:
-        logger.error(f"[set_user_password] {err}")
+        logger.error(f'[set_user_password] {err}')
 
 
 def authenticate_user(

pyscora_wrangler/aws/dynamodb/__init__.py

@@ -104,7 +104,7 @@ def get_data_by_key(
 
     dynamo_table = resource.Table(table_name)
     response = dynamo_table.query(KeyConditionExpression=Key(key).eq(value))
-    response = response.get("Items", [])
+    response = response.get('Items', [])
 
     if len(response) > 0:
         data = get_data_decoded(response[0]) if decode_data else response[0]

pyscora_wrangler/aws/utils.py

@@ -16,7 +16,7 @@ def get_user_secret_hash(client_id: str, app_client_secret: str, username: str)
         return None
 
     try:
-        dig = hmac.new(app_client_secret.encode("utf-8"), msg=msg.encode("utf-8"), digestmod=hashlib.sha256).digest()
+        dig = hmac.new(app_client_secret.encode('utf-8'), msg=msg.encode('utf-8'), digestmod=hashlib.sha256).digest()
         d2 = base64.b64encode(dig).decode()
 
         return d2

pyscora_wrangler/constants.py

@@ -1,7 +1,7 @@
 # COLORS AND OTHERS FOR LOGGING FORMAT
-GREY = "\x1b[38;20m"
-BLUE = "\x1b[34;1m"
-YELLOW = "\x1b[33;20m"
-RED = "\x1b[31;20m"
-BOLD_RED = "\x1b[31;1m"
-RESET = "\x1b[0m"
+GREY = '\x1b[38;20m'
+BLUE = '\x1b[34;1m'
+YELLOW = '\x1b[33;20m'
+RED = '\x1b[31;20m'
+BOLD_RED = '\x1b[31;1m'
+RESET = '\x1b[0m'

pyscora_wrangler/ldap/README.md

@@ -10,8 +10,8 @@ Requires only a config dictionary. An example with types can be seen below:
 
 ```python
 {
-  "root_dn": "CN=GS_1,OU=Grupos,DC=service,DC=local", # REQUIRED. Type = str.
-  "server": "ldap://localhost.389", # REQUIRED. Type = str.
+  "root_dn": "DC=service,DC=local", # REQUIRED. Type = str.
+  "server": "ldap://localhost:389", # REQUIRED. Type = str.
   "port": 636, # OPTIONAL. Type = int. Default is 389.
   "server_alias": ["service.com.br"] # OPTIONAL. Type = List[str]. Default is [].
 }

pyscora_wrangler/ldap/service/__init__.py

@@ -122,6 +122,7 @@ def auth(self, username: str, password: str) -> bool:
 
                 raise ValueError('Invalid credentials.')
 
+            root_dn = self.ldap_config.get('root_dn')
             port = int(self.ldap_config.get('port', 389))
             server_alias = self.ldap_config.get('server_alias', [])
 
@@ -136,17 +137,17 @@ def auth(self, username: str, password: str) -> bool:
 
             self.__ldap_connection = Connection(
                 server,
-                user=self.__ldap_username,
+                user=f'CN={self.__ldap_username},{root_dn}',
                 password=self.__ldap_password,
                 authentication='SIMPLE',
                 raise_exceptions=False,
             )
 
             if self.__ldap_connection.bind():
                 self.__user_is_authenticated = True
-                logger.info("[auth] Successful bind to ldap server.")
+                logger.info('[auth] Successful bind to ldap server.')
             else:
-                logger.error(f"[auth] Cannot bind to ldap server: {self.__ldap_connection.last_error}.")
+                logger.error(f'[auth] Cannot bind to ldap server: {self.__ldap_connection.last_error}.')
         except Exception as err:
             logger.error(f'[auth] {err}')
 
@@ -162,61 +163,60 @@ def logout(self) -> None:
                 logger.error(f'[logout] {err}')
 
     def get_ldap_groups(self) -> List[str]:
-        """Returns A list containing the ldap groups."""
+        """Returns A list containing the ldap groups"""
 
         return self.__ldap_groups
 
     def get_ldap_users(self) -> List[str]:
-        """Returns A list containing the ldap users."""
+        """Returns A list containing the ldap users"""
 
         return self.__ldap_users
 
-    def set_ldap_users_and_groups(self) -> Tuple[List[str], List[str]] | None:
-        """Set the ldap groups and users.
+    def __set_ldap_arrays(self, search_filter: str) -> List[str]:
+        root_dn = self.ldap_config.get('root_dn', '')
+
+        self.__ldap_connection.search(
+            search_base=root_dn, search_filter=search_filter, search_scope=SUBTREE, size_limit=0
+        )
+
+        arr = []
+        for entry in self.__ldap_connection.entries:
+            arr.append(entry.entry_dn)
+
+        return arr
+
+    def set_ldap_users(self) -> List[str] | None:
+        """Set the ldap users
 
         Returns:
-            Tuple[List[str], List[str]] | None: Returns None if an error occurs in the process. Otherwise, returns a Tuple the lists of the ldap groups and the ldap users.
+            List[str] | None: Returns None if an error occurs in the process. Otherwise, returns the list of the ldap users.
         """
 
         if not self.is_user_authenticated():
-            logger.warning('[set_ldap_users_and_groups] User is not authenticated. Skipping...')
+            logger.warning('[set_ldap_users] User is not authenticated. Skipping...')
             return None
 
-        groups = []
-        root_dn = self.ldap_config.get('root_dn', '')
+        users_search_filter = '(objectClass=person)'
+        users_dn = self.__set_ldap_arrays(search_filter=users_search_filter)
 
-        try:
-            self.__ldap_connection.search(
-                search_base=root_dn,
-                search_filter="(objectclass=*)",
-                search_scope=SUBTREE,
-                attributes=["member"],
-                size_limit=0,
-            )
+        self.__ldap_users = [user.split('=')[1].split(',')[0] for user in users_dn]
 
-            response = json.loads(self.__ldap_connection.response_to_json())
+        return self.get_ldap_users()
 
-            if type(response.get("entries")) == list and len(response.get("entries")) > 0:
-                for entry in response.get("entries"):
-                    for member in entry.member.values:
-                        self.__ldap_connection.search(
-                            search_base=root_dn,
-                            search_filter=f"(distinguishedName={member})",
-                            attributes=["sAMAccountName"],
-                        )
+    def set_ldap_groups(self) -> List[str] | None:
+        """Set the ldap groups
 
-                        user = self.__ldap_connection.entries[0].sAMAccountName.values
-
-                        self.__ldap_users.append(user)
+        Returns:
+            List[str] | None: Returns None if an error occurs in the process. Otherwise, returns the list of the ldap groups.
+        """
 
-                cn_groups = response.get("entries")[0].get("attributes").get("member")
+        if not self.is_user_authenticated():
+            logger.warning('[set_ldap_users] User is not authenticated. Skipping...')
+            return None
 
-                for cn_group in cn_groups:
-                    groups.append(cn_group.split(",")[0].replace("CN=", ""))
+        groups_search_filter = '(objectClass=groupOfNames)'
+        groups_dn = self.__set_ldap_arrays(search_filter=groups_search_filter)
 
-                self.__ldap_groups = groups
-        except Exception as err:
-            logger.error(f'[set_ldap_users_and_groups] {err}')
-            return None
+        self.__ldap_groups = [group.split('=')[1].split(',')[0] for group in groups_dn]
 
-        return self.get_ldap_users(), self.get_ldap_groups()
+        return self.get_ldap_groups()

pyscora_wrangler/utils.py

@@ -81,19 +81,19 @@ def _time_it(*args, **kwargs):
             return func(*args, **kwargs)
         finally:
             end_ = int(round(time() * 1000)) - start
-            print(f"{func.__name__} execution time: {end_ if end_ > 0 else 0} ms")
+            print(f'{func.__name__} execution time: {end_ if end_ > 0 else 0} ms')
 
     return _time_it
 
 
 def get_metadata_from_yaml(file_path: str) -> Any:
     data = []
 
-    with open(file_path, encoding="utf-8") as file:
+    with open(file_path, encoding='utf-8') as file:
         data = yaml.load(file, Loader=yaml.FullLoader)
 
     return data
 
 
 def get_copy_metadata(file_path: str) -> Any:
-    return get_metadata_from_yaml(file_path).get("copy")
+    return get_metadata_from_yaml(file_path).get('copy')