🔭 Bottom-Line Up Front

ADIDNS_Parser is an illustratation of the extensibility of graph databases within common tooling used by security assessment teams.

Specifically, this generates files that can reconcile ADIDNS dumped data to existing Computer objects in Bloodhound databases.

🤔 Why?

Fundamentally, professional security assessments are inelastic in duration.

Compoundingly, they are data synthesis problems more often than purely technical. The method of "winning" is enrichment of data, as quickly as possible, and acting upon it.

Arbitrary properties add additional contraints to filtering. The more sieves one has, the better their lens for finding what's desired.

🚀 Usage

.\adidns_parser.exe --help
Usage: adidns_parser.exe --dns-data <DNS_DATA> --known-computers <KNOWN_COMPUTERS>

Options:
  -a, --dns-data <DNS_DATA>
          The formatted Active Directory Integrated DNS data file, with format: HOSTNAME IP_ADDRESS per newline
  -c, --known-computers <KNOWN_COMPUTERS>
          The known computers file, with format: HOSTNAME per newline
  -h, --help
          Print help
  -V, --version
          Print version

🌱 Outputs

A single CSV file is generated upon execution: resolved_records.csv

👷 Building

Assuming you have rust installed: cargo build --release

🐇 Anticipated Throughput

 GCI *.txt | % { Get-Content $_ | Measure-Object | Select Count }; .\adidns_parser.exe --dns-data .\adidns_records.txt --known-computers .\full_systems.txt

 Count
 -----
216772
135110

Found 83120 matching records in 96.727ms

Quick Importation of Data

To be very brrr, one should do a few things: - Enable APOC for your Neo4j database - Ensure a the browser interface is enabled

If you'd prefer to do so via Docker, the following will accomplish both:

---
services:
  graph-db:
    image: docker.io/library/neo4j:4.4
    environment:
      NEO4J_AUTH: "neo4j/CHANGE_ME"
      NEO4JLABS_PLUGINS: '["apoc"]'
      NEO4J_dbms_allow__upgrade: true
      NEO4J_dbms_security_procedures_unrestricted: "apoc.*"
      NEO4J_dbms_security_procedures_allowlist: "apoc.*"
      NEO4J_apoc_import_file_enabled: "true"
      NEO4J_apoc_import_file_use__neo4j__config: "true"
    ports:
      - 7474:7474
      - 7687:7687
    volumes:
      - ./neo4j:/data

Execute via: docker-compose up -d

Upload your resolved_records.csv file to the /var/lib/neo4j/import directory: docker cp resolved_records.csv CONTAINER_NAME /var/lib/neo4j/import/

Within the browser interface (default of http://localhost:7474/browser):

CALL apoc.periodic.iterate(
    "LOAD CSV WITH HEADERS FROM 'file:///resolved_records.csv' AS Row Return Row",
    "MATCH (c:Computer {name: Row.system}) SET c.ipaddress = Row.ipaddress",
    {batchSize: 1000, parallel: true}
);

Example Filtering

Using Pythonic bindings to our database, we can now add additional filters to include/omit entire address ranges.

No more hand-jamming required!

#!/usr/bin/env python3
from neo4j import GraphDatabase
from ipaddress import IPv4Network

class GraphDB(object):
    def __init__(self, ip: str = "", port: int = 0, username:str = "", password: str = "") -> None:
        self.driver = GraphDatabase.driver(f'bolt://{ip}:{port}', auth=(username, password))
        self.session = self.driver.session()
        self._computers = None

    def close(self):
        self.driver.close()

    def getSystemsWithCriteria(self, splat: list = None):
        '''
        In this example, we assume the ipaddress is already set.
        We will abuse how Cypher interpolates arrays (they are the same in Python's str representation)
        '''
        query = f'''
MATCH (c:Computer)
WHERE NOT c.ipaddress IS NULL
AND c.ipaddress in {str(splat)}
RETURN c as Computer
        '''
        return list(self.session.run(query=query))


def main():
    neo4jdb = GraphDB(ip='localhost', port=0, username='neo4j', password='REPLACE_ME')
    candidateRange = [str(host) for host in IPv4Network(address='192.168.1.0/24')]

    if (data := neo4jdb.getSystemsWithCriteria(candidateRange)):
        for system in data:
            print(system['Computer']['name'], system['Computer']['ipaddress'])


if __name__ == '__main__':
    main()