syzkaller_crawler.py

import json
import logging
import re
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union
from urllib.parse import urljoin

import pandas as pd
import requests
from bs4 import BeautifulSoup, Tag

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load Linux version mappings from CSV
LINUX_VERSIONS: Dict[str, str] = {}
try:
    versions_df = pd.read_csv("linux_versions.csv")
    LINUX_VERSIONS = dict(zip(versions_df["commit"], versions_df["version"]))
    logger.info(f"Loaded {len(LINUX_VERSIONS)} Linux version mappings")
except FileNotFoundError:
    logger.warning("linux_versions.csv not found - linux_version field will be empty")
except Exception as e:
    logger.warning(
        f"Error loading linux_versions.csv: {e} - linux_version field will be empty"
    )


def lookup_linux_version(commit_hash: str) -> Optional[str]:
    """
    Lookup Linux version tag from commit hash using the loaded CSV data.

    Args:
        commit_hash: Git commit hash to lookup

    Returns:
        Linux version tag if found, None otherwise
    """
    if not commit_hash:
        return None

    return LINUX_VERSIONS.get(commit_hash, None)


@dataclass
class BugReport:
    """Structured data model for a syzkaller bug report."""

    bug_id: str
    title: str
    bug_url: str
    reproducer_type: Optional[str] = None
    reproducer_url: Optional[str] = None
    config_url: Optional[str] = None
    disk_url: Optional[str] = None
    linux_commit: Optional[str] = None
    linux_version: Optional[str] = None
    architecture: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary representation."""
        return {
            "bug_id": self.bug_id,
            "title": self.title,
            "bug_url": self.bug_url,
            "reproducer_type": self.reproducer_type,
            "reproducer_url": self.reproducer_url,
            "config_url": self.config_url,
            "disk_url": self.disk_url,
            "linux_commit": self.linux_commit,
            "linux_version": self.linux_version,
            "architecture": self.architecture,
        }


@dataclass
class CrawlResult:
    """Results summary for a crawl operation."""

    total_bugs: int
    filtered_bugs: int
    crawl_duration: float
    errors: List[str] = field(default_factory=list)

    @property
    def success_rate(self) -> float:
        """Calculate success rate as percentage."""
        return (
            (self.filtered_bugs / self.total_bugs * 100) if self.total_bugs > 0 else 0.0
        )


def get_syzkaller_bug_table(
    url: str = "https://syzkaller.appspot.com/linux-6.1/fixed",
    timeout: int = 30,
    delay: float = 1.0,
) -> Optional[pd.DataFrame]:
    """
    Retrieve HTML page from Syzkaller and parse the bug table into a pandas DataFrame.

    Enhanced version with better error handling, rate limiting, and structured data extraction.

    Args:
        url: The URL to fetch the bug table from
        timeout: Request timeout in seconds
        delay: Delay between requests in seconds (for rate limiting)

    Returns:
        pandas DataFrame containing structured bug table data, or None if failed
    """
    try:
        # Set headers to mimic a browser request
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Accept-Encoding": "gzip, deflate",
            "Connection": "keep-alive",
        }

        logger.info(f"Fetching bug table from: {url}")

        # Rate limiting
        time.sleep(delay)

        # Make the HTTP request
        response = requests.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()

        # Parse the HTML content
        soup = BeautifulSoup(response.content, "html.parser")

        # Find all tables on the page
        tables = soup.find_all("table")
        bug_table = None

        # Enhanced table detection
        for table in tables:
            header_row = table.find("tr")
            if header_row:
                headers_text = [
                    th.get_text(strip=True).lower()
                    for th in header_row.find_all(["th", "td"])
                ]
                # Check if this looks like the bug table by looking for key columns
                if any(
                    keyword in " ".join(headers_text) for keyword in ["title", "repro"]
                ):
                    bug_table = table
                    logger.info(f"Found bug table with headers: {headers_text}")
                    break

        if not bug_table:
            logger.error(
                "No bug table found with expected headers (Title, Repro, etc.)"
            )
            return None

        # Extract structured data
        structured_data = extract_structured_table_data(bug_table, url)

        if structured_data:
            df = pd.DataFrame(structured_data)
            logger.info(
                f"Successfully parsed table with {len(df)} rows and {len(df.columns)} columns"
            )
            return df
        else:
            logger.warning("No structured data extracted from table")
            return None

    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching URL {url}: {e}")
        return None
    except Exception as e:
        logger.error(f"Error parsing HTML from {url}: {e}")
        return None


def extract_structured_table_data(table: Tag, base_url: str) -> List[Dict[str, Any]]:
    """
    Extract structured data from a parsed HTML table.

    Args:
        table: BeautifulSoup table element
        base_url: Base URL for resolving relative links

    Returns:
        List of dictionaries containing structured bug data
    """
    # Extract table headers with better column mapping
    headers = []
    header_row = table.find("tr")
    if header_row:
        for th in header_row.find_all(["th", "td"]):
            header_text = th.get_text(strip=True)
            headers.append(header_text)

    # Extract table rows with structured data
    structured_data = []
    data_rows = table.find_all("tr")[1:]  # Skip header row

    for i, row in enumerate(data_rows):
        cells = row.find_all(["td", "th"])
        if not cells:
            continue

        row_data = {}

        for j, cell in enumerate(cells):
            # Determine column name
            col_name = headers[j] if j < len(headers) else f"column_{j}"

            # Extract cell content and links
            cell_text = cell.get_text(strip=True)
            links = extract_links(cell, base_url)

            # Structure the data based on column type
            if col_name.lower() in ["title"]:
                row_data.update(
                    {
                        "title": cell_text,
                        "bug_url": links[0] if links else None,
                        "bug_id": extract_bug_id(links[0]) if links else None,
                    }
                )
            elif col_name.lower() in ["repro", "reproducer"]:
                # Check for text indicators (C, syz) in the cell text
                # Determine reproducer type based only on text indicators
                if "C" in cell_text.upper():
                    # default to c even if "c" and "syz" are both present
                    reproducer_type = "c"
                elif "SYZ" in cell_text.upper():
                    reproducer_type = "syz"
                else:
                    reproducer_type = "unknown"

                row_data.update(
                    {
                        "reproducer_type": reproducer_type,
                    }
                )

        # Add row index for reference and a field for C repro URLs from detail pages
        row_data["row_index"] = i
        structured_data.append(row_data)

    return structured_data


def extract_links(cell: Tag, base_url: str) -> List[str]:
    """
    Extract all links from a table cell.

    Args:
        cell: BeautifulSoup cell element
        base_url: Base URL for resolving relative links

    Returns:
        List of absolute URLs
    """
    links = []
    for link in cell.find_all("a"):
        href = link.get("href")
        if href:
            if str(href).startswith("http"):
                absolute_url = href
            elif str(href).startswith("/"):
                absolute_url = urljoin(base_url, str(href))
            else:
                raise ValueError(f"Unexpected link format: {href}")
            links.append(absolute_url)
    return links


def extract_bug_id(bug_url: str) -> Optional[str]:
    """
    Extract bug ID from a syzkaller bug URL.

    Args:
        bug_url: URL to the bug report

    Returns:
        Bug ID if found, None otherwise
    """
    if not bug_url:
        return None

    # Pattern for syzkaller bug URLs
    match = re.search(r"\/bug\?(ext)?id=([a-f0-9-]+)", bug_url)
    if match:
        if match.group(2):
            return match.group(2)
        return match.group(1)

    return None


def export_to_json(
    df: pd.DataFrame, filepath: Union[str, Path], include_metadata: bool = True
) -> bool:
    """
    Export DataFrame to JSON format.

    Args:
        df: DataFrame to export
        filepath: Output file path
        include_metadata: Whether to include metadata like export timestamp

    Returns:
        True if successful, False otherwise
    """
    try:
        filepath = Path(filepath)
        filepath.parent.mkdir(parents=True, exist_ok=True)

        export_data = (
            {
                "bugs": df.to_dict("records"),
                "export_timestamp": datetime.now().isoformat(),
                "total_bugs": len(df),
                "columns": list(df.columns),
            }
            if include_metadata
            else df.to_dict("records")
        )

        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(export_data, f, indent=2, default=str, ensure_ascii=False)

        logger.info(f"Exported {len(df)} bugs to JSON: {filepath}")
        return True

    except Exception as e:
        logger.error(f"Error exporting to JSON {filepath}: {e}")
        return False


def filter_bugs_by_architecture(
    df: pd.DataFrame, target_arch: str = "x86"
) -> pd.DataFrame:
    """
    Filter bugs by target architecture.

    Args:
        df: DataFrame containing bug data
        target_arch: Target architecture to filter for (x86, ARM, etc.)

    Returns:
        Filtered DataFrame containing only bugs for the specified architecture
    """
    if "architecture" not in df.columns:
        logger.warning("Architecture column not found in DataFrame")
        return df

    filtered_df = df[df["architecture"] == target_arch].copy()
    logger.info(f"Filtered {len(df)} bugs to {len(filtered_df)} {target_arch} bugs")
    return filtered_df


def filter_bugs_with_c_reproducer(df: pd.DataFrame) -> pd.DataFrame:
    """
    Filter bugs that have C reproducers available.

    Args:
        df: DataFrame containing bug data

    Returns:
        Filtered DataFrame containing only bugs with reproducers
    """
    if "reproducer_type" not in df.columns:
        logger.warning("reproducer_type column not found in DataFrame")
        return df

    filtered_df = df[df["reproducer_type"] == "c"].copy()
    logger.info(f"Filtered {len(df)} bugs to {len(filtered_df)} bugs with reproducers")
    return filtered_df


def enhance_bug_table(df: pd.DataFrame) -> pd.DataFrame:
    enhanced_columns = [
        "reproducer_url",
        "config_url",
        "disk_url",
        "linux_commit",
        "linux_version",
        "architecture",
    ]
    enhanced_df = df.copy()
    for col in enhanced_columns:
        if col not in df.columns:
            enhanced_df[col] = None

    assert "bug_url" in df.columns, "bug_url column is required for enhancement"
    # for each bug with a bug_url, try to extract missing details from the detail page
    for index, row in enhanced_df.iterrows():
        bug_details = extract_earliest_c_repro_x86_item_from_detail_page(row["bug_url"])
        if bug_details:
            for col in enhanced_columns:
                if bug_details.get(col, None) and not row.get(col):
                    enhanced_df.at[index, col] = bug_details[col]
        time.sleep(1)  # rate limiting

    return enhanced_df


def create_data_directories(base_dir: str = "data") -> Tuple[Path, Path]:
    """
    Create the standard data directory structure.

    Args:
        base_dir: Base directory for data storage

    Returns:
        Tuple of Path objects: (raw_dir, filtered_dir)
    """
    base_path = Path(base_dir)

    # Create directories
    raw_dir = base_path / "raw_bugs"
    filtered_dir = base_path / "filtered_bugs"

    for directory in [raw_dir, filtered_dir]:
        directory.mkdir(parents=True, exist_ok=True)
        logger.info(f"Created directory: {directory}")

    return raw_dir, filtered_dir


def extract_earliest_c_repro_x86_item_from_detail_page(
    bug_url: str, timeout: int = 30
) -> Optional[Dict[str, Optional[str]]]:
    """
    Extract the earliest bug entry with C reproducer from a syzkaller bug detail page.

    This function parses the table on the bug detail page and finds the first entry
    (earliest by time) that has a non-empty C repro column, returning the C repro link
    and other related information including disk URL.

    Args:
        bug_url: URL to the individual bug report detail page
        timeout: Request timeout in seconds

    Returns:
        Dictionary with keys: reproducer_url, config_url, disk_url, linux_commit,
        linux_version, architecture
    """
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Accept-Encoding": "gzip, deflate",
            "Connection": "keep-alive",
        }

        logger.info(f"Extracting earliest C repro from detail page: {bug_url}")

        # Rate limiting
        time.sleep(0.5)

        # Retry logic with maximum 3 attempts
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = requests.get(bug_url, headers=headers, timeout=timeout)
                response.raise_for_status()
                break  # Success, exit retry loop
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:  # Last attempt
                    raise
                logger.warning(
                    f"Request failed (attempt {attempt + 1}/{max_retries}): {e}"
                )
                time.sleep(1 * (attempt + 1))  # Exponential backoff

        soup = BeautifulSoup(response.content, "html.parser")

        # Find the table that contains the crash reports with C repro column
        tables = soup.find_all("table")
        crash_table = None

        for table in tables:
            header_row = table.find("tr")
            if header_row:
                headers_text = [
                    th.get_text(strip=True).lower()
                    for th in header_row.find_all(["th", "td"])
                ]
                # Look for the crash table by checking for "c repro" column
                if "c repro" in headers_text:
                    crash_table = table
                    logger.info(f"Found crash table with headers: {headers_text}")
                    break

        if not crash_table:
            logger.warning(
                f"No crash table with 'C repro' column found on page: {bug_url}"
            )
            return None

        # Extract table data
        data_rows = crash_table.find_all("tr")[1:]  # Skip header row

        # Extract headers to find C repro column index
        headers = []
        header_row = crash_table.find("tr")
        if header_row:
            for th in header_row.find_all(["th", "td"]):
                header_text = th.get_text(strip=True)
                headers.append(header_text)

        # Find the index of the "C repro" column
        c_repro_col_idx = None
        config_col_idx = None
        linux_commit_col_idx = None
        manager_col_idx = None
        assets_col_idx = None
        for i, header in enumerate(headers):
            if header.lower() == "c repro":
                c_repro_col_idx = i
            elif header.lower() == "config":
                config_col_idx = i
            elif header.lower() == "commit":
                linux_commit_col_idx = i
            elif header.lower() == "manager":
                manager_col_idx = i
            elif "assets" in header.lower():
                assets_col_idx = i

        if c_repro_col_idx is None:
            logger.warning(
                f"Could not find 'C repro' column in table headers: {headers}"
            )
            return None
        if config_col_idx is None:
            logger.warning(
                f"Could not find 'Config' column in table headers: {headers}"
            )
            return None
        if linux_commit_col_idx is None:
            logger.warning(
                f"Could not find 'Commit' column in table headers: {headers}"
            )
            return None
        if manager_col_idx is None:
            logger.warning(
                f"Could not find 'Manager' column in table headers: {headers}"
            )
            return None

        # iterate through rows to find the earliest entry with C repro
        column_indices = [
            idx
            for idx in [
                c_repro_col_idx,
                config_col_idx,
                linux_commit_col_idx,
                manager_col_idx,
                assets_col_idx,
            ]
            if idx is not None
        ]
        _max_col_idx = max(column_indices)
        for row in data_rows:
            cells = row.find_all(["td", "th"])
            if len(cells) <= _max_col_idx:
                continue

            # architecture
            manager_cell = cells[manager_col_idx]
            architecture = (
                "arm" if "arm" in manager_cell.get_text(strip=True).lower() else "x86"
            )
            if architecture != "x86":
                continue

            # c reproducer
            c_repro_cell = cells[c_repro_col_idx]
            # Check if this cell has a link
            c_repro_links = extract_links(c_repro_cell, "https://syzkaller.appspot.com")
            if not c_repro_links:
                continue
            # Found the earliest entry with C repro
            assert len(c_repro_links) == 1, "Expected exactly one C repro link"
            c_repro_url = c_repro_links[0]
            logger.info(f"Found earliest C repro link: {c_repro_url}")

            # config
            config_cell = cells[config_col_idx]
            config_links = extract_links(config_cell, "https://syzkaller.appspot.com")
            if not config_links:
                config_url = None
            else:
                assert len(config_links) == 1, "Expected exactly one config link"
                config_url = config_links[0]
            if config_url:
                logger.info(f"Found config link: {config_url}")
            else:
                logger.warning(f"No config link found in cell: {config_cell}")

            # linux commit
            linux_commit_cell = cells[linux_commit_col_idx]
            linux_commit_links = extract_links(
                linux_commit_cell, "https://syzkaller.appspot.com"
            )
            assert len(linux_commit_links) == 1, (
                "Expected exactly one linux commit link"
            )

            # Filter out non-mainline kernel commits (e.g., KMSAN, other forks)
            commit_link = linux_commit_links[0]
            if not commit_link.startswith(
                "https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/log/?id="
            ):
                logger.info(
                    f"Skipping row with non-mainline commit link: {commit_link}"
                )
                continue  # Skip this row and continue to next one

            linux_commit = commit_link.split("?id=")[-1]
            if not linux_commit:
                logger.warning(f"No Linux commit found in cell: {linux_commit_cell}")
                continue  # Skip this row if commit parsing failed

            logger.info(f"Found Linux commit: {linux_commit}")

            # disk URL from assets column
            disk_url = None
            if assets_col_idx is not None:
                assets_cell = cells[assets_col_idx]
                assets_links = extract_links(
                    assets_cell, "https://syzkaller.appspot.com"
                )
                # Look for disk image links (containing "disk" in the URL)
                for link in assets_links:
                    if "disk" in link.lower():
                        disk_url = link
                        logger.info(f"Found disk URL: {disk_url}")
                        break
                if not disk_url:
                    logger.debug(f"No disk URL found in assets cell: {assets_cell}")

            linux_version = lookup_linux_version(linux_commit)

            return {
                "reproducer_url": c_repro_url,
                "config_url": config_url,
                "disk_url": disk_url,
                "linux_commit": linux_commit,
                "linux_version": linux_version,
                "architecture": architecture,
            }

        logger.info(f"No entries with appropriate conditions found on page: {bug_url}")
        return None

    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching bug detail page {bug_url}: {e}")
        return None
    except Exception as e:
        logger.error(f"Error parsing bug detail page {bug_url}: {e}")
        return None


def main():
    """
    Main function for running the crawler with command line interface.
    """
    import argparse

    parser = argparse.ArgumentParser(description="Syzkaller Bug Crawler")
    parser.add_argument(
        "--url",
        default="https://syzkaller.appspot.com/linux-6.1/fixed",
        help="URL to crawl for bugs",
    )
    parser.add_argument("--arch", default=None, help="Filter by architecture")
    parser.add_argument(
        "--c-reproducer-only",
        action="store_true",
        help="Only include bugs with C reproducers",
    )
    parser.add_argument("--output", default="data", help="Output directory")
    parser.add_argument(
        "--test",
        action="store_true",
        help="Run in test mode with limited data",
    )

    args = parser.parse_args()

    # Create data directories
    raw_dir, filtered_dir = create_data_directories(args.output)

    # Crawl the bug table
    logger.info(f"Starting crawl of {args.url}")
    start_time = time.time()

    df = get_syzkaller_bug_table(args.url)
    if df is None:
        logger.error("Failed to retrieve bug table")
        return
    if args.test:
        df = df.head(50)
        logger.info("Test mode: limited to first 50 bugs")

    # Apply filters
    filtered_df = df.copy()

    if args.c_reproducer_only:
        filtered_df = filter_bugs_with_c_reproducer(filtered_df)

    filtered_df = enhance_bug_table(filtered_df)

    if args.arch:
        filtered_df = filter_bugs_by_architecture(filtered_df, args.arch)

    # Save raw and filtered data
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    # Save raw data
    raw_file = raw_dir / f"raw_bugs_{timestamp}.json"
    export_to_json(df, raw_file)

    # Save filtered data
    if not filtered_df.empty:
        filtered_file = filtered_dir / f"filtered_bugs_{timestamp}.json"
        export_to_json(filtered_df, filtered_file)

    # Generate summary
    end_time = time.time()
    duration = end_time - start_time

    logger.info("\nCrawl Summary:")
    logger.info(f"  Total bugs: {len(df)}")
    logger.info(f"  Filtered bugs: {len(filtered_df)}")
    logger.info(f"  Duration: {duration:.2f} seconds")
    logger.info(f"  Success rate: {(len(filtered_df) / len(df) * 100):.1f}%")
    logger.info(f"  Data saved to: {args.output}")


if __name__ == "__main__":
    main()