scripts/domain_intel.py
scripts/domain_intel.pyBrowse 2 files
3,830 tokens
15,710 bytes
Token encoding: o200k_base
Snapshot 24fd22b
← Back to SKILL.md
1#!/usr/bin/env python32"""3Domain Intelligence — Passive OSINT via Python stdlib.4 5Usage:6 python domain_intel.py subdomains example.com7 python domain_intel.py ssl example.com8 python domain_intel.py whois example.com9 python domain_intel.py dns example.com10 python domain_intel.py available example.com11 python domain_intel.py bulk example.com github.com google.com --checks ssl,dns12 13All output is structured JSON. No dependencies beyond Python stdlib.14Works on Linux, macOS, and Windows.15"""16 17import json18import re19import socket20import ssl21import sys22import urllib.request23import urllib.parse24from concurrent.futures import ThreadPoolExecutor, as_completed25from datetime import datetime, timezone26 27 28# ─── Subdomain Discovery (crt.sh) ──────────────────────────────────────────29 30def subdomains(domain, include_expired=False, limit=200):31 """Find subdomains via Certificate Transparency logs."""32 url = f"https://crt.sh/?q=%25.{urllib.parse.quote(domain)}&output=json"33 req = urllib.request.Request(url, headers={34 "User-Agent": "domain-intel-skill/1.0", "Accept": "application/json",35 })36 with urllib.request.urlopen(req, timeout=15) as r:37 entries = json.loads(r.read().decode())38 39 seen, results = set(), []40 now = datetime.now(timezone.utc)41 for e in entries:42 not_after = e.get("not_after", "")43 if not include_expired and not_after:44 try:45 dt = datetime.strptime(not_after[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)46 if dt <= now:47 continue48 except ValueError:49 pass50 for name in e.get("name_value", "").splitlines():51 name = name.strip().lower()52 if name and name not in seen:53 seen.add(name)54 results.append({55 "subdomain": name,56 "issuer": e.get("issuer_name", ""),57 "not_after": not_after,58 })59 60 results.sort(key=lambda r: (r["subdomain"].startswith("*"), r["subdomain"]))61 return {"domain": domain, "count": min(len(results), limit), "subdomains": results[:limit]}62 63 64# ─── SSL Certificate Inspection ────────────────────────────────────────────65 66def check_ssl(host, port=443, timeout=10):67 """Inspect the TLS certificate of a host."""68 def flat(rdns):69 r = {}70 for rdn in rdns:71 for item in rdn:72 if isinstance(item, (list, tuple)) and len(item) == 2:73 r[item[0]] = item[1]74 return r75 76 def parse_date(s):77 for fmt in ("%b %d %H:%M:%S %Y %Z", "%b %d %H:%M:%S %Y %Z"):78 try:79 return datetime.strptime(s, fmt).replace(tzinfo=timezone.utc)80 except ValueError:81 pass82 return None83 84 warning = None85 try:86 ctx = ssl.create_default_context()87 with socket.create_connection((host, port), timeout=timeout) as sock:88 with ctx.wrap_socket(sock, server_hostname=host) as s:89 cert, cipher, proto = s.getpeercert(), s.cipher(), s.version()90 except ssl.SSLCertVerificationError as e:91 warning = str(e)92 ctx = ssl.create_default_context()93 ctx.check_hostname = False94 ctx.verify_mode = ssl.CERT_NONE95 with socket.create_connection((host, port), timeout=timeout) as sock:96 with ctx.wrap_socket(sock, server_hostname=host) as s:97 cert, cipher, proto = s.getpeercert(), s.cipher(), s.version()98 99 not_after = parse_date(cert.get("notAfter", ""))100 now = datetime.now(timezone.utc)101 days = (not_after - now).days if not_after else None102 is_expired = days is not None and days < 0103 104 if is_expired:105 status = f"EXPIRED ({abs(days)} days ago)"106 elif days is not None and days <= 14:107 status = f"CRITICAL — {days} day(s) left"108 elif days is not None and days <= 30:109 status = f"WARNING — {days} day(s) left"110 else:111 status = f"OK — {days} day(s) remaining" if days is not None else "unknown"112 113 return {114 "host": host, "port": port,115 "subject": flat(cert.get("subject", [])),116 "issuer": flat(cert.get("issuer", [])),117 "subject_alt_names": [f"{t}:{v}" for t, v in cert.get("subjectAltName", [])],118 "not_before": parse_date(cert.get("notBefore", "")).isoformat() if parse_date(cert.get("notBefore", "")) else "",119 "not_after": not_after.isoformat() if not_after else "",120 "days_remaining": days, "is_expired": is_expired, "expiry_status": status,121 "tls_version": proto,122 "cipher_suite": cipher[0] if cipher else None,123 "serial_number": cert.get("serialNumber", ""),124 "verification_warning": warning,125 }126 127 128# ─── WHOIS Lookup ──────────────────────────────────────────────────────────129 130WHOIS_SERVERS = {131 "com": "whois.verisign-grs.com", "net": "whois.verisign-grs.com",132 "org": "whois.pir.org", "io": "whois.nic.io", "co": "whois.nic.co",133 "ai": "whois.nic.ai", "dev": "whois.nic.google", "app": "whois.nic.google",134 "tech": "whois.nic.tech", "shop": "whois.nic.shop", "store": "whois.nic.store",135 "online": "whois.nic.online", "site": "whois.nic.site", "cloud": "whois.nic.cloud",136 "digital": "whois.nic.digital", "media": "whois.nic.media", "blog": "whois.nic.blog",137 "info": "whois.afilias.net", "biz": "whois.biz", "me": "whois.nic.me",138 "tv": "whois.nic.tv", "cc": "whois.nic.cc", "ws": "whois.website.ws",139 "uk": "whois.nic.uk", "co.uk": "whois.nic.uk", "de": "whois.denic.de",140 "nl": "whois.domain-registry.nl", "fr": "whois.nic.fr", "it": "whois.nic.it",141 "es": "whois.nic.es", "pl": "whois.dns.pl", "ru": "whois.tcinet.ru",142 "se": "whois.iis.se", "no": "whois.norid.no", "fi": "whois.fi",143 "ch": "whois.nic.ch", "at": "whois.nic.at", "be": "whois.dns.be",144 "cz": "whois.nic.cz", "br": "whois.registro.br", "ca": "whois.cira.ca",145 "mx": "whois.mx", "au": "whois.auda.org.au", "jp": "whois.jprs.jp",146 "cn": "whois.cnnic.cn", "in": "whois.inregistry.net", "kr": "whois.kr",147 "sg": "whois.sgnic.sg", "hk": "whois.hkirc.hk", "tr": "whois.nic.tr",148 "ae": "whois.aeda.net.ae", "za": "whois.registry.net.za",149 "space": "whois.nic.space", "zone": "whois.nic.zone", "ninja": "whois.nic.ninja",150 "guru": "whois.nic.guru", "rocks": "whois.nic.rocks", "live": "whois.nic.live",151 "game": "whois.nic.game", "games": "whois.nic.games",152}153 154 155def whois_lookup(domain):156 """Query WHOIS servers for domain registration info."""157 parts = domain.split(".")158 server = WHOIS_SERVERS.get(".".join(parts[-2:])) or WHOIS_SERVERS.get(parts[-1])159 if not server:160 return {"error": f"No WHOIS server for .{parts[-1]}"}161 162 try:163 with socket.create_connection((server, 43), timeout=10) as s:164 s.sendall((domain + "\r\n").encode())165 chunks = []166 while True:167 c = s.recv(4096)168 if not c:169 break170 chunks.append(c)171 raw = b"".join(chunks).decode("utf-8", errors="replace")172 except Exception as e:173 return {"error": str(e)}174 175 patterns = {176 "registrar": r"(?:Registrar|registrar):\s*(.+)",177 "creation_date": r"(?:Creation Date|Created|created):\s*(.+)",178 "expiration_date": r"(?:Registry Expiry Date|Expiration Date|Expiry Date):\s*(.+)",179 "updated_date": r"(?:Updated Date|Last Modified):\s*(.+)",180 "name_servers": r"(?:Name Server|nserver):\s*(.+)",181 "status": r"(?:Domain Status|status):\s*(.+)",182 "dnssec": r"DNSSEC:\s*(.+)",183 }184 result = {"domain": domain, "whois_server": server}185 for key, pat in patterns.items():186 matches = re.findall(pat, raw, re.IGNORECASE)187 if matches:188 if key in {"name_servers", "status"}:189 result[key] = list(dict.fromkeys(m.strip().lower() for m in matches))190 else:191 result[key] = matches[0].strip()192 193 for field in ("creation_date", "expiration_date", "updated_date"):194 if field in result:195 for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):196 try:197 dt = datetime.strptime(result[field][:19], fmt).replace(tzinfo=timezone.utc)198 result[field] = dt.isoformat()199 if field == "expiration_date":200 days = (dt - datetime.now(timezone.utc)).days201 result["expiration_days_remaining"] = days202 result["is_expired"] = days < 0203 break204 except ValueError:205 pass206 return result207 208 209# ─── DNS Records ───────────────────────────────────────────────────────────210 211def dns_records(domain, types=None):212 """Resolve DNS records using system DNS + Google DoH."""213 if not types:214 types = ["A", "AAAA", "MX", "NS", "TXT", "CNAME"]215 records = {}216 217 for qtype in types:218 if qtype == "A":219 try:220 records["A"] = list(dict.fromkeys(221 i[4][0] for i in socket.getaddrinfo(domain, None, socket.AF_INET)222 ))223 except Exception:224 records["A"] = []225 elif qtype == "AAAA":226 try:227 records["AAAA"] = list(dict.fromkeys(228 i[4][0] for i in socket.getaddrinfo(domain, None, socket.AF_INET6)229 ))230 except Exception:231 records["AAAA"] = []232 else:233 url = f"https://dns.google/resolve?name={urllib.parse.quote(domain)}&type={qtype}"234 try:235 req = urllib.request.Request(url, headers={"User-Agent": "domain-intel-skill/1.0"})236 with urllib.request.urlopen(req, timeout=10) as r:237 data = json.loads(r.read())238 records[qtype] = [239 a.get("data", "").strip().rstrip(".")240 for a in data.get("Answer", []) if a.get("data")241 ]242 except Exception:243 records[qtype] = []244 245 return {"domain": domain, "records": records}246 247 248# ─── Domain Availability Check ─────────────────────────────────────────────249 250def check_available(domain):251 """Check domain availability using passive signals (DNS + WHOIS + SSL)."""252 signals = {}253 254 # DNS255 try:256 a = [i[4][0] for i in socket.getaddrinfo(domain, None, socket.AF_INET)]257 except Exception:258 a = []259 260 try:261 ns_url = f"https://dns.google/resolve?name={urllib.parse.quote(domain)}&type=NS"262 req = urllib.request.Request(ns_url, headers={"User-Agent": "domain-intel-skill/1.0"})263 with urllib.request.urlopen(req, timeout=10) as r:264 ns = [x.get("data", "") for x in json.loads(r.read()).get("Answer", [])]265 except Exception:266 ns = []267 268 signals["dns_a"] = a269 signals["dns_ns"] = ns270 dns_exists = bool(a or ns)271 272 # SSL273 ssl_up = False274 try:275 ctx = ssl.create_default_context()276 ctx.check_hostname = False277 ctx.verify_mode = ssl.CERT_NONE278 with socket.create_connection((domain, 443), timeout=3) as s:279 with ctx.wrap_socket(s, server_hostname=domain):280 ssl_up = True281 except Exception:282 pass283 signals["ssl_reachable"] = ssl_up284 285 # WHOIS (quick check)286 tld = domain.rsplit(".", 1)[-1]287 server = WHOIS_SERVERS.get(tld)288 whois_avail = None289 whois_note = ""290 if server:291 try:292 with socket.create_connection((server, 43), timeout=10) as s:293 s.sendall((domain + "\r\n").encode())294 raw = b""295 while True:296 c = s.recv(4096)297 if not c:298 break299 raw += c300 raw = raw.decode("utf-8", errors="replace").lower()301 if any(p in raw for p in ["no match", "not found", "no data found", "status: free"]):302 whois_avail = True303 whois_note = "WHOIS: not found"304 elif "registrar:" in raw or "creation date:" in raw:305 whois_avail = False306 whois_note = "WHOIS: registered"307 else:308 whois_note = "WHOIS: inconclusive"309 except Exception as e:310 whois_note = f"WHOIS error: {e}"311 312 signals["whois_available"] = whois_avail313 signals["whois_note"] = whois_note314 315 if not dns_exists and whois_avail is True:316 verdict, conf = "LIKELY AVAILABLE", "high"317 elif dns_exists or whois_avail is False or ssl_up:318 verdict, conf = "REGISTERED / IN USE", "high"319 elif not dns_exists and whois_avail is None:320 verdict, conf = "POSSIBLY AVAILABLE", "medium"321 else:322 verdict, conf = "UNCERTAIN", "low"323 324 return {"domain": domain, "verdict": verdict, "confidence": conf, "signals": signals}325 326 327# ─── Bulk Analysis ─────────────────────────────────────────────────────────328 329COMMAND_MAP = {330 "subdomains": subdomains,331 "ssl": check_ssl,332 "whois": whois_lookup,333 "dns": dns_records,334 "available": check_available,335}336 337 338def bulk_check(domains, checks=None, max_workers=5):339 """Run multiple checks across multiple domains in parallel."""340 if not checks:341 checks = ["ssl", "whois", "dns"]342 343 def run_one(d):344 entry = {"domain": d}345 for check in checks:346 fn = COMMAND_MAP.get(check)347 if fn:348 try:349 entry[check] = fn(d)350 except Exception as e:351 entry[check] = {"error": str(e)}352 return entry353 354 results = []355 with ThreadPoolExecutor(max_workers=min(max_workers, 10)) as ex:356 futures = {ex.submit(run_one, d): d for d in domains[:20]}357 for f in as_completed(futures):358 results.append(f.result())359 360 return {"total": len(results), "checks": checks, "results": results}361 362 363# ─── CLI Entry Point ───────────────────────────────────────────────────────364 365def main():366 if len(sys.argv) < 3:367 print(__doc__)368 sys.exit(1)369 370 command = sys.argv[1].lower()371 args = sys.argv[2:]372 373 if command == "bulk":374 # Parse --checks flag375 checks = None376 domains = []377 i = 0378 while i < len(args):379 if args[i] == "--checks" and i + 1 < len(args):380 checks = [c.strip() for c in args[i + 1].split(",")]381 i += 2382 else:383 domains.append(args[i])384 i += 1385 result = bulk_check(domains, checks)386 elif command in COMMAND_MAP:387 result = COMMAND_MAP[command](args[0])388 else:389 print(f"Unknown command: {command}")390 print(f"Available: {', '.join(COMMAND_MAP.keys())}, bulk")391 sys.exit(1)392 393 print(json.dumps(result, indent=2))394 395 396if __name__ == "__main__":397 main()398