techstack.py

"""
Tech stack fingerprinting via HTTP headers, HTML content, and DNS/WHOIS signals.
No external Wappalyzer dependency — pure pattern matching.
"""

import re
import json
import asyncio
import shutil
from dataclasses import dataclass, field
from typing import Optional
import httpx

# ── Fingerprint signatures ────────────────────────────────────────────────────

# (pattern, category, name, confidence)
HEADER_SIGNATURES: list[tuple[str, str, str, str, int]] = [
    # field, pattern, category, name, confidence
    ("server",              r"nginx",               "web-server",   "Nginx",            90),
    ("server",              r"apache",              "web-server",   "Apache",           90),
    ("server",              r"cloudflare",          "cdn",          "Cloudflare",       95),
    ("server",              r"AmazonS3",            "hosting",      "AWS S3",           95),
    ("server",              r"openresty",           "web-server",   "OpenResty",        90),
    ("server",              r"Microsoft-IIS/([\d.]+)","web-server", "IIS",              90),
    ("server",              r"LiteSpeed",           "web-server",   "LiteSpeed",        90),
    ("server",              r"gunicorn",            "web-server",   "Gunicorn",         85),
    ("server",              r"uvicorn",             "web-server",   "Uvicorn",          85),
    ("x-powered-by",        r"PHP/([\d.]+)",        "language",     "PHP",              95),
    ("x-powered-by",        r"ASP\.NET",            "framework",    "ASP.NET",          95),
    ("x-powered-by",        r"Express",             "framework",    "Express.js",       95),
    ("x-powered-by",        r"Next\.js",            "framework",    "Next.js",          95),
    ("x-powered-by",        r"Nuxt",                "framework",    "Nuxt.js",          95),
    ("x-generator",         r"Drupal",              "cms",          "Drupal",           95),
    ("x-generator",         r"WordPress",           "cms",          "WordPress",        95),
    ("x-drupal-cache",      r"",                    "cms",          "Drupal",           90),
    ("x-wp-total",          r"",                    "cms",          "WordPress",        95),
    ("cf-ray",              r"",                    "cdn",          "Cloudflare",       95),
    ("x-vercel-id",         r"",                    "hosting",      "Vercel",           95),
    ("x-amz-cf-id",         r"",                    "cdn",          "AWS CloudFront",   95),
    ("x-amz-request-id",    r"",                    "hosting",      "AWS",              80),
    ("x-cache",             r"cloudfront",          "cdn",          "AWS CloudFront",   90),
    ("x-served-by",         r"cache-",              "cdn",          "Fastly",           85),
    ("via",                 r"varnish",             "cache",        "Varnish",          85),
    ("set-cookie",          r"PHPSESSID",           "language",     "PHP",              85),
    ("set-cookie",          r"laravel_session",     "framework",    "Laravel",          90),
    ("set-cookie",          r"JSESSIONID",          "language",     "Java",             85),
    ("set-cookie",          r"_rails",              "framework",    "Ruby on Rails",    90),
    ("set-cookie",          r"django",              "framework",    "Django",           85),
    ("content-security-policy", r"shopify",         "platform",     "Shopify",          90),
    ("x-shopify-stage",     r"",                    "platform",     "Shopify",          95),
    ("x-shopid",            r"",                    "platform",     "Shopify",          95),
]

HTML_SIGNATURES: list[tuple[str, str, str, int]] = [
    # pattern, category, name, confidence
    (r'<meta[^>]+name=["\']generator["\'][^>]+content=["\']WordPress ([\d.]+)',  "cms",          "WordPress",        95),
    (r'wp-content/|wp-includes/',                                                 "cms",          "WordPress",        90),
    (r'Drupal\.settings|sites/default/files',                                     "cms",          "Drupal",           90),
    (r'Joomla!|joomla',                                                           "cms",          "Joomla",           85),
    (r'ghost-url|content="Ghost',                                                 "cms",          "Ghost",            90),
    (r'__NEXT_DATA__|_next/static',                                               "framework",    "Next.js",          95),
    (r'<div id="__nuxt">|nuxt\.js',                                               "framework",    "Nuxt.js",          90),
    (r'data-reactroot|react-dom',                                                 "framework",    "React",            85),
    (r'ng-version=|angular\.js',                                                  "framework",    "Angular",          85),
    (r'data-v-[a-f0-9]+|vue\.js|vue\.min\.js',                                   "framework",    "Vue.js",           85),
    (r'Svelte|svelte',                                                            "framework",    "Svelte",           80),
    (r'ember\.js|Ember\.VERSION',                                                 "framework",    "Ember.js",         85),
    (r'cdn\.shopify\.com|Shopify\.theme',                                         "platform",     "Shopify",          95),
    (r'static\.wixstatic\.com|wix\.com',                                          "platform",     "Wix",              95),
    (r'squarespace\.com|static1\.squarespace',                                    "platform",     "Squarespace",      95),
    (r'webflow\.com|site-url.*webflow',                                           "platform",     "Webflow",          90),
    (r'framer\.com|framerusercontent',                                            "platform",     "Framer",           90),
    (r'gatsby-|gatsby\.js',                                                       "framework",    "Gatsby",           85),
    (r'astro-|@astrojs',                                                          "framework",    "Astro",            85),
    (r'bootstrap\.min\.css|bootstrap\.css',                                       "css-framework","Bootstrap",        80),
    (r'tailwindcss|tailwind\.min\.css',                                           "css-framework","Tailwind CSS",     80),
    (r'jquery\.min\.js|jquery-[0-9]',                                             "library",      "jQuery",           85),
    (r'gtag\(|googletagmanager\.com',                                             "analytics",    "Google Analytics", 90),
    (r'segment\.io|analytics\.js',                                                "analytics",    "Segment",          85),
    (r'hotjar\.com',                                                              "analytics",    "Hotjar",           90),
    (r'intercom\.io|intercomSettings',                                            "support",      "Intercom",         90),
    (r'crisp\.chat|CRISP_WEBSITE_ID',                                             "support",      "Crisp",            90),
    (r'stripe\.com/v3|Stripe\(',                                                  "payment",      "Stripe",           90),
    (r'graphql|__typename',                                                       "api",          "GraphQL",          75),
    (r'data-sentry-dsn|@sentry',                                                  "monitoring",   "Sentry",           85),
]

CATEGORY_ICONS = {
    "web-server":   "🖥️",
    "cdn":          "🌐",
    "hosting":      "☁️",
    "language":     "💻",
    "framework":    "⚙️",
    "cms":          "📝",
    "platform":     "🏪",
    "cache":        "⚡",
    "css-framework":"🎨",
    "library":      "📦",
    "analytics":    "📊",
    "support":      "💬",
    "payment":      "💳",
    "api":          "🔌",
    "monitoring":   "🔍",
}


@dataclass
class Technology:
    name: str
    category: str
    confidence: int
    version: Optional[str] = None
    icon: str = ""

    def __post_init__(self):
        self.icon = CATEGORY_ICONS.get(self.category, "🔧")

    def dict(self):
        return {
            "name": self.name,
            "category": self.category,
            "confidence": self.confidence,
            "version": self.version,
            "icon": self.icon,
        }


@dataclass
class TechStackResult:
    url: str
    technologies: list[Technology] = field(default_factory=list)
    elapsed_ms: float = 0.0
    status_code: int = 0

    def dict(self):
        # Group by category
        by_category: dict[str, list] = {}
        seen = set()
        for t in sorted(self.technologies, key=lambda x: -x.confidence):
            key = f"{t.name}-{t.category}"
            if key in seen:
                continue
            seen.add(key)
            by_category.setdefault(t.category, []).append(t.dict())

        return {
            "url": self.url,
            "status_code": self.status_code,
            "elapsed_ms": self.elapsed_ms,
            "technologies": by_category,
            "summary": [t.name for t in self.technologies if t.confidence >= 80 and f"{t.name}-{t.category}" in seen],
        }


async def detect_tech_stack(url: str, timeout: float = 15.0) -> TechStackResult:
    result = TechStackResult(url=url)

    headers_to_send = {
        "User-Agent": "Mozilla/5.0 (compatible; TechStackBot/1.0)",
        "Accept": "text/html,application/xhtml+xml,*/*",
        "Accept-Language": "en-US,en;q=0.9",
    }

    import time
    start = time.perf_counter()

    async with httpx.AsyncClient(
        follow_redirects=True,
        timeout=timeout,
        verify=True,
    ) as client:
        response = await client.get(url, headers=headers_to_send)

    result.elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
    result.status_code = response.status_code

    resp_headers = {k.lower(): v for k, v in response.headers.items()}
    html = response.text

    # ── Header fingerprinting ─────────────────────────────────────────────────
    for field_name, pattern, category, name, confidence in HEADER_SIGNATURES:
        val = resp_headers.get(field_name, "")
        if not val:
            continue
        if pattern == "" or re.search(pattern, val, re.IGNORECASE):
            version = None
            if pattern:
                m = re.search(pattern, val, re.IGNORECASE)
                if m and m.lastindex:
                    version = m.group(1)
            result.technologies.append(Technology(name, category, confidence, version))

    # ── HTML fingerprinting ───────────────────────────────────────────────────
    for pattern, category, name, confidence in HTML_SIGNATURES:
        m = re.search(pattern, html, re.IGNORECASE)
        if m:
            version = m.group(1) if m.lastindex else None
            result.technologies.append(Technology(name, category, confidence, version))

    return result


async def _discover_crtsh(domain: str, timeout: float) -> set[str]:
    """Query crt.sh Certificate Transparency logs for known subdomains."""
    crtsh_url = f"https://crt.sh/?q=%.{domain}&output=json"
    found: set[str] = set()

    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
            response = await client.get(
                crtsh_url,
                headers={"User-Agent": "Mozilla/5.0 (compatible; TechStackBot/1.0)"},
            )
            response.raise_for_status()
            entries = response.json()
    except Exception:
        return found

    for entry in entries:
        for name in entry.get("name_value", "").splitlines():
            name = name.strip().lower()
            if name.startswith("*."):
                name = name[2:]
            if name == domain or name.endswith(f".{domain}"):
                found.add(name)

    return found


async def _discover_subfinder(domain: str, timeout: float) -> set[str]:
    """
    Run the subfinder binary for passive subdomain enumeration.

    Silently returns an empty set if subfinder is not installed.
    The domain is passed directly as a subprocess argument (no shell=True),
    so the already-validated domain string cannot cause injection.
    """
    if not shutil.which("subfinder"):
        return set()

    try:
        proc = await asyncio.create_subprocess_exec(
            "subfinder", "-d", domain, "-silent",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        try:
            stdout, _ = await asyncio.wait_for(
                proc.communicate(),
                timeout=timeout * 3,  # subfinder fans out across many sources
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.communicate()
            return set()

        return {
            line.strip().lower()
            for line in stdout.decode().splitlines()
            if line.strip() and line.strip().endswith(f".{domain}" ) or line.strip() == domain
        }
    except Exception:
        return set()


async def discover_subdomains(domain: str, timeout: float = 10.0) -> list[str]:
    """
    Discover subdomains by running crt.sh and subfinder concurrently.

    Results from both sources are merged, deduplicated, and returned sorted.
    Either source failing gracefully returns an empty contribution.
    The root domain itself is excluded — the caller adds it as the first target.
    """
    crtsh_results, subfinder_results = await asyncio.gather(
        _discover_crtsh(domain, timeout),
        _discover_subfinder(domain, timeout),
    )

    combined = crtsh_results | subfinder_results
    combined.discard(domain)
    return sorted(combined)