10_audit_binary_code.py

from pathlib import Path
import re

import pandas as pd

PROJECT_DIR = Path(__file__).resolve().parent
INPUT_PATH = PROJECT_DIR / "SBAN_clean.parquet"
REPORT_DIR = PROJECT_DIR / "reports" / "audit"

COLUMN = "binary_code"
ID_COL = "ID"
LABEL_COL = "dataset_name"

HEX_LINE_RE = re.compile(r"^[0-9A-Fa-f]+$")

PROMPT_PHRASES = (
    "paste the full source code",
    "insert source code",
    "paste code",
    "your code",
    "add main function",
    "code goes",
    "implementation goes",
    "insert code",
    "no modifications needed",
    "corrected",
    "here",
    "no comments",
    "additional",
)

ASSISTANT_PHRASES = (
    "here is the",
    "here's the",
    "below is",
    "as follows",
    "sure!",
    "certainly",
    "without comments",
    "only the code",
    "only code",
    "corrected version",
    "fix the code",
    "complete_cleaned_code",
    "main function is missing",
    "no c++ solution",
    "no output as the code",
)

SOURCE_LIKE_RE = re.compile(
    r"#include|#define|\bint\s+main\s*\(|\bvoid\s+main\s*\(",
    re.I,
)

ASM_LIKE_RE = re.compile(
    r"^\s*(mov|push|pop|call|ret|jmp|lea)\s+",
    re.I | re.M,
)


def load_data() -> pd.DataFrame:
    if not INPUT_PATH.exists():
        raise FileNotFoundError(f"Dosya bulunamadı: {INPUT_PATH}")
    return pd.read_parquet(INPUT_PATH, columns=[ID_COL, LABEL_COL, COLUMN])


def hex_line_stats(text: str) -> tuple[int, int, int]:
    valid_lines = 0
    invalid_lines = 0
    hex_chars = 0

    for line in str(text).splitlines():
        line = "".join(line.split())
        if not line:
            continue
        if HEX_LINE_RE.fullmatch(line):
            valid_lines += 1
            hex_chars += len(line) - (len(line) % 2)
        else:
            invalid_lines += 1

    return valid_lines, invalid_lines, hex_chars


def byte_token_count(text: str) -> int:
    count = 0
    for line in str(text).splitlines():
        line = "".join(line.split())
        if not line or not HEX_LINE_RE.fullmatch(line):
            continue
        usable_length = len(line) - (len(line) % 2)
        count += usable_length // 2
    return count


def non_hex_char_count(text: str) -> int:
    compact = re.sub(r"\s+", "", str(text))
    return sum(1 for char in compact if char not in "0123456789abcdefABCDEF")


def collect_reasons(text: str) -> set[str]:
    text = str(text)
    lower = text.lower()
    reasons: set[str] = set()

    for phrase in PROMPT_PHRASES:
        if phrase in lower:
            reasons.add(f"prompt_phrase:{phrase}")

    for phrase in ASSISTANT_PHRASES:
        if phrase in lower:
            reasons.add(f"assistant_phrase:{phrase}")

    if "```" in text:
        reasons.add("markdown_fence")

    if not text.strip():
        reasons.add("empty_content")

    valid_lines, invalid_lines, hex_chars = hex_line_stats(text)
    if valid_lines == 0:
        reasons.add("no_valid_hex_line")
    if invalid_lines > 0:
        reasons.add("invalid_hex_line")

    if non_hex_char_count(text) > 0:
        reasons.add("non_hex_character")

    if len(text.strip()) < 20:
        reasons.add("too_short:lt20")

    byte_count = byte_token_count(text)
    if byte_count == 0:
        reasons.add("no_bytes")
    elif byte_count <= 2:
        reasons.add("very_few_bytes")

    if SOURCE_LIKE_RE.search(text):
        reasons.add("source_like_content")

    if ASM_LIKE_RE.search(text):
        reasons.add("asm_like_content")

    return reasons


def severity_for(reasons: set[str]) -> str:
    high_prefixes = {
        "prompt_phrase",
        "assistant_phrase",
        "markdown_fence",
        "empty_content",
        "no_valid_hex_line",
        "invalid_hex_line",
        "non_hex_character",
        "no_bytes",
        "source_like_content",
        "asm_like_content",
    }
    if any(reason.split(":")[0] in high_prefixes for reason in reasons):
        return "high"
    return "medium"


def audit_binary(dataframe: pd.DataFrame) -> pd.DataFrame:
    records = []
    for _, row in dataframe.iterrows():
        reasons = collect_reasons(row[COLUMN])
        if not reasons:
            continue

        valid_lines, invalid_lines, hex_chars = hex_line_stats(row[COLUMN])
        preview = str(row[COLUMN]).replace("\n", " ")[:200]
        records.append(
            {
                ID_COL: row[ID_COL],
                LABEL_COL: row[LABEL_COL],
                "severity": severity_for(reasons),
                "reasons": "|".join(sorted(reasons)),
                "reason_count": len(reasons),
                "valid_hex_lines": valid_lines,
                "hex_chars": hex_chars,
                "byte_count": byte_token_count(row[COLUMN]),
                "preview": preview,
            }
        )
    return pd.DataFrame(records)


def print_summary(flagged: pd.DataFrame, total_rows: int) -> None:
    print(f"Toplam satır: {total_rows:,}")
    print(f"Flagged ID: {len(flagged):,} ({len(flagged) / total_rows:.2%})")
    print(f"  high: {(flagged['severity'] == 'high').sum():,}")
    print(f"  medium: {(flagged['severity'] == 'medium').sum():,}")
    print("\nDataset kırılımı:")
    print(flagged[LABEL_COL].value_counts().sort_index().to_string())

    reason_counts: dict[str, int] = {}
    for reasons in flagged["reasons"]:
        for reason in reasons.split("|"):
            reason_counts[reason] = reason_counts.get(reason, 0) + 1

    print("\nNeden sayıları:")
    for reason, count in sorted(reason_counts.items(), key=lambda item: -item[1]):
        print(f"  {reason}: {count:,}")


def main() -> None:
    dataframe = load_data()
    flagged = audit_binary(dataframe)

    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    output_path = REPORT_DIR / "binary_flagged_ids.csv"
    flagged.sort_values(["severity", LABEL_COL, ID_COL], ascending=[True, True, True]).to_csv(
        output_path, index=False
    )

    ids_path = REPORT_DIR / "binary_flagged_ids.txt"
    ids_path.write_text("\n".join(flagged[ID_COL].astype(str).tolist()) + "\n")

    print(f"Girdi: {INPUT_PATH.name} ({COLUMN})")
    print_summary(flagged, len(dataframe))
    print(f"\nKaydedildi:\n  {output_path}\n  {ids_path}")


if __name__ == "__main__":
    main()