Query AD Sites and Services for configured subnets
(self)
| 93 | return [] |
| 94 | |
| 95 | def query_subnets(self): |
| 96 | """Query AD Sites and Services for configured subnets""" |
| 97 | if not self.connection: |
| 98 | logger.warning("No AD connection available - returning empty subnets") |
| 99 | return {} |
| 100 | |
| 101 | try: |
| 102 | # Check if this is an impacket connection (has query_ad_subnets method) |
| 103 | if hasattr(self.connection, 'impacket_auth'): |
| 104 | logger.debug("Using impacket connection for subnet query") |
| 105 | # Import the query function from main_final |
| 106 | import sys |
| 107 | import importlib |
| 108 | main_module = sys.modules.get('__main__') |
| 109 | if main_module and hasattr(main_module, 'query_ad_subnets'): |
| 110 | return main_module.query_ad_subnets(self.connection, self.domain) |
| 111 | else: |
| 112 | logger.warning("query_ad_subnets function not available") |
| 113 | return {} |
| 114 | |
| 115 | # Original ldap3 code for backward compatibility |
| 116 | # Convert domain to DN format - handle None domain |
| 117 | if not self.domain: |
| 118 | logger.warning("No domain specified - cannot query AD subnets") |
| 119 | return {} |
| 120 | |
| 121 | domain_dn = ','.join([f"DC={part}" for part in self.domain.split('.')]) |
| 122 | |
| 123 | # Search for subnet objects in AD Sites and Services |
| 124 | search_base = f"CN=Subnets,CN=Sites,CN=Configuration,{domain_dn}" |
| 125 | search_filter = '(objectClass=subnet)' |
| 126 | |
| 127 | self.connection.search(search_base, search_filter, attributes=LDAP_SUBNET_ATTRIBUTES) |
| 128 | |
| 129 | ad_subnets = {} |
| 130 | for entry in self.connection.entries: |
| 131 | if entry.cn: |
| 132 | subnet_name = str(entry.cn) |
| 133 | site_dn = str(entry.siteObject) if entry.siteObject else None |
| 134 | description = str(entry.description) if entry.description else None |
| 135 | |
| 136 | # Extract site name from DN |
| 137 | site_name = "Unknown" |
| 138 | if site_dn: |
| 139 | site_parts = site_dn.split(',') |
| 140 | for part in site_parts: |
| 141 | if part.startswith('CN=') and 'Sites' not in part: |
| 142 | site_name = part.replace('CN=', '') |
| 143 | break |
| 144 | |
| 145 | try: |
| 146 | network = ipaddress.IPv4Network(subnet_name) |
| 147 | ad_subnets[subnet_name] = { |
| 148 | 'network': network, |
| 149 | 'site': site_name, |
| 150 | 'description': description, |
| 151 | 'hosts': [] |
| 152 | } |
no test coverage detected