External Attack Surface Audit: How FOFA, Shodan, and Censys Help You Find Forgotten Services and Leaks

				
					import os, base64, requests
from typing import List, Tuple
import shodan 
from censys.search import CensysHosts
# --- Config ---
ORG = "example.com"  # your primary domain for Censys pivots
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")
# --- Helpers ---
def post_slack(title: str, lines: List[str]) -> None:
    if not SLACK_WEBHOOK or not lines:
        return
    text = f"*{title}*\n" + "\n".join(lines[:30])
    requests.post(SLACK_WEBHOOK, json={"text": text}, timeout=15)
# --- FOFA: Swagger UI with auth hints ---
def fofa_swagger() -> List[Tuple[str, str, str]]:
    email = os.getenv("FOFA_EMAIL"); key = os.getenv("FOFA_KEY")
    if not (email and key): return []
    query = 'title="Swagger UI" && (body="Bearer" || body="jwt")'
    qbase64 = base64.b64encode(query.encode()).decode()
    url = f"https://fofa.info/api/v1/search/all"
    params = {
        "email": email,
        "key": key,
        "qbase64": qbase64,
        "fields": "host,port,title"
    }
    r = requests.get(url, params=params, timeout=30)
    r.raise_for_status()
    rows = r.json().get("results", [])
    # Return tuples (host, port, title)
    return [(str(h), str(p), str(t)) for h, p, t in rows]
# --- Shodan: Index listings leaking .env or id_rsa (fixed query) ---
def shodan_index_leaks() -> List[Tuple[str, int, str]]:
    api_key = os.getenv("SHODAN_API_KEY")
    if not api_key: return []
    api = shodan.Shodan(api_key)
    query = 'http.title:"Index of /" http.html:".env" http.html:"config.php"'
    data = api.search(query, page=1)  # mind your credits
    out = []
    for m in data.get("matches", []):
        ip = m.get("ip_str"); port = m.get("port"); title = (m.get("http", {}) or {}).get("title")
        if ip and port:
            out.append((ip, int(port), title or ""))
    return out
# --- Censys: hosts whose leaf cert references your domain ---
def censys_cert_hosts(domain: str) -> List[str]:
    c = CensysHosts()  # reads CENSYS_API_ID / CENSYS_API_SECRET from env
    query = f'services.tls.certificates.leaf_data.names: "{domain}"'
    results = []
    for hit in c.search(query, per_page=50):
        ip = hit.get("ip")
        if ip:
            results.append(ip)
        if len(results) >= 200:  # keep it light
            break
    return results
# --- Main run ---
if __name__ == "__main__":
    try:
        fofa_rows = fofa_swagger()
        shodan_rows = shodan_index_leaks()
        censys_ips = censys_cert_hosts(ORG)
        # crude filters to surface likely “interesting” items
        fofa_interesting = [f"- {h}:{p} — {t}" for h, p, t in fofa_rows if p in {"8080","9200","5601","3000"}]
        shodan_interesting = [f"- {ip}:{port} — {title}" for ip, port, title in shodan_rows]
        censys_lines = [f"- {ip} (cert SAN contains {ORG})" for ip in censys_ips]
        if fofa_interesting:
            post_slack("FOFA watch: Swagger UI with auth hints", fofa_interesting)
        if shodan_interesting:
            post_slack("Shodan watch: Index listings with secrets", shodan_interesting)
        if censys_lines:
            post_slack("Censys watch: Hosts presenting your domain in cert SAN", censys_lines)
    except Exception as e:
        post_slack("Automation error", [str(e)])