Query all computer objects from Active Directory
(connection, domain)
| 1395 | return None |
| 1396 | |
| 1397 | def query_computers(connection, domain): |
| 1398 | """Query all computer objects from Active Directory""" |
| 1399 | try: |
| 1400 | logger.info("Collecting computer objects from Active Directory...") |
| 1401 | |
| 1402 | # Check if this is an impacket connection |
| 1403 | if hasattr(connection, 'is_impacket') and connection.is_impacket: |
| 1404 | logger.info("🔐 Using impacket for AD computer query") |
| 1405 | |
| 1406 | # Convert domain to DN format (e.g., company.local -> DC=company,DC=local) |
| 1407 | domain_dn = ','.join([f"DC={part}" for part in domain.split('.')]) |
| 1408 | |
| 1409 | # Search for computer objects using real impacket LDAP |
| 1410 | search_base = domain_dn |
| 1411 | search_filter = '(objectClass=computer)' |
| 1412 | attributes = ['objectSid', 'cn', 'dNSHostName', 'operatingSystem'] |
| 1413 | |
| 1414 | logger.info(f"🔍 Searching for computers in: {search_base}") |
| 1415 | logger.info(f"🔍 Filter: {search_filter}") |
| 1416 | |
| 1417 | # Use the real search method |
| 1418 | results = connection.search(search_base, search_filter, attributes=attributes) |
| 1419 | |
| 1420 | computers = [] |
| 1421 | skipped_computers = 0 |
| 1422 | synthetic_sid_count = 0 |
| 1423 | |
| 1424 | for entry in connection.entries: # Use stored entries |
| 1425 | # Get computer attributes |
| 1426 | computer_name = getattr(entry, 'cn', None) |
| 1427 | |
| 1428 | # Skip if no computer name |
| 1429 | if not computer_name: |
| 1430 | logger.warning(f"⚠️ Skipping entry with no computer name (cn)") |
| 1431 | skipped_computers += 1 |
| 1432 | continue |
| 1433 | |
| 1434 | computer_sid_binary = getattr(entry, 'objectSid', None) |
| 1435 | |
| 1436 | # Get the real SID parsed from AD |
| 1437 | clean_sid = computer_sid_binary # This is now already parsed as string from LDAP |
| 1438 | |
| 1439 | # Check if SID is valid (starts with S- and looks like a real SID) |
| 1440 | if not clean_sid or not isinstance(clean_sid, str) or not clean_sid.startswith('S-1-5-21-'): |
| 1441 | # Generate synthetic SID for computers with failed/invalid SID parsing |
| 1442 | # Use a deterministic hash of computer name to ensure uniqueness |
| 1443 | import hashlib |
| 1444 | name_hash = int(hashlib.md5(computer_name.encode()).hexdigest()[:8], 16) |
| 1445 | synthetic_sid = f"S-1-5-21-SYNTHETIC-{name_hash}-{abs(hash(computer_name)) % 100000}" |
| 1446 | |
| 1447 | logger.warning(f"⚠️ Failed to parse SID for '{computer_name}', using synthetic SID") |
| 1448 | logger.debug(f" Original SID value: {clean_sid}") |
| 1449 | logger.debug(f" Synthetic SID: {synthetic_sid}") |
| 1450 | |
| 1451 | clean_sid = synthetic_sid |
| 1452 | synthetic_sid_count += 1 |
| 1453 | else: |
| 1454 | logger.debug(f"✅ Using real SID from AD for {computer_name}: {clean_sid}") |