Bump version to 1.2.5 and add OU filtering functionality in BloodHound CLI

This commit updates the version to 1.2.5 in the project metadata and introduces a new feature that allows users to filter by Organizational Unit (OU) when querying users. The implementation includes updates to the command parser and core client methods, along with corresponding unit tests to ensure expected behavior.

Author: ADScanPro

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "bloodhound-cli"  # The package name as installed via pip (pip install bloodhound-cli)
-version = "1.2.4"          # Bump: add CE skeleton, --edition, --verbose, rich optional
+version = "1.2.5"          # Bump: add CE skeleton, --edition, --verbose, rich optional
 description = "CLI for querying BloodHound data (Legacy Neo4j + CE skeleton)."
 readme = "README.md"
 authors = [

src/bloodhound_cli/__init__.py

@@ -1,6 +1,6 @@
 """BloodHound CLI package entrypoint."""
 
-__version__ = "1.2.4"
+__version__ = "1.2.5"
 __all__ = ["__version__"]
 
 # End of package metadata

src/bloodhound_cli/core/ce.py

@@ -198,7 +198,45 @@ def get_users(self, domain: str) -> List[str]:
                         users.append(samaccountname)
 
             return users
+        except Exception:
+            return []
+
+    def get_users_in_ou(self, domain: str, ou_distinguished_name: str) -> List[str]:
+        """Get enabled users that belong to a specific OU using its distinguished name.
+
+        Args:
+            domain: AD domain name to filter users by (e.g. "north.sevenkingdoms.local").
+            ou_distinguished_name: Distinguished Name (DN) of the OU to search under.
+
+        Returns:
+            List of `samaccountname` values for users that belong to the OU.
+        """
+        try:
+            # Escape single quotes to avoid breaking the Cypher string
+            sanitized_ou_dn = ou_distinguished_name.replace("'", "\\'")
+
+            cypher_query = f"""
+            MATCH (ou:OU)
+            WHERE toLower(ou.distinguishedname) = toLower('{sanitized_ou_dn}')
+            MATCH (u:User)
+            WHERE u.enabled = true
+              AND toUpper(u.domain) = '{domain.upper()}'
+              AND toLower(u.distinguishedname) CONTAINS toLower(ou.distinguishedname)
+            RETURN u
+            """
+
+            result = self.execute_query(cypher_query)
+            users: List[str] = []
 
+            if result and isinstance(result, list):
+                for node_properties in result:
+                    samaccountname = node_properties.get("samaccountname") or node_properties.get("name", "")
+                    if samaccountname:
+                        if "@" in samaccountname:
+                            samaccountname = samaccountname.split("@")[0]
+                        users.append(samaccountname)
+
+            return users
         except Exception:
             return []
 

src/bloodhound_cli/main.py

@@ -156,7 +156,15 @@ def cmd_users(args):
             output_results(results, args.output, args.verbose, "password info")
             return
 
-        if args.high_value:
+        if getattr(args, "ou_dn", None):
+            if args.edition.lower() != "ce":
+                print(
+                    "Filtering users by OU is only available for BloodHound CE (--edition ce)."
+                )
+                return
+            users = client.get_users_in_ou(args.domain, args.ou_dn)
+            user_type = f"users in OU {args.ou_dn}"
+        elif args.high_value:
             users = client.get_highvalue_users(args.domain)
             user_type = "high value users"
         elif args.admin_count:
@@ -704,6 +712,11 @@ def main():
     users_parser.add_argument(
         "-u", "--user", help="Specific user to query (for password-last-change)"
     )
+    users_parser.add_argument(
+        "--ou-dn",
+        dest="ou_dn",
+        help="Distinguished Name of an OU to list its users (CE only)",
+    )
     users_parser.add_argument(
         "--high-value", action="store_true", help="Show only high value users"
     )

tests/unit/test_ce_queries.py

@@ -309,6 +309,36 @@ def test_get_users_exception_handling(self, mock_ce_client):
         users = mock_ce_client.get_users("error.domain.local")
 
         assert len(users) == 0
+
+    def test_get_users_in_ou_exception_handling(self, mock_ce_client):
+        """Test exception handling in get_users_in_ou"""
+        mock_ce_client.execute_query.side_effect = Exception("Query error")
+
+        users = mock_ce_client.get_users_in_ou(
+            "error.domain.local",
+            "OU=Winterfell,DC=north,DC=sevenkingdoms,DC=local",
+        )
+
+        assert len(users) == 0
+
+    def test_get_users_in_ou_uses_ou_dn_and_domain(self, mock_ce_client):
+        """Test OU-based user enumeration builds the correct Cypher query and parses results."""
+        ou_dn = "OU=Winterfell,DC=north,DC=sevenkingdoms,DC=local"
+        domain = "north.sevenkingdoms.local"
+
+        users = mock_ce_client.get_users_in_ou(domain, ou_dn)
+
+        # Same mocked execute_query data as for get_users
+        assert len(users) == 13
+        assert "jeor.mormont" in users
+
+        mock_ce_client.execute_query.assert_called_once()
+        query = mock_ce_client.execute_query.call_args[0][0]
+        assert "MATCH (ou:OU)" in query
+        assert "MATCH (u:User)" in query
+        assert "ou.distinguishedname" in query
+        assert domain.upper() in query
+        assert ou_dn in query
 
     def test_get_computers_basic(self, mock_ce_client_computers):
         """Test basic computer enumeration with CySQL query"""
@@ -736,6 +766,69 @@ def close(self):
         assert dummy_client.calls == [("essos.local", "daenerys.targaryen", True)]
         assert dummy_client.closed is True
 
+    def test_user_command_ou_filter_outputs_expected_users(self, monkeypatch, capsys):
+        """Simulate `bloodhound-cli user --ou-dn ... -d sevenkingdoms.local` output."""
+        expected_users = [
+            "maester.pycelle",
+            "lord.varys",
+            "petyer.baelish",
+            "stannis.baratheon",
+            "joffrey.baratheon",
+            "renly.baratheon",
+            "robert.baratheon",
+            "cersei.lannister",
+            "jaime.lannister",
+            "tywin.lannister",
+        ]
+
+        class DummyClient:
+            def __init__(self):
+                self.closed = False
+                self.calls = []
+
+            def get_users_in_ou(self, domain, ou_distinguished_name):
+                self.calls.append((domain, ou_distinguished_name))
+                return expected_users
+
+            def close(self):
+                self.closed = True
+
+        dummy_client = DummyClient()
+        monkeypatch.setattr(
+            cli_main,
+            "get_client",
+            lambda *_, **__: dummy_client,
+        )
+
+        args = SimpleNamespace(
+            edition="ce",
+            uri=None,
+            user=None,
+            password=None,
+            base_url="http://localhost:8080",
+            username="admin",
+            ce_password="Bloodhound123!",
+            debug=False,
+            verbose=False,
+            domain="sevenkingdoms.local",
+            ou_dn="OU=crownlands,DC=sevenkingdoms,DC=local",
+            high_value=False,
+            admin_count=False,
+            password_never_expires=False,
+            password_not_required=False,
+            password_last_change=False,
+            output=None,
+        )
+
+        cli_main.cmd_users(args)
+
+        captured = capsys.readouterr()
+        assert captured.out.strip().splitlines() == expected_users
+        assert dummy_client.calls == [
+            ("sevenkingdoms.local", "OU=crownlands,DC=sevenkingdoms,DC=local")
+        ]
+        assert dummy_client.closed is True
+
     def test_version_command_prints_version(self, capsys):
         """Ensure `bloodhound-cli version` prints the version string."""
         cli_main.cmd_version(SimpleNamespace())