Mastering Active Directory Management with Python

Basic Active Directory Operations

1. Connecting to Active Directory

Establish a secure connection to your AD server:

from ldap3 import Server, Connection, ALL, NTLM, SUBTREE
from getpass import getpass

def connect_to_ad():
    server = Server('ldap://your_ad_server:389', get_info=ALL)
    username = input("Enter your username: ")
    password = getpass("Enter your password: ")
    conn = Connection(server, user=f'your_domain\\{username}', password=password, authentication=NTLM)
    
    if not conn.bind():
        print('Error in connection:', conn.result)
        return None
    print('Connection established successfully')
    return conn

ad_conn = connect_to_ad()

2. Creating a New User

Automate user creation with built-in safeguards:

from ldap3 import MODIFY_REPLACE
import re

def create_user(conn, username, firstname, lastname, password):
    # Validate inputs
    if not all([username, firstname, lastname, password]):
        print("All fields are required.")
        return False
    
    if not re.match(r'^[a-zA-Z0-9_-]+$', username):
        print("Invalid username format.")
        return False
    
    if len(password) < 8:
        print("Password must be at least 8 characters long.")
        return False

    user_dn = f'CN={firstname} {lastname},OU=Users,DC=your_domain,DC=com'
    attributes = {
        'objectClass': ['top', 'person', 'organizationalPerson', 'user'],
        'cn': f'{firstname} {lastname}',
        'sAMAccountName': username,
        'userPrincipalName': f'{username}@your_domain.com',
        'givenName': firstname,
        'sn': lastname,
        'displayName': f'{firstname} {lastname}',
        'userAccountControl': '512'  # Normal account
    }

    if conn.add(user_dn, attributes=attributes):
        print(f"User {username} created successfully.")
        # Set password
        conn.extend.microsoft.modify_password(user_dn, password)
        # Enable account
        conn.modify(user_dn, {'userAccountControl': [(MODIFY_REPLACE, ['512'])]})
        return True
    else:
        print(f"Error creating user: {conn.result}")
        return False

# Usage
create_user(ad_conn, 'jdoe', 'John', 'Doe', 'SecurePass123!')

3. Modifying User Attributes

Update user information safely:

def modify_user(conn, username, attributes):
    user_dn = f'CN={username},OU=Users,DC=your_domain,DC=com'
    
    print(f"Are you sure you want to modify the following attributes for {username}?")
    for attr, value in attributes.items():
        print(f"{attr}: {value}")
    
    confirm = input("Type 'YES' to confirm: ")
    if confirm != 'YES':
        print("Operation cancelled.")
        return False

    if conn.modify(user_dn, changes=attributes):
        print(f"User {username} modified successfully.")
        return True
    else:
        print(f"Error modifying user: {conn.result}")
        return False

# Usage
modify_user(ad_conn, 'jdoe', {'title': [(MODIFY_REPLACE, ['Senior Developer'])]})

Ethical Hacking and Red Teaming Techniques

Note: These techniques are for educational purposes only. Always obtain proper authorization before performing any security testing.

1. Password Spraying Detection

Detect potential password spraying attacks:

import datetime

def detect_password_spraying(conn, time_window_minutes=60, failed_attempt_threshold=5):
    now = datetime.datetime.now()
    time_window = now - datetime.timedelta(minutes=time_window_minutes)
    
    search_filter = f'(&(objectCategory=person)(objectClass=user)(badPasswordTime>={time_window.strftime("%Y%m%d%H%M%S.0Z")}))'
    attributes = ['sAMAccountName', 'badPasswordTime', 'badPwdCount']
    
    results = advanced_ad_query(conn, 'DC=your_domain,DC=com', search_filter, attributes)
    
    suspicious_accounts = [entry for entry in results if int(entry.badPwdCount.value) >= failed_attempt_threshold]
    
    if suspicious_accounts:
        print(f"Potential password spraying attack detected! The following accounts have had {failed_attempt_threshold} or more failed login attempts in the last {time_window_minutes} minutes:")
for account in suspicious_accounts:
print(f"Username: {account.sAMAccountName}, Failed Attempts: {account.badPwdCount}")
else:
print("No signs of password spraying detected.")
Usage
detect_password_spraying(ad_conn)

2. Privilege Escalation Detection

Monitor for unexpected privilege escalations:

def detect_privilege_escalation(conn):
    admin_group_dn = "CN=Domain Admins,CN=Users,DC=your_domain,DC=com"
    
    # Get current members of the Domain Admins group
    conn.search(admin_group_dn, '(objectClass=group)', attributes=['member'])
    current_admins = set(conn.entries[0].member.values)
    
    # Compare with a stored list of known admins (you'd need to implement this storage mechanism)
    known_admins = get_known_admins()  # Implement this function to retrieve known admins
    
    new_admins = current_admins - known_admins
    if new_admins:
        print("Potential privilege escalation detected! The following accounts have been added to the Domain Admins group:")
        for admin in new_admins:
            print(admin)
    else:
        print("No unexpected privilege escalations detected.")

# Usage
detect_privilege_escalation(ad_conn)

3. Kerberoasting Detection

Identify potential Kerberoasting attacks:

import datetime

def detect_kerberoasting(conn, time_window_hours=24):
    now = datetime.datetime.now()
    time_window = now - datetime.timedelta(hours=time_window_hours)
    
    search_filter = f'(&(objectCategory=person)(objectClass=user)(servicePrincipalName=*)(lastLogon>={time_window.strftime("%Y%m%d%H%M%S.0Z")}))'
    attributes = ['sAMAccountName', 'servicePrincipalName', 'lastLogon']
    
    results = advanced_ad_query(conn, 'DC=your_domain,DC=com', search_filter, attributes)
    
    if results:
        print(f"Potential Kerberoasting activity detected! The following accounts with SPNs have logged on in the last {time_window_hours} hours:")
        for account in results:
            print(f"Username: {account.sAMAccountName}, SPN: {account.servicePrincipalName}, Last Logon: {account.lastLogon}")
    else:
        print("No signs of Kerberoasting detected.")

# Usage
detect_kerberoasting(ad_conn)

Best Practices and Security Considerations

1. Use Least Privilege Principle: Always use accounts with the minimum necessary permissions for AD operations.
2. Implement Proper Error Handling: Catch and log exceptions to prevent exposing sensitive information in error messages.
3. Secure Credential Storage: Never hardcode credentials. Use environment variables or secure vaults for storing sensitive information.
4. Input Validation: Validate and sanitize all user inputs to prevent injection attacks.
5. Use SSL/TLS: Always use encrypted connections when communicating with Active Directory.
6. Implement Logging: Log all significant actions for auditing purposes.
7. Regular Code Reviews: Conduct regular code reviews to identify and fix potential security issues.
8. Keep Libraries Updated: Regularly update your Python libraries to ensure you have the latest security patches.