check_weak_acls_bloodhound.py

check_weak_acls_bloodhound.py

from neo4j import GraphDatabase
from tabulate import tabulate
import argparse
from itertools import groupby  # Make sure to import groupby from itertools
import sys

# Initialize the driver
uri = "bolt://localhost:7687"  # Change to your Neo4j server URI
username = "neo4j"  # Your Neo4j username
password = "bloodhoundcommunityedition"  # Your Neo4j password

driver = GraphDatabase.driver(uri, auth=(username, password))

check_list = []
check_list.append(["GenericAll/GenericWrite/WriteProperty on Computer/Use/Groups", "MATCH (p)-[r]->(t:Group) WHERE type(r) IN ['GenericAll', 'GenericWrite', 'WriteProperty'] AND (p:User OR p:Group OR p:Computer) RETURN p.name AS Principal, labels(p) AS PrincipalType, type(r) AS Permission, t.name AS TargetGroup, labels(t) AS TargetType"])
check_list.append(["Lists all accounts or groups with WriteOwner access to groups", "MATCH (p)-[r:WriteOwner]->(g:Group) RETURN p.name AS PrincipalName, labels(p) AS PrincipalType, g.name AS TargetGroup, labels(g) AS TargetType ORDER BY PrincipalName"])
check_list.append(["Find Users or Groups with WriteDACL access to Domain Groups", "MATCH (p)-[r:WriteDACL]->(g:Group) WHERE (p:User OR p:Group) RETURN p.name AS Principal, labels(p) AS PrincipalType, r.name AS Permission, g.name AS TargetGroup, labels(g) AS TargetType ORDER BY Principal"])
check_list.append(["Find Users or Groups with AllExtendedRights access to Domain Groups.", "MATCH (p)-[r:AllExtendedRights]->(t) WHERE (t:User OR t:Group) RETURN p.name AS Principal, labels(p) AS PrincipalType, type(r) AS Permission, t.name AS Target, labels(t) AS TargetType"])
check_list.append(["Finds Users or Groups with the ForceChangePassword permission to change the password of User objects", "MATCH (p)-[r:ForceChangePassword]->(t:User) RETURN p.name AS Principal, labels(p) AS PrincipalType, type(r) AS Permission, t.name AS TargetUser, labels(t) AS TargetType"])
check_list.append(["Finds Users or Groups with the Self-Membership permission on Groups", "MATCH (p)-[r:SelfMembership]->(g:Group) RETURN p.name AS Principal, labels(p) AS PrincipalType, type(r) AS Permission, g.name AS TargetGroup, labels(g) AS TargetType"])

def query_neo4j(username_filters=None, show_keys=False):
    """
    Queries the Neo4j database, optionally filtering rows based on username filters.

    :param username_filters: List of usernames to filter by. If None, no filtering is applied.
    :param show_keys: If True, display only source users/groups.
    """
    with driver.session() as session:
        for description, query in check_list:
            print(f"\n- {description}")
            result = session.run(query)
            rows = []

            for record in result:
                data = record.data()
                if username_filters:
                    first_field = next(iter(data.values()), None)
                    if (
                        isinstance(first_field, str)
                        and not any(filter.lower() in first_field.lower() for filter in username_filters)
                    ):
                        continue
                rows.append(data)

            if not rows:
                print("None")
                continue

            rows.sort(key=lambda x: next(iter(x.values())))
            grouped_rows = groupby(rows, key=lambda x: next(iter(x.values())))

            for key, group in grouped_rows:
                print(f"{key}")
                if not show_keys:
                    grouped_list = list(group)
                    print(tabulate(grouped_list, headers="keys", tablefmt="simple_outline"))


def query_groups(username):
    """
    Retrieve the groups a user belongs to.

    :param username: The username to query.
    :return: A list of group names.
    """
    cypher_query = """
    MATCH (u:User {name: $username})-[:MemberOf]->(g:Group)
    RETURN g.name AS GroupName
    ORDER BY g.name
    """
    with driver.session() as session:
        result = session.run(cypher_query, {"username": username})
        return [record["GroupName"] for record in result]


def main():
    parser = argparse.ArgumentParser(description="Execute Neo4j queries on a BloodHound database to identify weak permissions in Active Directory's Discretionary Access Control Lists (DACLs) and Access Control Entries (ACEs). The script allows filtering by username or group membership, helping to identify misconfigured permissions that could be exploited for privilege escalation or lateral movement.")
    parser.add_argument("-u", "--username", type=str, help="Filter results by username (e.g., [email protected]).")
    parser.add_argument("-k", "--showKeys", action="store_true", help="Show only source users/groups.")
    parser.add_argument("-o", "--output", type=str, help="File to write the output.")
    args = parser.parse_args()

    output_file = None
    if args.output:
        output_file = open(args.output, "w")
        sys.stdout = output_file

    try:
        if args.username:
            group_names = query_groups(args.username)
            if group_names:
                print("- Members of the below groups:")
                print("\n".join(group_names))
            group_names.append(args.username)
            query_neo4j(username_filters=group_names, show_keys=args.showKeys)
        else:
            query_neo4j(username_filters=None, show_keys=args.showKeys)
    finally:
        if output_file:
            sys.stdout = sys.__stdout__
            output_file.close()


if __name__ == "__main__":
    main()