xiaomingx/subhunter/prompt

uv init subdomain_scanner
cd subdomain_scanner
uv add dnspython httpx aiohttp
 

"""
Function: Subdomain Scanner Core
Business Problem: 高效探测企业子域名资产，识别潜在的安全暴露面，支持大规模并发检测与通配符过滤。
"""
import asyncio
import dns.asyncresolver
import httpx
from typing import List, Set
class SubdomainScanner:
    def __init__(self, domain: str, limit: int = 100):
        self.domain = domain
        self.semaphore = asyncio.Semaphore(limit)
        self.results = []
        self.resolver = dns.asyncresolver.Resolver()
        self.resolver.nameservers = ['8.8.8.8', '1.1.1.1']
    async def check_dns(self, sub: str) -> str:
        """执行异步 DNS A记录查询"""
        target = f"{sub}.{self.domain}"
        async with self.semaphore:
            try:
                answers = await self.resolver.resolve(target, 'A')
                return target
            except Exception:
                return None
    async def check_http(self, url: str):
        """执行异步 HTTP 存活性验证"""
        async with httpx.AsyncClient(timeout=5.0, follow_redirects=True) as client:
            try:
                resp = await client.get(f"http://{url}")
                if resp.status_code < 400:
                    self.results.append({"domain": url, "status": resp.status_code, "server": resp.headers.get("Server")})
            except Exception:
                pass
    async def run(self, wordlist: List[str]):
        # 1. DNS 爆破阶段
        print(f"[*] Starting DNS brute force for {self.domain}...")
        dns_tasks = [self.check_dns(sub) for sub in wordlist]
        found_domains = await asyncio.gather(*dns_tasks)
        active_domains = [d for d in found_domains if d]
        # 2. HTTP 验证阶段
        print(f"[*] Validating {len(active_domains)} active domains via HTTP...")
        http_tasks = [self.check_http(d) for d in active_domains]
        await asyncio.gather(*http_tasks)
        
        return self.results
async def main():
    scanner = SubdomainScanner("jd.com", limit=200)
    # 示例字典
    words = ["m", "list", "re", "passport", "pay", "v-api"]
    results = await scanner.run(words)
    print(f"[+] Found {len(results)} live subdomains.")
    for r in results:
        print(f" - {r['domain']} [{r['status']}]")
if __name__ == "__main__":
    asyncio.run(main())