Chapter 19: Bounty Hunting and Vulnerability Research¶

import asyncio
import json
import requests
from datetime import datetime
from dataclasses import dataclass, field
from typing import List, Dict, Set, Optional
import subprocess

@dataclass
class ReconTarget:
    """Target organization for security research reconnaissance"""
    organization: str
    program_url: str
    in_scope_domains: List[str]
    in_scope_ip_ranges: List[str]
    out_of_scope: List[str]
    notes: str = ""

@dataclass
class AttackSurface:
    """Discovered attack surface for a target"""
    target: ReconTarget
    domains: Set[str] = field(default_factory=set)
    subdomains: Set[str] = field(default_factory=set)
    ip_addresses: Set[str] = field(default_factory=set)
    open_ports: Dict[str, List[int]] = field(default_factory=dict)
    web_technologies: Dict[str, Dict] = field(default_factory=dict)
    interesting_endpoints: List[str] = field(default_factory=list)
    certificates: List[Dict] = field(default_factory=list)
    github_exposures: List[Dict] = field(default_factory=list)
    discovery_timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class SecurityReconWorkflow:
    """
    Authorized security research reconnaissance workflow
    All methods are for use within authorized bug bounty or pen test scope only
    """

    def __init__(self, target: ReconTarget):
        self.target = target
        self.surface = AttackSurface(target=target)
        self.log = []

    def log_action(self, action: str, details: str):
        entry = {
            'timestamp': datetime.now().isoformat(),
            'action': action,
            'details': details
        }
        self.log.append(entry)
        print(f"[{entry['timestamp'][:19]}] {action}: {details}")

    def verify_scope(self, domain_or_ip: str) -> bool:
        """
        Verify an asset is within scope before any testing
        CRITICAL: Always verify scope before taking any action
        """
        # Check if in explicitly out-of-scope list
        for exclusion in self.target.out_of_scope:
            if exclusion.lower() in domain_or_ip.lower():
                return False

        # Check if domain is in scope
        for scope_domain in self.target.in_scope_domains:
            if domain_or_ip.endswith(scope_domain) or domain_or_ip == scope_domain:
                return True

        # Check if IP is in scope ranges
        try:
            import ipaddress
            for ip_range in self.target.in_scope_ip_ranges:
                network = ipaddress.ip_network(ip_range, strict=False)
                if ipaddress.ip_address(domain_or_ip) in network:
                    return True
        except ValueError:
            pass  # Not an IP address

        return False

    def passive_subdomain_discovery(self) -> Set[str]:
        """
        Discover subdomains using only passive/public data sources
        No active probing of target systems
        """
        discovered = set()

        for root_domain in self.target.in_scope_domains:
            self.log_action("passive_recon", f"Discovering subdomains for {root_domain}")

            # 1. Certificate Transparency
            try:
                response = requests.get(
                    f"https://crt.sh/?q=%.{root_domain}&output=json",
                    timeout=30
                )
                if response.status_code == 200:
                    for cert in response.json():
                        for name in cert.get('name_value', '').split('\n'):
                            name = name.strip().lstrip('*.')
                            if name.endswith(root_domain) and self.verify_scope(name):
                                discovered.add(name)

                self.log_action("ct_discovery", f"Found {len(discovered)} subdomains via CT logs")
            except Exception as e:
                self.log_action("ct_error", str(e))

            # 2. HackerTarget subdomain search (free API)
            try:
                response = requests.get(
                    f"https://api.hackertarget.com/hostsearch/?q={root_domain}",
                    timeout=30
                )
                for line in response.text.strip().split('\n'):
                    if ',' in line:
                        subdomain = line.split(',')[0].strip()
                        if subdomain.endswith(root_domain) and self.verify_scope(subdomain):
                            discovered.add(subdomain)
            except Exception as e:
                self.log_action("hackertarget_error", str(e))

            # 3. SecurityTrails (if API key available)
            # ... additional passive sources

        self.surface.subdomains.update(discovered)
        self.log_action("subdomain_discovery_complete", f"Total: {len(self.surface.subdomains)} unique subdomains")
        return discovered

    def technology_fingerprinting(self, urls: List[str]) -> Dict[str, Dict]:
        """
        Identify web technologies from public HTTP headers and responses
        Uses only HEAD/GET requests to discovered URLs
        """
        technologies = {}

        for url in urls[:50]:  # Limit initial batch
            # Verify scope before any request
            from urllib.parse import urlparse
            domain = urlparse(url).netloc
            if not self.verify_scope(domain):
                continue

            try:
                response = requests.head(url, timeout=10, allow_redirects=True,
                                        headers={'User-Agent': 'Security Research Bot'})

                tech = {
                    'url': response.url,
                    'status': response.status_code,
                    'server': response.headers.get('Server', ''),
                    'powered_by': response.headers.get('X-Powered-By', ''),
                    'framework': response.headers.get('X-Framework', ''),
                    'content_type': response.headers.get('Content-Type', ''),
                    'security_headers': {
                        'HSTS': bool(response.headers.get('Strict-Transport-Security')),
                        'CSP': bool(response.headers.get('Content-Security-Policy')),
                        'X-Frame-Options': bool(response.headers.get('X-Frame-Options')),
                        'X-Content-Type-Options': bool(response.headers.get('X-Content-Type-Options')),
                    }
                }

                technologies[url] = tech

            except Exception as e:
                technologies[url] = {'error': str(e)}

        self.surface.web_technologies.update(technologies)
        return technologies

    def github_code_search(self, organization: str) -> List[Dict]:
        """
        Search GitHub for code exposures related to the target
        Uses GitHub's public search API
        """
        exposures = []
        headers = {'Accept': 'application/vnd.github.v3+json'}

        # Queries that commonly reveal sensitive information
        queries = [
            f'org:{organization} password',
            f'org:{organization} secret_key',
            f'org:{organization} api_key',
            f'org:{organization} DB_PASSWORD',
            f'"{organization}" api_key site:github.com',
        ]

        for query in queries:
            try:
                response = requests.get(
                    'https://api.github.com/search/code',
                    params={'q': query, 'per_page': 10},
                    headers=headers,
                    timeout=10
                )

                if response.status_code == 200:
                    results = response.json()
                    for item in results.get('items', []):
                        exposures.append({
                            'query': query,
                            'repo': item.get('repository', {}).get('full_name'),
                            'file': item.get('path'),
                            'url': item.get('html_url'),
                            'note': 'Potential credential/secret exposure'
                        })

            except Exception as e:
                self.log_action("github_search_error", str(e))

        self.surface.github_exposures = exposures
        if exposures:
            self.log_action("github_exposure", f"Found {len(exposures)} potential code exposures")
        return exposures

    def generate_recon_report(self) -> str:
        """Generate a structured recon report for security research"""
        report = [
            f"# Reconnaissance Report",
            f"**Organization**: {self.target.organization}",
            f"**Program**: {self.target.program_url}",
            f"**Discovery Date**: {self.surface.discovery_timestamp[:10]}",
            "",
            "## Attack Surface Summary",
            f"- **Domains enumerated**: {len(self.surface.domains)}",
            f"- **Subdomains discovered**: {len(self.surface.subdomains)}",
            f"- **IP addresses identified**: {len(self.surface.ip_addresses)}",
            f"- **Potential code exposures**: {len(self.surface.github_exposures)}",
            "",
            "## Subdomains Discovered",
            "",
        ]

        for subdomain in sorted(self.surface.subdomains):
            tech = self.surface.web_technologies.get(f"https://{subdomain}", {})
            server = tech.get('server', 'Unknown')
            status = tech.get('status', 'Not probed')
            report.append(f"- `{subdomain}` — Status: {status}, Server: {server}")

        if self.surface.github_exposures:
            report.extend([
                "",
                "## Potential Code Exposures (GitHub)",
                "**Note**: Verify these are valid findings before reporting",
                ""
            ])
            for exposure in self.surface.github_exposures:
                report.append(f"- [{exposure['repo']}/{exposure['file']}]({exposure['url']})")
                report.append(f"  Query: `{exposure['query']}`")

        report.extend([
            "",
            "## Security Header Analysis",
            "",
            "| Endpoint | HSTS | CSP | X-Frame | X-Content-Type |",
            "|---|---|---|---|---|"
        ])

        for url, tech in list(self.surface.web_technologies.items())[:20]:
            headers = tech.get('security_headers', {})
            report.append(
                f"| {url[:50]} | {'✓' if headers.get('HSTS') else '✗'} | "
                f"{'✓' if headers.get('CSP') else '✗'} | "
                f"{'✓' if headers.get('X-Frame-Options') else '✗'} | "
                f"{'✓' if headers.get('X-Content-Type-Options') else '✗'} |"
            )

        report.extend([
            "",
            "## Reconnaissance Log",
            ""
        ])
        for entry in self.log:
            report.append(f"- [{entry['timestamp'][:19]}] **{entry['action']}**: {entry['details']}")

        return '\n'.join(report)

import anthropic

def ai_assisted_attack_surface_analysis(recon_data: dict) -> str:
    """
    Use AI to analyze recon findings and suggest areas of interest
    """
    client = anthropic.Anthropic()

    # Format recon data for AI analysis
    recon_text = json.dumps(recon_data, indent=2)

    prompt = f"""You are an experienced security researcher analyzing reconnaissance data for a bug bounty program.

RECONNAISSANCE DATA:
{recon_text[:4000]}

Based on this reconnaissance data, analyze:

1. HIGH-VALUE TARGETS
   - Which subdomains or endpoints are most likely to contain vulnerabilities?
   - What patterns suggest less-maintained or forgotten systems?
   - What technologies suggest known vulnerability classes?

2. ATTACK VECTORS TO INVESTIGATE
   - Based on the technology stack identified, what vulnerability classes should be prioritized?
   - Which security header gaps are most significant?
   - What test cases should be constructed?

3. INFORMATION LEAKAGE
   - What information in the recon data could be sensitive?
   - What code exposure findings warrant immediate investigation?

4. RECOMMENDED NEXT STEPS
   - Prioritized list of investigation targets
   - Specific test scenarios to try within scope

Focus on actionable, specific recommendations. Note that all testing must remain within the defined program scope."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text