entity.py

#!/usr/bin/env python3
"""Entity: a scoped wireless assessment and evidence collection CLI."""

from __future__ import annotations

import argparse
from collections import Counter, defaultdict
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
import html
import json
import os
from pathlib import Path
import platform
import re
import shlex
import shutil
import subprocess
import tempfile
from typing import Any, TypeVar


VERSION = "1.0"
T = TypeVar("T")

REQUIRED_TOOLS = (
    "airmon-ng",
    "airodump-ng",
    "aireplay-ng",
    "aircrack-ng",
    "bully",
    "crunch",
    "hcxdumptool",
    "mdk3",
    "reaver",
    "tshark",
    "wifite",
)
OPTIONAL_TOOLS = ("airgeddon", "xterm")

INTERFACE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,14}$")
BSSID_RE = re.compile(r"^(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$")

CHARSETS = {
    "1": "abcdefghijklmnopqrstuvwxyz",
    "2": "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
    "3": "0123456789",
    "4": "!#$%/=?{}[]-*:;",
    "5": "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ",
    "6": "abcdefghijklmnopqrstuvwxyz0123456789",
    "7": "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
    "8": "!#$%/=?{}[]-*:;0123456789",
    "9": "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
    "10": "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ!#$%/=?{}[]-*:;",
    "11": (
        "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
        "0123456789!#$%/=?{}[]-*:;"
    ),
}

BANNER = r"""
    |\     ____
    | \.-./ .-'
     \ _  _(
     | .)(./
     |   \(
     |     \   |
     |   vvv   |
     |  |__    |
    /      `-. |

       ENTITY
    dx73r | v1.0
"""

MENU = """
[1]  Start monitor mode
[2]  Stop monitor mode
[3]  Scan networks
[4]  Capture handshake
[5]  Audit with RockYou
[6]  Audit with generated candidates
[7]  Scan for WPS networks
[8]  WPS testing
[9]  Deauthentication test
[10] Beacon flood test
[11] TCP traffic summary
[12] PMKID capture
[13] Test a known WPS PIN
[14] Launch airgeddon
[15] Analyze PCAP and validate handshake
[16] Passive live deauthentication/rogue-AP detection
[c]  Run dependency check
[0]  Exit
"""


class ValidationError(ValueError):
    pass


class ToolNotFoundError(RuntimeError):
    pass


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def validate_interface(value: str) -> str:
    value = value.strip()
    if not INTERFACE_RE.fullmatch(value):
        raise ValidationError(
            "interface must be 1-15 safe Linux interface-name characters"
        )
    return value


def validate_bssid(value: str) -> str:
    value = value.strip()
    if not BSSID_RE.fullmatch(value):
        raise ValidationError("BSSID must look like AA:BB:CC:DD:EE:FF")
    return value.upper()


def bounded_integer(
    value: str, *, label: str, minimum: int, maximum: int
) -> int:
    try:
        number = int(value)
    except ValueError as exc:
        raise ValidationError(f"{label} must be a whole number") from exc
    if not minimum <= number <= maximum:
        raise ValidationError(f"{label} must be between {minimum} and {maximum}")
    return number


def validate_channel(value: str) -> int:
    return bounded_integer(value, label="channel", minimum=1, maximum=233)


def validate_packet_count(value: str) -> int:
    return bounded_integer(value, label="packet count", minimum=1, maximum=100_000)


def validate_password_length(value: str) -> int:
    return bounded_integer(value, label="password length", minimum=8, maximum=64)


def validate_output_path(value: str) -> str:
    value = value.strip()
    if not value or "\x00" in value:
        raise ValidationError("path cannot be empty or contain NUL bytes")
    return str(Path(value).expanduser().resolve(strict=False))


def validate_existing_file(value: str) -> str:
    path = Path(validate_output_path(value)).resolve()
    if not path.is_file():
        raise ValidationError(f"file does not exist: {path}")
    return str(path)


def validate_essid(value: str) -> str:
    value = value.strip()
    if not 1 <= len(value) <= 32:
        raise ValidationError("ESSID must contain 1-32 characters")
    if any(ord(character) < 32 for character in value):
        raise ValidationError("ESSID cannot contain control characters")
    return value


def validate_optional_essid(value: str) -> str:
    return validate_essid(value) if value.strip() else ""


def validate_optional_bssid(value: str) -> str:
    return validate_bssid(value) if value.strip() else ""


def validate_charset(value: str) -> str:
    if not 1 <= len(value) <= 256:
        raise ValidationError("character set must contain 1-256 characters")
    if value.startswith("-") or any(ord(character) < 32 for character in value):
        raise ValidationError("character set cannot begin with '-' or contain controls")
    return value


def validate_wps_pin(value: str) -> str:
    value = value.strip()
    if not value.isdigit() or len(value) not in {4, 8}:
        raise ValidationError("WPS PIN must contain exactly 4 or 8 digits")
    return value


def prompt(
    label: str,
    validator: Callable[[str], T],
    *,
    default: str | None = None,
) -> T:
    while True:
        suffix = f" [{default}]" if default else ""
        raw = input(f"{label}{suffix}: ").strip()
        if not raw and default is not None:
            raw = default
        try:
            return validator(raw)
        except ValidationError as exc:
            print(f"[!] {exc}")


def confirm(message: str) -> bool:
    return input(f"{message} [y/N]: ").strip().lower() in {"y", "yes"}


@dataclass(frozen=True)
class PreflightReport:
    system: str
    is_root: bool
    available: tuple[str, ...]
    missing_required: tuple[str, ...]
    missing_optional: tuple[str, ...]

    @property
    def supported_system(self) -> bool:
        return self.system == "Linux"

    @property
    def ready(self) -> bool:
        return self.supported_system and not self.missing_required


def check_environment() -> PreflightReport:
    available = tuple(
        tool
        for tool in (*REQUIRED_TOOLS, *OPTIONAL_TOOLS)
        if shutil.which(tool) is not None
    )
    getuid = getattr(os, "geteuid", None)
    return PreflightReport(
        system=platform.system(),
        is_root=bool(getuid and getuid() == 0),
        available=available,
        missing_required=tuple(
            tool for tool in REQUIRED_TOOLS if tool not in available
        ),
        missing_optional=tuple(
            tool for tool in OPTIONAL_TOOLS if tool not in available
        ),
    )


def print_preflight(report: PreflightReport) -> None:
    print(f"Operating system: {report.system}")
    print(f"Elevated privileges: {'yes' if report.is_root else 'no'}")
    print(
        "Missing required tools: " + ", ".join(report.missing_required)
        if report.missing_required
        else "Required tools: available"
    )
    if report.missing_optional:
        print("Missing optional tools: " + ", ".join(report.missing_optional))
    if not report.supported_system:
        print("Entity supports Linux; use --dry-run for previews on other systems.")


class EvidenceRecorder:
    """Append-only session evidence with JSON and HTML report export."""

    def __init__(self, directory: str | Path) -> None:
        requested_directory = Path(directory).expanduser().resolve()
        try:
            requested_directory.mkdir(parents=True, exist_ok=True)
            self.directory = requested_directory
        except OSError as exc:
            fallback_directory = Path(tempfile.gettempdir()) / "entity-evidence"
            try:
                fallback_directory.mkdir(parents=True, exist_ok=True)
            except OSError as fallback_exc:
                raise RuntimeError(
                    "Entity could not create either the requested evidence "
                    f"directory ({requested_directory}) or the fallback "
                    f"directory ({fallback_directory})."
                ) from fallback_exc
            self.directory = fallback_directory.resolve()
            print(
                f"[!] Could not create evidence directory {requested_directory}: "
                f"{exc}. Using {self.directory} instead."
            )
        self.session_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
        self.log_path = self.directory / f"entity-{self.session_id}.jsonl"
        self.record("session_started", version=VERSION)

    def record(self, event: str, **details: Any) -> None:
        entry = {
            "timestamp": utc_now(),
            "session_id": self.session_id,
            "event": event,
            "details": details,
        }
        with self.log_path.open("a", encoding="utf-8", newline="\n") as handle:
            handle.write(json.dumps(entry, sort_keys=True, default=str) + "\n")

    def export_report(self, report: dict[str, Any], label: str) -> tuple[Path, Path]:
        safe_label = re.sub(r"[^A-Za-z0-9_.-]+", "-", label).strip("-") or "report"
        stem = f"{safe_label}-{self.session_id}"
        json_path = self.directory / f"{stem}.json"
        html_path = self.directory / f"{stem}.html"
        json_path.write_text(
            json.dumps(report, indent=2, sort_keys=True, default=str) + "\n",
            encoding="utf-8",
        )
        html_path.write_text(self._html_report(report), encoding="utf-8")
        self.record(
            "report_exported",
            report_type=label,
            json_path=str(json_path),
            html_path=str(html_path),
        )
        return json_path, html_path

    @staticmethod
    def _html_report(report: dict[str, Any]) -> str:
        findings = report.get("findings", [])
        rows = "".join(
            "<tr>"
            f"<td>{html.escape(str(item.get('severity', 'info')))}</td>"
            f"<td>{html.escape(str(item.get('title', 'Finding')))}</td>"
            f"<td>{html.escape(str(item.get('detail', '')))}</td>"
            "</tr>"
            for item in findings
        )
        summary = html.escape(
            json.dumps(report.get("summary", {}), indent=2, default=str)
        )
        source = html.escape(str(report.get("source", "")))
        generated = html.escape(str(report.get("generated_at", "")))
        return f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<title>Entity assessment report</title>
<style>
body{{font-family:system-ui,sans-serif;max-width:1100px;margin:2rem auto;padding:0 1rem}}
table{{border-collapse:collapse;width:100%}}th,td{{border:1px solid #bbb;padding:.5rem}}
th{{background:#eee;text-align:left}}pre{{background:#f5f5f5;padding:1rem;overflow:auto}}
</style></head><body>
<h1>Entity assessment report</h1>
<p><strong>Source:</strong> {source}<br><strong>Generated:</strong> {generated}</p>
<h2>Findings</h2>
<table><thead><tr><th>Severity</th><th>Finding</th><th>Evidence</th></tr></thead>
<tbody>{rows or '<tr><td colspan="3">No findings</td></tr>'}</tbody></table>
<h2>Summary</h2><pre>{summary}</pre>
</body></html>
"""


class CommandRunner:
    """Execute argument lists without invoking a shell."""

    def __init__(
        self, *, dry_run: bool = False, evidence: EvidenceRecorder | None = None
    ) -> None:
        self.dry_run = dry_run
        self.evidence = evidence

    @staticmethod
    def normalize(command: Sequence[str]) -> list[str]:
        normalized = [str(part) for part in command]
        if not normalized or not normalized[0]:
            raise ValueError("command must contain an executable")
        if any("\x00" in part for part in normalized):
            raise ValueError("command arguments cannot contain NUL bytes")
        return normalized

    @staticmethod
    def redact(command: Sequence[str]) -> list[str]:
        redacted = list(command)
        for index, value in enumerate(redacted[:-1]):
            if value in {"-p", "--password", "--pin"}:
                redacted[index + 1] = "***"
        return redacted

    def run(self, command: Sequence[str]) -> int:
        normalized = self.normalize(command)
        displayed = self.redact(normalized)
        print(f"$ {shlex.join(displayed)}")
        if self.evidence:
            self.evidence.record("command_started", command=displayed)
        if self.dry_run:
            returncode = 0
        else:
            try:
                returncode = subprocess.run(normalized, check=False).returncode
            except FileNotFoundError as exc:
                raise ToolNotFoundError(normalized[0]) from exc
        if self.evidence:
            self.evidence.record(
                "command_finished", command=displayed, returncode=returncode
            )
        return returncode

    def run_pipeline(self, commands: Sequence[Sequence[str]]) -> int:
        normalized = [self.normalize(command) for command in commands]
        if not normalized:
            raise ValueError("pipeline cannot be empty")
        displayed = [self.redact(command) for command in normalized]
        print("$ " + " | ".join(shlex.join(command) for command in displayed))
        if self.evidence:
            self.evidence.record("pipeline_started", commands=displayed)
        if self.dry_run:
            return 0

        processes: list[subprocess.Popen[bytes]] = []
        previous_stdout = None
        try:
            for index, command in enumerate(normalized):
                process = subprocess.Popen(
                    command,
                    stdin=previous_stdout,
                    stdout=None if index == len(normalized) - 1 else subprocess.PIPE,
                )
                if previous_stdout is not None:
                    previous_stdout.close()
                previous_stdout = process.stdout
                processes.append(process)
            returncode = processes[-1].wait()
            for process in processes[:-1]:
                process.wait()
        except FileNotFoundError as exc:
            self.terminate(processes)
            raise ToolNotFoundError(normalized[len(processes)][0]) from exc
        except KeyboardInterrupt:
            self.terminate(processes)
            returncode = 130
        if self.evidence:
            self.evidence.record("pipeline_finished", returncode=returncode)
        return returncode

    def run_parallel(self, commands: Sequence[Sequence[str]]) -> list[int]:
        normalized = [self.normalize(command) for command in commands]
        if not normalized:
            raise ValueError("parallel command list cannot be empty")
        displayed = [self.redact(command) for command in normalized]
        for command in displayed:
            print(f"$ {shlex.join(command)}")
        if self.evidence:
            self.evidence.record("parallel_commands_started", commands=displayed)
        if self.dry_run:
            return [0 for _ in normalized]

        processes: list[subprocess.Popen[bytes]] = []
        try:
            processes = [subprocess.Popen(command) for command in normalized]
            returncodes = [process.wait() for process in processes]
        except FileNotFoundError as exc:
            self.terminate(processes)
            raise ToolNotFoundError(normalized[len(processes)][0]) from exc
        except KeyboardInterrupt:
            self.terminate(processes)
            returncodes = [130 for _ in normalized]
        if self.evidence:
            self.evidence.record(
                "parallel_commands_finished", returncodes=returncodes
            )
        return returncodes

    @staticmethod
    def terminate(processes: Sequence[subprocess.Popen[bytes]]) -> None:
        for process in processes:
            if process.poll() is None:
                process.terminate()
        for process in processes:
            if process.poll() is None:
                try:
                    process.wait(timeout=3)
                except subprocess.TimeoutExpired:
                    process.kill()


def import_pyshark() -> Any:
    try:
        import pyshark
    except ImportError as exc:
        raise RuntimeError(
            "pyshark is not installed; install the project with `python -m pip install .`"
        ) from exc
    return pyshark


def layer_names(packet: Any) -> list[str]:
    return [
        str(getattr(layer, "layer_name", "")).lower()
        for layer in getattr(packet, "layers", [])
    ]


def layer_value(layer: Any, *names: str) -> str:
    for name in names:
        value = getattr(layer, name, None)
        if value not in (None, ""):
            return str(value)
    fields = getattr(layer, "_all_fields", {})
    for key, value in fields.items():
        if any(str(key).endswith(name.replace("_", ".")) for name in names):
            return str(value)
    return ""


def parse_subtype(value: str) -> int | None:
    try:
        return int(value, 0)
    except (TypeError, ValueError):
        match = re.search(r"0x[0-9a-fA-F]+|\d+", str(value))
        return int(match.group(0), 0) if match else None


def analyze_packets(
    packets: Iterable[Any],
    *,
    source: str,
    expected_ssid: str = "",
    trusted_bssid: str = "",
) -> dict[str, Any]:
    protocols: Counter[str] = Counter()
    networks: dict[str, set[str]] = defaultdict(set)
    bssids: set[str] = set()
    handshake_messages: set[int] = set()
    packet_count = 0
    deauth_count = 0
    disassociation_count = 0
    eapol_count = 0

    for packet in packets:
        packet_count += 1
        protocols[str(getattr(packet, "highest_layer", "UNKNOWN"))] += 1
        names = layer_names(packet)
        if "eapol" in names or hasattr(packet, "eapol"):
            eapol_count += 1

        wlan = getattr(packet, "wlan", None)
        management = getattr(packet, "wlan_mgt", None)
        if wlan is not None:
            packet_bssid = layer_value(wlan, "bssid").split(",")[0].strip().upper()
            if BSSID_RE.fullmatch(packet_bssid):
                bssids.add(packet_bssid)
            subtype = parse_subtype(layer_value(wlan, "fc_type_subtype"))
            if subtype == 0x0C:
                deauth_count += 1
            elif subtype == 0x0A:
                disassociation_count += 1
            if management is not None:
                ssid = layer_value(management, "ssid").strip()
                if ssid and ssid not in {"<MISSING>", "<Malformed>"} and packet_bssid:
                    networks[ssid].add(packet_bssid)

        for layer in getattr(packet, "layers", []):
            if "eapol" not in str(getattr(layer, "layer_name", "")).lower():
                continue
            message = layer_value(layer, "keydes_msgnr", "msgnr")
            parsed = parse_subtype(message)
            if parsed in {1, 2, 3, 4}:
                handshake_messages.add(parsed)

    expected = expected_ssid.strip()
    trusted = trusted_bssid.upper().strip()
    findings: list[dict[str, str]] = []

    if deauth_count:
        severity = "medium" if deauth_count >= 20 else "low"
        findings.append(
            {
                "severity": severity,
                "title": "Deauthentication frames observed",
                "detail": f"{deauth_count} deauthentication frames were present.",
            }
        )
    if disassociation_count:
        findings.append(
            {
                "severity": "info",
                "title": "Disassociation frames observed",
                "detail": f"{disassociation_count} disassociation frames were present.",
            }
        )

    if expected:
        observed = networks.get(expected, set())
        unexpected = sorted(observed - ({trusted} if trusted else set()))
        if trusted and unexpected:
            findings.append(
                {
                    "severity": "medium",
                    "title": "Possible rogue access point",
                    "detail": (
                        f"SSID {expected!r} was advertised by unexpected BSSID(s): "
                        + ", ".join(unexpected)
                    ),
                }
            )
        elif not trusted and len(observed) > 1:
            findings.append(
                {
                    "severity": "info",
                    "title": "SSID advertised by multiple BSSIDs",
                    "detail": (
                        ", ".join(sorted(observed))
                        + "; this can be legitimate on mesh or enterprise networks."
                    ),
                }
            )

    if handshake_messages == {1, 2, 3, 4}:
        handshake_status = "complete-four-message-sequence-observed"
    elif handshake_messages:
        handshake_status = "partial-messages-" + "-".join(
            str(item) for item in sorted(handshake_messages)
        )
    elif eapol_count:
        handshake_status = "eapol-observed-message-order-unavailable"
    else:
        handshake_status = "not-observed"
    findings.append(
        {
            "severity": "info",
            "title": "Handshake validation",
            "detail": (
                f"Status: {handshake_status}; EAPOL frames: {eapol_count}. "
                "This is capture validation, not cryptographic proof."
            ),
        }
    )

    network_summary = {
        ssid: sorted(observed_bssids)
        for ssid, observed_bssids in sorted(networks.items())
    }
    return {
        "report_type": "wireless-capture-assessment",
        "generated_at": utc_now(),
        "source": source,
        "scope": {
            "expected_ssid": expected,
            "trusted_bssid": trusted,
        },
        "summary": {
            "packets": packet_count,
            "protocols": dict(protocols.most_common()),
            "bssids": sorted(bssids),
            "networks": network_summary,
            "deauthentication_frames": deauth_count,
            "disassociation_frames": disassociation_count,
            "eapol_frames": eapol_count,
            "handshake_messages": sorted(handshake_messages),
            "handshake_status": handshake_status,
        },
        "findings": findings,
    }


def print_analysis(report: dict[str, Any]) -> None:
    summary = report["summary"]
    print(f"Packets analyzed: {summary['packets']}")
    print(f"Networks observed: {len(summary['networks'])}")
    print(f"Deauthentication frames: {summary['deauthentication_frames']}")
    print(f"EAPOL frames: {summary['eapol_frames']}")
    print(f"Handshake status: {summary['handshake_status']}")
    for finding in report["findings"]:
        print(f"[{finding['severity'].upper()}] {finding['title']}: {finding['detail']}")


class EntityOperations:
    def __init__(self, runner: CommandRunner, evidence: EvidenceRecorder) -> None:
        self.runner = runner
        self.evidence = evidence

    def execute(self, operation: Callable[[], None]) -> None:
        try:
            operation()
        except ToolNotFoundError as exc:
            print(f"[!] Required tool not found: {exc}. Run `entity --check`.")
            self.evidence.record("tool_missing", tool=str(exc))
        except RuntimeError as exc:
            print(f"[!] {exc}")
            self.evidence.record("operation_error", error=str(exc))
        except KeyboardInterrupt:
            print("\n[!] Operation cancelled.")
            self.evidence.record("operation_cancelled")

    def start_monitor(self) -> None:
        iface = prompt("Interface", validate_interface, default="wlan0")
        self.runner.run(["airmon-ng", "start", iface])

    def stop_monitor(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        self.runner.run(["airmon-ng", "stop", iface])

    def scan_networks(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        print("[i] Press Ctrl+C to stop scanning.")
        self.runner.run(["airodump-ng", "-M", iface])

    def capture_handshake(self) -> None:
        if not confirm("Proceed with this disruptive assessment step"):
            return
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        target = prompt("Target BSSID", validate_bssid)
        channel = prompt("Target channel", validate_channel)
        path = prompt("Output path prefix", validate_output_path)
        packets = prompt("Deauthentication packet count", validate_packet_count)
        self.runner.run_parallel(
            [
                [
                    "airodump-ng",
                    "--bssid",
                    target,
                    "-c",
                    str(channel),
                    "-w",
                    path,
                    iface,
                ],
                [
                    "xterm",
                    "-e",
                    "aireplay-ng",
                    "-0",
                    str(packets),
                    "-a",
                    target,
                    iface,
                ],
            ]
        )

    def crack_with_rockyou(self) -> None:
        handshake = prompt("Handshake file", validate_existing_file)
        wordlist = Path("/usr/share/wordlists/rockyou.txt")
        if not wordlist.is_file() and not self.runner.dry_run:
            print("[!] rockyou.txt was not found; Entity will not alter system files.")
            return
        self.runner.run(["aircrack-ng", handshake, "-w", str(wordlist)])

    def crack_generated(self) -> None:
        network_name = prompt("ESSID", validate_essid)
        handshake = prompt("Handshake file", validate_existing_file)
        minimum = prompt("Minimum password length", validate_password_length)
        maximum = prompt("Maximum password length", validate_password_length)
        if minimum > maximum:
            print("[!] Minimum length cannot exceed maximum length.")
            return
        print(
            "Sets: 1 lower, 2 upper, 3 numeric, 4 symbols, 5 lower+upper,\n"
            "6 lower+numeric, 7 upper+numeric, 8 symbol+numeric,\n"
            "9 lower+upper+numeric, 10 lower+upper+symbol, 11 all, 12 custom"
        )
        choice = input("Choice: ").strip()
        charset = (
            prompt("Custom character set", validate_charset)
            if choice == "12"
            else CHARSETS.get(choice)
        )
        if not charset:
            print("[!] Unknown character-set choice.")
            return
        self.runner.run_pipeline(
            [
                ["crunch", str(minimum), str(maximum), charset],
                ["aircrack-ng", handshake, "-e", network_name, "-w-"],
            ]
        )

    def scan_wps(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        self.runner.run(["airodump-ng", "-M", "--wps", iface])

    def wps_attack(self) -> None:
        if not confirm("Proceed with this assessment step"):
            return
        print("1 Reaver\n2 Bully\n3 Wifite\n4 Reaver PixieWPS")
        choice = input("Choice: ").strip()
        if choice == "3":
            self.runner.run(["wifite"])
            return
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        target = prompt("Target BSSID", validate_bssid)
        if choice == "1":
            self.runner.run(["reaver", "-i", iface, "-b", target, "-vv"])
        elif choice == "2":
            channel = prompt("Target channel", validate_channel)
            self.runner.run(
                ["bully", "-b", target, "-c", str(channel), "--pixiewps", iface]
            )
        elif choice == "4":
            self.runner.run(["reaver", "-i", iface, "-b", target, "-K"])
        else:
            print("[!] Unknown WPS choice.")

    def deauthenticate(self) -> None:
        if not confirm("Proceed with this disruptive assessment step"):
            return
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        target = prompt("Target BSSID", validate_bssid)
        channel = prompt("Target channel", validate_channel)
        self.runner.run_parallel(
            [
                [
                    "airodump-ng",
                    "--bssid",
                    target,
                    "--channel",
                    str(channel),
                    iface,
                ],
                ["aireplay-ng", "--deauth", "0", "-a", target, iface],
            ]
        )

    def beacon_flood(self) -> None:
        if not confirm("Proceed with this disruptive assessment step"):
            return
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        self.runner.run(["mdk3", iface, "b", "-s", "250"])

    def tcp_summary(self) -> None:
        iface = prompt("Interface", validate_interface, default="wlan0")
        count = prompt("Packet count", validate_packet_count)
        if self.runner.dry_run:
            print(f"[dry-run] passive capture on {iface} for {count} packets")
            return
        pyshark = import_pyshark()
        capture = pyshark.LiveCapture(interface=iface)
        try:
            for packet in capture.sniff_continuously(packet_count=count):
                try:
                    protocol = packet.transport_layer
                    print(
                        f"{packet.sniff_time} IP {packet.ip.src}:"
                        f"{packet[protocol].srcport} <-> {packet.ip.dst}:"
                        f"{packet[protocol].dstport} ({protocol})"
                    )
                except (AttributeError, KeyError, TypeError):
                    continue
        finally:
            capture.close()

    def pmkid_capture(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        target = prompt("Target BSSID", validate_bssid)
        path = prompt("Output path", validate_output_path)
        self.runner.run(
            [
                "hcxdumptool",
                "-i",
                iface,
                "-o",
                path,
                "--enable_status=1",
                f"--filterlist_ap={target}",
                "--filtermode=2",
            ]
        )

    def wps_pin(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        target = prompt("Target BSSID", validate_bssid)
        channel = prompt("Target channel", validate_channel)
        pin = prompt("Known WPS PIN", validate_wps_pin)
        self.runner.run(
            [
                "reaver",
                "-i",
                iface,
                "-c",
                str(channel),
                "-b",
                target,
                "-p",
                pin,
                "-vv",
            ]
        )

    def airgeddon(self) -> None:
        self.runner.run(["airgeddon"])

    def analyze_pcap(self) -> None:
        capture_path = prompt("PCAP/capture file", validate_existing_file)
        expected_ssid = prompt(
            "Expected SSID (optional)", validate_optional_essid, default=""
        )
        trusted_bssid = prompt(
            "Trusted BSSID (optional)", validate_optional_bssid, default=""
        )
        if self.runner.dry_run:
            print(f"[dry-run] analyze capture {capture_path}")
            return
        pyshark = import_pyshark()
        capture = pyshark.FileCapture(capture_path, keep_packets=False)
        try:
            report = analyze_packets(
                capture,
                source=capture_path,
                expected_ssid=expected_ssid,
                trusted_bssid=trusted_bssid,
            )
        finally:
            capture.close()
        print_analysis(report)
        json_path, html_path = self.evidence.export_report(report, "pcap-analysis")
        print(f"JSON report: {json_path}\nHTML report: {html_path}")

    def passive_detection(self) -> None:
        iface = prompt("Monitor interface", validate_interface, default="wlan0mon")
        count = prompt("Packet count", validate_packet_count, default="1000")
        expected_ssid = prompt(
            "Expected SSID (optional)", validate_optional_essid, default=""
        )
        trusted_bssid = prompt(
            "Trusted BSSID (optional)", validate_optional_bssid, default=""
        )
        if self.runner.dry_run:
            print(f"[dry-run] passive detection on {iface} for {count} packets")
            return
        pyshark = import_pyshark()
        capture = pyshark.LiveCapture(interface=iface)
        try:
            report = analyze_packets(
                capture.sniff_continuously(packet_count=count),
                source=f"live:{iface}",
                expected_ssid=expected_ssid,
                trusted_bssid=trusted_bssid,
            )
        finally:
            capture.close()
        print_analysis(report)
        json_path, html_path = self.evidence.export_report(
            report, "passive-detection"
        )
        print(f"JSON report: {json_path}\nHTML report: {html_path}")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Wireless assessment and evidence collection CLI"
    )
    parser.add_argument("--check", action="store_true", help="check dependencies")
    parser.add_argument(
        "--dry-run", action="store_true", help="print commands without running them"
    )
    parser.add_argument("--no-banner", action="store_true")
    parser.add_argument(
        "--evidence-dir",
        default="entity-evidence",
        help="directory for JSONL evidence and JSON/HTML reports",
    )
    parser.add_argument("--version", action="version", version=f"Entity {VERSION}")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    preflight = check_environment()
    if args.check:
        print_preflight(preflight)
        return 0 if preflight.ready else 1
    if not preflight.supported_system and not args.dry_run:
        print_preflight(preflight)
        return 2

    if not args.no_banner:
        print(BANNER)
    print("Operate only within the assessment scope approved for this project.")
    evidence = EvidenceRecorder(args.evidence_dir)
    print(f"Evidence log: {evidence.log_path}")
    if preflight.missing_required:
        print("[!] Some external tools are missing; run `entity --check`.")

    operations = EntityOperations(
        CommandRunner(dry_run=args.dry_run, evidence=evidence), evidence
    )
    choices: dict[str, Callable[[], None]] = {
        "1": operations.start_monitor,
        "2": operations.stop_monitor,
        "3": operations.scan_networks,
        "4": operations.capture_handshake,
        "5": operations.crack_with_rockyou,
        "6": operations.crack_generated,
        "7": operations.scan_wps,
        "8": operations.wps_attack,
        "9": operations.deauthenticate,
        "10": operations.beacon_flood,
        "11": operations.tcp_summary,
        "12": operations.pmkid_capture,
        "13": operations.wps_pin,
        "14": operations.airgeddon,
        "15": operations.analyze_pcap,
        "16": operations.passive_detection,
    }

    while True:
        print(MENU)
        choice = input("entity > ").strip().lower()
        if choice == "0":
            evidence.record("session_finished")
            return 0
        if choice == "c":
            print_preflight(check_environment())
            continue
        operation = choices.get(choice)
        if operation is None:
            print("[!] Unknown option.")
            continue
        evidence.record("menu_option_selected", option=choice)
        operations.execute(operation)


if __name__ == "__main__":
    raise SystemExit(main())