Merge pull request #11 from PaperMtn/feature/username-password-check

Version 3.2.0 Release

Author: PaperMtn

CHANGELOG.md

@@ -1,3 +1,16 @@
+## [3.2.0] - 2024-08-14
+### Added
+- Functionality to search for users who are using their username as the password
+  - Converts the users username into the following formats:
+    - All uppercase
+    - All lowercase
+    - Remove dot "."
+    - camelCase (E.g. johnSmith)
+    - PascalCase (E.g. JohnSmith)
+
+### Fixed
+- SUCCESS level logging not properly working for JSON output
+
 ## [3.1.0] - 2024-08-13
 ### Added
 - Added new functionality to enhance the custom passwords passed to lil-pwny

README.md

@@ -17,6 +17,7 @@ More information about Lil Pwny can be found [on my blog](https://papermtn.co.uk
 - **Custom Password Auditing**: Ability to provide a list of your own custom passwords to check AD users against. This allows you to check user passwords against passwords relevant to your organisation that you suspect people might be using. 
   - Pass a .txt file with the plaintext passwords you want to search for, these are then NTLM hashed and AD hashes are then compared with this as well as the HIBP hashes.
 - **Detect Duplicates**: Return a list of accounts using the same passwords. Useful for finding users using the same password for their administrative and standard accounts.
+- **Username as Password**: Detect users that are using their username, or variations of it, as their password.
 - **Obfuscated Output**: Obfuscate hashes in output, for if you don't want to handle or store live user NTLM hashes.
 
 ### Custom Password List Enhancement
@@ -29,6 +30,20 @@ Lil Pwny provides the functionality to enhance your custom password list by addi
 - Passwords with dates appended starting from the year 1950 up to 10 years from today's date (e.g. `password1950`, `password2034`)
 
 A custom password list of 100 plaintext passwords generates 49848660 variations.
+
+### Usernames in Passwords
+Lil Pwny looks for users that are using variations of their username as their password.
+
+It converts the users username into the following formats:
+
+- All uppercase
+- All lowercase
+- Remove dot "."
+- camelCase (E.g. johnSmith)
+- PascalCase (E.g. JohnSmith)
+
+These are then converted to NTLM hashes, and audited against the AD hashes
+
 ## Resources
 This application has been developed to make the most of multiprocessing in Python, with the aim of it working as fast as possible on consumer level hardware.
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "lil-pwny"
-version = "3.1.0"
+version = "3.2.0"
 description = "Fast offline auditing of Active Directory passwords using Python and multiprocessing"
 authors = ["PaperMtn <papermtn@protonmail.com>"]
 license = "GPL-3.0"

src/lil_pwny/__init__.py

@@ -6,9 +6,11 @@
 import traceback
 from datetime import timedelta
 from importlib import metadata
+from typing import List, Dict
 
 from lil_pwny import password_audit, hashing
-from lil_pwny.custom_password_enhancer import CustomPasswordEnhancer
+from lil_pwny.variant_generators.custom_variant_generator import CustomVariantGenerator
+from lil_pwny.variant_generators.username_variant_generator import UsernameVariantGenerator
 from lil_pwny.exceptions import FileReadError
 from lil_pwny.loggers import JSONLogger, StdoutLogger
 
@@ -40,18 +42,58 @@ def get_readable_file_size(file_path: str) -> str:
     """
 
     file_size_bytes = os.path.getsize(file_path)
+    for unit in ['bytes', 'KB', 'MB', 'GB']:
+        if file_size_bytes < 1024:
+            return f'{file_size_bytes:.2f} {unit}'
+        file_size_bytes /= 1024
 
-    if file_size_bytes < 1024:  # Less than 1 KB
-        return f"{file_size_bytes} bytes"
-    elif file_size_bytes < 1024 ** 2:  # Less than 1 MB
-        file_size_kb = file_size_bytes / 1024
-        return f"{file_size_kb:.2f} KB"
-    elif file_size_bytes < 1024 ** 3:  # Less than 1 GB
-        file_size_mb = file_size_bytes / (1024 ** 2)
-        return f"{file_size_mb:.2f} MB"
-    else:  # 1 GB or more
-        file_size_gb = file_size_bytes / (1024 ** 3)
-        return f"{file_size_gb:.2f} GB"
+
+def write_hash_temp_file(hash_list: List[str]) -> str:
+    """ Writes a list of hashes to a temporary file and returns the file path.
+
+        Args:
+            hash_list: A list of hash strings to be written to the temporary file.
+        Returns:
+            str: The file path of the temporary file containing the hashes.
+    """
+
+    with tempfile.NamedTemporaryFile('w', delete=False) as temp_file:
+        temp_file.write('\n'.join(hash_list))
+    return temp_file.name
+
+
+def find_matches(log_handler: JSONLogger or StdoutLogger,
+                 filepath: str,
+                 ad_user_hashes: Dict[str, List[str]],
+                 finding_type: str,
+                 obfuscated: bool,
+                 logging_type: str) -> int:
+    """ Searches for matches between Active Directory user hashes and a provided hash file, logs the results,
+        and returns the number of matches found.
+
+        Args:
+            log_handler: The logger instance used to log messages.
+            filepath: The path to the file containing the hash data to compare against.
+            ad_user_hashes: A dictionary of NTLM hashes from Active Directory users.
+            finding_type: The type of match being searched for (e.g., 'hibp', 'custom', 'username').
+            obfuscated: Whether to obfuscate the matches found by hashing with a random salt.
+            logging_type: The type of logging output to use ('stdout', 'json', etc.).
+        Returns:
+            The number of matches found.
+    """
+
+    matches = password_audit.search(
+        log_handler=log_handler,
+        hibp_hashes_filepath=filepath,
+        ad_user_hashes=ad_user_hashes,
+        finding_type=finding_type,
+        obfuscated=obfuscated)
+    number_of_matches = len(matches)
+    if logging_type != 'stdout':
+        for match in matches:
+            log_handler.log('NOTIFY', match, notify_type=finding_type)
+
+    return number_of_matches
 
 
 def main():
@@ -143,6 +185,22 @@ def main():
             logger.log('CRITICAL', f'Error loading AD user hashes: {str(e)}')
             sys.exit(1)
 
+        # Check username variations
+        logger.log('SUCCESS', f'Finding users using passwords that are a variation of their username...')
+        username_variants = UsernameVariantGenerator().generate_variations(ad_users)
+        logger.log('DEBUG', f'{len(username_variants)} username variants generated ')
+        username_hashes = hasher.get_hashes(username_variants)
+        logger.log('DEBUG', f'Converting username variants to NTLM hashes ')
+        username_temp_filepath = write_hash_temp_file(username_hashes)
+
+        username_count = find_matches(
+            log_handler=logger,
+            filepath=username_temp_filepath,
+            ad_user_hashes=ad_users,
+            finding_type='username',
+            obfuscated=obfuscate,
+            logging_type=logging_type)
+
         # Check HIBP file size
         try:
             logger.log('SUCCESS', f'Size of HIBP file provided {get_readable_file_size(hibp_file)}')
@@ -153,16 +211,13 @@ def main():
         # Compare AD users against HIBP hashes
         logger.log('SUCCESS', f'Comparing {ad_lines} AD users against HIBP compromised passwords...')
         try:
-            hibp_results = password_audit.search(
+            hibp_count = find_matches(
                 log_handler=logger,
-                hibp_hashes_filepath=hibp_file,
+                filepath=hibp_file,
                 ad_user_hashes=ad_users,
                 finding_type='hibp',
-                obfuscated=obfuscate)
-            hibp_count = len(hibp_results)
-            if logging_type != 'stdout':
-                for hibp_match in hibp_results:
-                    logger.log('NOTIFY', hibp_match, notify_type='hibp')
+                obfuscated=obfuscate,
+                logging_type=logging_type)
         except FileNotFoundError as e:
             logger.log('CRITICAL', f'HIBP file not found: {e.filename}')
             sys.exit(1)
@@ -183,66 +238,53 @@ def main():
                     custom_count = 0
                     variants_count = 0
                     logger.log('INFO', 'Enhancing custom password list by adding variations...')
-                    custom_client = CustomPasswordEnhancer(min_password_length=int(custom_enhance))
+                    custom_client = CustomVariantGenerator(min_password_length=int(custom_enhance))
                     for custom_pwd in custom_passwords:
                         logger.log('DEBUG', f'Generating variants for `{custom_pwd}`...')
                         temp_custom_passwords = custom_client.enhance_password(custom_pwd)
+
                         logger.log('DEBUG', 'Converting custom passwords to NTLM hashes...')
                         custom_password_hashes = hasher.get_hashes(temp_custom_passwords)
                         variants_count += len(custom_password_hashes)
                         logger.log('SUCCESS', f'Generated {len(custom_password_hashes)} variants for `{custom_pwd}`')
-                        with tempfile.NamedTemporaryFile('w', delete=False) as temp_file:
-                            for h in custom_password_hashes:
-                                temp_file.write(f'{h}\n')
-                            temp_file_path = temp_file.name
-                        logger.log('DEBUG', f'Custom hashes written to temp file {temp_file_path}')
 
+                        custom_temp_file_path = write_hash_temp_file(custom_password_hashes)
+                        logger.log('DEBUG', f'Custom hashes written to temp file {custom_temp_file_path}')
                         logger.log('INFO', f'Comparing {ad_lines} Active Directory'
                                            f' users against {len(custom_password_hashes)} custom password hashes...')
-                        custom_matches = password_audit.search(
+
+                        custom_count += find_matches(
                             log_handler=logger,
-                            hibp_hashes_filepath=temp_file_path,
+                            filepath=custom_temp_file_path,
                             ad_user_hashes=ad_users,
                             finding_type='custom',
-                            obfuscated=obfuscate)
-                        os.remove(temp_file_path)
-                        logger.log('DEBUG', f'Temp file {temp_file_path} deleted')
-                        custom_count += len(custom_matches)
-                        if logging_type != 'stdout':
-                            for result in custom_matches:
-                                logger.log('NOTIFY', result, notify_type='custom')
+                            obfuscated=obfuscate,
+                            logging_type=logging_type)
+                        os.remove(custom_temp_file_path)
+                        logger.log('DEBUG', f'Temp file {custom_temp_file_path} deleted')
                 else:
                     logger.log('DEBUG', 'Converting custom passwords to NTLM hashes...')
                     custom_password_hashes = hasher.get_hashes(custom_passwords)
-                    with tempfile.NamedTemporaryFile('w', delete=False) as temp_file:
-                        for h in custom_password_hashes:
-                            temp_file.write(f'{h}\n')
-                        temp_file_path = temp_file.name
-                    logger.log('DEBUG', f'Custom hashes written to temp file {temp_file_path}')
+                    custom_temp_file_path = write_hash_temp_file(custom_password_hashes)
+                    logger.log('DEBUG', f'Custom hashes written to temp file {custom_temp_file_path}')
 
                     logger.log('INFO', f'Comparing {ad_lines} Active Directory'
                                        f' users against {len(custom_password_hashes)} custom password hashes...')
-                    custom_matches = password_audit.search(
+                    custom_count += find_matches(
                         log_handler=logger,
-                        hibp_hashes_filepath=temp_file_path,
+                        filepath=custom_temp_file_path,
                         ad_user_hashes=ad_users,
                         finding_type='custom',
-                        obfuscated=obfuscate)
-                    os.remove(temp_file_path)
-                    logger.log('DEBUG', f'Temp file {temp_file_path} deleted')
-                    custom_count = len(custom_matches)
-                    if logging_type != 'stdout':
-                        for result in custom_matches:
-                            logger.log('NOTIFY', result, notify_type='custom')
+                        obfuscated=obfuscate,
+                        logging_type=logging_type)
+                    os.remove(custom_temp_file_path)
+                    logger.log('DEBUG', f'Temp file {custom_temp_file_path} deleted')
             except FileNotFoundError as e:
                 logger.log('CRITICAL', f'Custom password file not found: {e.filename}')
                 sys.exit(1)
             except Exception as e:
                 logger.log('CRITICAL', f'Error during custom password search: {str(e)}')
                 sys.exit(1)
-            finally:
-                if os.path.exists(temp_file_path):
-                    os.remove(temp_file_path)
 
         # Handle duplicates if requested
         duplicate_count = 0
@@ -262,6 +304,7 @@ def main():
 
         logger.log('SUCCESS', 'Audit completed')
         logger.log('SUCCESS', f'Total compromised passwords: {total_comp_count}')
+        logger.log('SUCCESS', f'Passwords matching a variation of the username: {username_count}')
         logger.log('SUCCESS', f'Passwords matching HIBP: {hibp_count}')
         logger.log('SUCCESS', f'Passwords matching custom password dictionary: {custom_count}')
         if custom_enhance:

src/lil_pwny/loggers.py

@@ -42,6 +42,10 @@ def log(self,
             message = f'CUSTOM_MATCH: \n' \
                       f'    ACCOUNT: {message.get("username").lower()} HASH: {message.get("hash")} PASSWORD: {message.get("plaintext_password")} OBFUSCATED: {message.get("obfuscated")}'
             mes_type = 'CUSTOM'
+        if notify_type == "username":
+            message = f'USERNAME_MATCH: \n' \
+                      f'    ACCOUNT: {message.get("username").lower()} HASH: {message.get("hash")} PASSWORD: {message.get("plaintext_password")} OBFUSCATED: {message.get("obfuscated")}'
+            mes_type = 'USERNAME'
         if notify_type == "duplicate":
             message = 'DUPLICATE: \n' \
                       f'    ACCOUNTS: {message.get("users")} HASH: {message.get("hash")} OBFUSCATED: {message.get("obfuscated")}'
@@ -116,6 +120,12 @@ def log_to_stdout(self,
                 key_color = Fore.YELLOW
                 style = Style.NORMAL
                 mes_type = '!'
+            elif mes_type == "USERNAME":
+                base_color = Fore.BLUE
+                high_color = Fore.BLUE
+                key_color = Fore.BLUE
+                style = Style.NORMAL
+                mes_type = '!'
 
             # Make log level word/symbol coloured
             type_colorer = re.compile(r'([A-Z]{3,})', re.VERBOSE)
@@ -195,6 +205,9 @@ def __init__(self, name: str = 'lil pwny', log_queue: Queue = None, **kwargs):
         self.notify_format = logging.Formatter(
             '{"localtime": "%(asctime)s", "level": "NOTIFY", "source": "%(name)s", "match_type": "%(type)s", '
             '"detection_data": %(message)s}')
+        self.success_format = logging.Formatter(
+            '{"localtime": "%(asctime)s", "level": "SUCCESS", "source": "%(name)s", '
+            '"detection_data": %(message)s}')
         self.info_format = logging.Formatter(
             '{"localtime": "%(asctime)s", "level": "%(levelname)s", "source": "%(name)s", "message":'
             ' %(message)s}')
@@ -218,6 +231,11 @@ def log(self, level: str, log_data: str or Dict, **kwargs):
             self.logger.info(
                 json.dumps(log_data, cls=EnhancedJSONEncoder),
                 extra={'type': kwargs.get('notify_type', '')})
+        elif level.upper() == 'SUCCESS':
+            self.handler.setFormatter(self.success_format)
+            self.logger.info(
+                json.dumps(log_data, cls=EnhancedJSONEncoder),
+                extra={'type': kwargs.get('notify_type', '')})
         elif level.upper() in ['INFO', 'DEBUG']:
             self.handler.setFormatter(self.info_format)
             self.logger.log(getattr(logging, level.upper()), json.dumps(log_data))

src/lil_pwny/variant_generators/__init__.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 
 
-class CustomPasswordEnhancer:
+class CustomVariantGenerator:
     """ Enhances the custom password with additional variations
     """
 

src/lil_pwny/custom_password_enhancer.py renamed to src/lil_pwny/variant_generators/custom_variant_generator.py

@@ -0,0 +1,41 @@
+from typing import Dict, List
+
+
+class UsernameVariantGenerator:
+
+    def generate_variations(self, ad_user_list: Dict[str, List[str]]) -> List[str]:
+        """ Generates variations of usernames based on specific rules.
+            - All uppercase
+            - All lowercase
+            - Remove dot "."
+            - camelCase
+            - PascalCase
+
+        Args:
+            ad_user_list: A dictionary where keys are NTLM hashes and values are lists of usernames.
+        Returns:
+            List: A list of generated username variations.
+        """
+
+        variations = []
+
+        for ntlm_hash, usernames in ad_user_list.items():
+            for uname in usernames:
+                if '.' in uname:
+                    split_uname = uname.split('.')
+
+                    if len(split_uname) > 1:
+                        camel_uname = split_uname[0].lower() + ''.join(part.capitalize() for part in split_uname[1:])
+                        variations.append(camel_uname)
+
+                        pascal_uname = ''.join(part.capitalize() for part in split_uname)
+                        variations.append(pascal_uname)
+
+                    stripped_uname = uname.replace('.', '')
+                    variations.append(stripped_uname.upper())
+                    variations.append(stripped_uname.lower())
+
+                variations.append(uname.upper())
+                variations.append(uname.lower())
+
+        return variations