knocksense.py

#!/usr/bin/env python3
# ============================================================
#  KnockSense v1.0 — Smart Port Knock Analyzer & Executor
#  By: Manjeet Thakur (darkdisaster08)
#  GitHub: https://github.com/darkdisaster08/knocksense
#  For authorized security testing only!
# ============================================================

import sys, os, argparse, time, socket, threading
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from modules.colors import *
from modules.recon import (full_tcp_scan, udp_scan, discover_hosts,
                            auto_detect_interface, quick_scan_ports,
                            analyze_pcap_protocols)
from modules.extractor import (extract_all, http_crawl, nc_extract,
                                tftp_download, ftp_extract)
from modules.decoder import try_all_decoders, decode_file, decode_url
from modules.knocker import (sequential_knock, simultaneous_knock,
                              nc_knock, smart_knock, smart_bruteforce,
                              silent_knock, CTF_SEQUENCES)
from modules.verifier import (take_snapshot, verify_knock, print_report,
                               suggest_next_steps, CHECK_PORTS)
from chain_engine import execute_chain, auto_chain, knock_simultaneous, quick_scan

# ─── AUTO MODE ────────────────────────────────────────────
def run_auto(args):
    """
    Fully autonomous port knock solver.
    Handles: static sequences, dynamic hint ports, PCAP chains, encoded sequences.
    """
    import json, shutil
    host = args.target
    verbose = not args.quiet
    fast = args.fast

    session = {
        'tcp_ports': {}, 'udp_ports': {},
        'downloaded_pcaps': [], 'opened_ports': set(),
        'knock_sequence': None, 'knock_protocol': 'tcp',
        'knock_mode': 'simultaneous', 'banners': {}
    }

    dl_dir = '/tmp/knocksense_downloads'
    if os.path.exists(dl_dir): shutil.rmtree(dl_dir)
    os.makedirs(dl_dir, exist_ok=True)

    # ── PHASE 1: RECON ─────────────────────────────────────
    if verbose: phase("PHASE 1 — RECONNAISSANCE")

    tcp_ports = full_tcp_scan(host, timeout=0.3 if fast else 0.5,
                               threads=500, verbose=verbose)
    session['tcp_ports'] = tcp_ports
    if verbose:
        if tcp_ports: success(f"TCP: {len(tcp_ports)} port(s) → {Y}{sorted(tcp_ports.keys())}{RESET}")
        else: info("TCP: No open ports")

    udp_ports = udp_scan(host, timeout=1.5, threads=20, verbose=verbose)
    session['udp_ports'] = udp_ports
    if verbose:
        if udp_ports: success(f"UDP: {len(udp_ports)} port(s) → {Y}{sorted(udp_ports.keys())}{RESET}")
        else: info("UDP: No open ports")

    http_ports = {p for p in tcp_ports if p in [80,443,8080,8443,8888]}

    # ── PHASE 2: EXTRACT ───────────────────────────────────
    if verbose: phase("PHASE 2 — DATA EXTRACTION")

    # Known service ports — skip nc extract on these
    SKIP_NC = {21,22,23,25,53,80,110,143,443,445,8080,8443,8888}

    # Helper: knock a sequence and immediately check + grab banners
    def knock_and_check(seq, label=''):
        from chain_engine import knock_simultaneous as _ks
        import socket as _sock_check
        # Take fast baseline — common ports only
        def fast_scan():
            open_ports = set()
            check = list(range(1, 10000)) + [10000,12992,20000,30000,40000,50000,60000]
            lock = __import__('threading').Lock()
            def _c(p):
                try:
                    s = _sock_check.socket(_sock_check.AF_INET, _sock_check.SOCK_STREAM)
                    s.settimeout(0.2)
                    if s.connect_ex((host, p)) == 0:
                        with lock: open_ports.add(p)
                    s.close()
                except: pass
            import threading
            ts = [threading.Thread(target=_c, args=(p,)) for p in check]
            for t in ts: t.start()
            for t in ts: t.join()
            return open_ports
        snap_before = fast_scan()
        # Try TCP simultaneous first (fastest)
        # Use nmap SYN knocks (like knockd client) — sequential with delay
        import subprocess as _sp
        for _kp in seq:
            _sp.run(['nmap', '-Pn', '--host-timeout', '200', '--max-retries', '0',
                     '-p', str(_kp), host], capture_output=True)
        # Fallback simultaneous connect knock
        _ks(host, {p: 'tcp' for p in seq})
        # Immediately try to connect to SSH and HTTP first
        import socket as _sock_imm
        for _imm_p in [22, 80, 443, 8080]:
            if _imm_p not in snap_before:
                try:
                    _si = _sock_imm.socket(_sock_imm.AF_INET, _sock_imm.SOCK_STREAM)
                    _si.settimeout(0.5)
                    if _si.connect_ex((host, _imm_p)) == 0:
                        _si.close()
                        found(f"{label}Port opened: {_imm_p} ✅")
                        session['opened_ports'].add(_imm_p)
                        session['knock_sequence'] = seq
                        return {_imm_p}
                    _si.close()
                except: pass
        # Check every 0.1s for 3 seconds
        for _ in range(30):
            time.sleep(0.1)
            snap_after = fast_scan()
            new_ports = snap_after - snap_before
            if new_ports:
                for p in sorted(new_ports):
                    found(f"{label}Port opened: {p} ✅")
                    # Grab banner immediately
                    try:
                        import socket as _s
                        s = _s.socket(_s.AF_INET, _s.SOCK_STREAM)
                        s.settimeout(1.5)
                        s.connect((host, p))
                        banner = s.recv(4096).decode('utf-8', errors='ignore').strip()
                        s.close()
                        if banner:
                            found(f"Banner [{p}]: {Y}{banner[:200]}{RESET}")
                            session['banners'][p] = banner
                    except: pass
                session['opened_ports'] |= new_ports
                session['knock_sequence'] = seq
                # Immediately check if opened port gives another sequence (chained dynamic)
                import json as _jc
                for _op in sorted(new_ports):
                    try:
                        import socket as _sc2
                        _s2 = _sc2.socket(_sc2.AF_INET, _sc2.SOCK_STREAM)
                        _s2.settimeout(1)
                        _s2.connect((host, _op))
                        _resp2 = _s2.recv(4096).decode('utf-8', errors='ignore').strip()
                        _s2.close()
                        if _resp2:
                            found(f"Response from {_op}: {Y}{_resp2[:100]}{RESET}")
                            session['banners'][_op] = _resp2
                            try:
                                _seq2 = [int(x) for x in _jc.loads(_resp2) if 1 <= int(x) <= 65535]
                                if len(_seq2) >= 2:
                                    found(f"Chained sequence: {Y}{' → '.join(map(str,_seq2))}{RESET}")
                                    _ks(host, {p: 'tcp' for p in _seq2})
                                    time.sleep(0.2)
                                    _snap3 = fast_scan()
                                    _new3 = _snap3 - snap_before
                                    if _new3:
                                        for _p3 in sorted(_new3):
                                            found(f"Chain opened: {_p3} ✅")
                                        new_ports |= _new3
                                        session['opened_ports'] |= _new3
                            except: pass
                    except: pass
                return new_ports
        # Also try UDP if TCP didn't work
        _ks(host, {p: 'udp' for p in seq})
        for _ in range(5):
            time.sleep(0.2)
            snap_after = _qsp(host, list(range(1,65536)), timeout=0.3, threads=500)
            new_ports = snap_after - snap_before
            if new_ports:
                for p in sorted(new_ports):
                    found(f"{label}Port opened: {p} ✅")
                    try:
                        import socket as _s
                        s = _s.socket(_s.AF_INET, _s.SOCK_STREAM)
                        s.settimeout(1.5)
                        s.connect((host, p))
                        banner = s.recv(4096).decode('utf-8', errors='ignore').strip()
                        s.close()
                        if banner:
                            found(f"Banner [{p}]: {Y}{banner[:200]}{RESET}")
                            session['banners'][p] = banner
                    except: pass
                session['opened_ports'] |= new_ports
                session['knock_sequence'] = seq
                return new_ports
        return set()

    # Step 1: Check unknown ports for dynamic hint sequences
    hint_ports = [p for p in tcp_ports if p not in SKIP_NC]
    hint_sequences = {}  # port -> sequence list
    for hp in hint_ports:
        resp = nc_extract(host, hp, timeout=3, verbose=verbose)
        if resp:
            try:
                plist = [int(x) for x in json.loads(resp.strip()) if 1 <= int(x) <= 65535]
                if len(plist) >= 2:
                    hint_sequences[hp] = plist
                    found(f"Hint port {hp}: {Y}{plist}{RESET}")
            except: pass

    # Step 2: Extract data (HTTP, TFTP, FTP)
    all_data, all_pcaps = extract_all(host, tcp_ports, udp_ports,
                                       verbose=verbose, download_dir=dl_dir)
    session['downloaded_pcaps'] = list(all_pcaps)
    # Also collect any NC responses from extract_all
    for src, content in all_data:
        if src.startswith('NC:'):
            try:
                port = int(src.split(':')[1])
                if port not in SKIP_NC:
                    plist = [int(x) for x in json.loads(content.strip()) if 1 <= int(x) <= 65535]
                    if len(plist) >= 2:
                        hint_sequences[port] = plist
            except: pass

    if verbose:
        success(f"Extracted {len(all_data)} source(s), {len(all_pcaps)} PCAP(s)")

    # ── PHASE 3: DECODE ────────────────────────────────────
    if verbose: phase("PHASE 3 — DECODE ENGINE")

    decode_results = []
    for source, content in all_data:
        if not content.strip(): continue
        results = try_all_decoders(content)
        if results:
            if verbose:
                info(f"Decoded from {Y}{source}{RESET}:")
                for method, decoded, ports in results[:2]:
                    found(f"  {method}: {' → '.join(map(str, ports))}")
            decode_results.extend(results)

    # ── PHASE 4: KNOCK ─────────────────────────────────────
    if verbose: phase("PHASE 4 — KNOCK ENGINE")

    baseline = take_snapshot(host, "before knock", verbose=verbose)
    print()
    opened = set()

    # Strategy 1: Dynamic hint sequences — get FRESH sequence right before knocking
    for hp, _ in hint_sequences.items():
        if verbose: info(f"Getting fresh sequence from port {hp}...")
        from itertools import permutations as _perms
        for attempt in range(5):
            fresh = nc_extract(host, hp, timeout=3, verbose=False)
            if not fresh: continue
            try:
                seq = [int(x) for x in json.loads(fresh.strip()) if 1 <= int(x) <= 65535]
                if len(seq) < 2: continue
                # Try all permutations of the sequence
                for perm in _perms(seq):
                    perm = list(perm)
                    if verbose: found(f"Dynamic sequence: {Y}{' → '.join(map(str, perm))}{RESET}")
                    new_ports = knock_and_check(perm, "Dynamic: ")
                    if new_ports:
                        opened |= new_ports
                        break
                if opened: break
            except: pass
            time.sleep(0.3)
        if opened: break

    # Strategy 2: PCAP sequences — build full chain
    if not opened and all_pcaps:
        if verbose: info(f"Analyzing {len(all_pcaps)} PCAP(s)...")
        import re as _re
        def pcap_key(f):
            m = _re.search(r'pcap([0-9]+)', os.path.basename(f))
            return int(m.group(1)) if m else 99
        all_sequences = []
        seen = set()
        for pcap_file in sorted(set(all_pcaps), key=pcap_key):
            seqs = analyze_pcap_protocols(pcap_file)
            for src, data in seqs.items():
                seq = tuple(data['ports'])
                if seq not in seen and len(seq) >= 1:
                    seen.add(seq)
                    mixed = data.get('mixed_ports', {})
                    UDP_HINTS = {22,53,69}
                    if not mixed or all(v=='tcp' for v in mixed.values()):
                        mixed = {p: 'udp' if p in UDP_HINTS else 'tcp' for p in seq}
                    all_sequences.append({'ports': list(seq), 'mixed_ports': mixed})
                    if verbose: found(f"PCAP sequence: {' → '.join(map(str, seq))}")

        if all_sequences:
            if verbose: info(f"Building {len(all_sequences)}-stage chain...")
            chain_result = execute_chain(host, all_sequences, baseline, verbose=verbose)
            chain_opened = chain_result[0] if isinstance(chain_result, tuple) else chain_result
            if chain_opened:
                opened |= chain_opened
                session['knock_sequence'] = all_sequences[0]['ports']

    # Strategy 3: Decoded text sequences
    if not opened and decode_results:
        if verbose: info(f"Trying {len(decode_results)} decoded sequence(s)...")
        for method, decoded, seq in decode_results:
            if verbose: info(f"Sequence from {M}{method}{RESET}: {Y}{' → '.join(map(str, seq))}{RESET}")
            new_ports = try_knock_all(host, seq, args, baseline, verbose)
            if new_ports:
                opened |= new_ports
                session['knock_sequence'] = seq
                break

    # Strategy 4: Smart bruteforce
    if not opened:
        if verbose: info("No sequence found — smart bruteforce...")
        def check_new():
            from modules.recon import quick_scan_ports as _qsp
            return _qsp(host, CHECK_PORTS, timeout=0.3, threads=300) - baseline
        seq, proto, mode, new_ports = smart_bruteforce(
            host, check_new, delay=args.delay, verbose=verbose, fast=fast)
        if seq and new_ports:
            opened |= new_ports
            session['knock_sequence'] = seq
            session['knock_protocol'] = proto
            session['knock_mode'] = mode

    session['opened_ports'] |= opened

    # ── PHASE 5: FOLLOW CHAIN ──────────────────────────────
    if verbose: phase("PHASE 5 — FOLLOWING THE CHAIN")

    from modules.extractor import _fetch_url, find_pcap_links, _download_file

    chain_stages = []
    if session.get('knock_sequence'):
        first_seq = session['knock_sequence']
        chain_stages.append({'ports': first_seq, 'mixed_ports': {p:'tcp' for p in first_seq}})

    current_opened = set(opened)
    all_opened = set(opened)
    visited_nc = set()
    chain_iter = 0

    while chain_iter < 8 and current_opened:
        chain_iter += 1
        new_pcap_files = []

        for port in sorted(current_opened):
            if chain_stages:
                knock_simultaneous(host, chain_stages[0]['mixed_ports'])
                time.sleep(0.3)

            # NC extract — get hints or dynamic sequences
            nc_key = f"{host}:{port}"
            if nc_key not in visited_nc:
                visited_nc.add(nc_key)
                response = nc_extract(host, port, verbose=verbose)
                if response:
                    # Dynamic list
                    try:
                        plist = [int(x) for x in json.loads(response.strip()) if 1 <= int(x) <= 65535]
                        if len(plist) >= 2:
                            found(f"Dynamic sequence from port {port}: {' → '.join(map(str, plist))}")
                            new_p = knock_and_check(plist, f"Chain[{port}]: ")
                            if new_p:
                                all_opened |= new_p
                                current_opened |= new_p
                    except: pass
                    # Path hint
                    if response.strip().startswith('/'):
                        hint_path = response.strip()
                        if verbose: found(f"Path hint: {Y}{hint_path}{RESET}")
                        for hport in sorted(http_ports | {80,8080}):
                            if chain_stages:
                                knock_simultaneous(host, chain_stages[0]['mixed_ports'])
                                time.sleep(0.2)
                            proto_str = 'https' if hport in [443,8443] else 'http'
                            url = f"{proto_str}://{host}:{hport}{hint_path}"
                            page = _fetch_url(url)
                            if page:
                                if verbose: step(f"Found: {Y}{url}{RESET}")
                                for pcap_url in find_pcap_links(page, url):
                                    pcap_name = pcap_url.split('/')[-1]
                                    save_path = os.path.join(dl_dir, pcap_name)
                                    counter = 1
                                    base = pcap_name.rsplit('.', 1)
                                    while os.path.exists(save_path):
                                        save_path = os.path.join(dl_dir, f"{base[0]}_{counter}.{base[1]}")
                                        counter += 1
                                    size = _download_file(pcap_url, save_path)
                                    if size > 0:
                                        found(f"Downloaded: {Y}{pcap_name}{RESET}")
                                        new_pcap_files.append(save_path)

            # HTTP crawl
            if port in [80,443,8080,8443,8888,9090]:
                ssl = port in [443,8443]
                if chain_stages:
                    knock_simultaneous(host, chain_stages[0]['mixed_ports'])
                    time.sleep(0.2)
                data, pcaps = http_crawl(host, port, ssl=ssl,
                                          verbose=verbose, download_dir=dl_dir)
                new_pcap_files.extend(pcaps)

        session['downloaded_pcaps'].extend(new_pcap_files)
        if not new_pcap_files: break

        import re as _re2
        def pcap_key2(f):
            m = _re2.search(r'pcap([0-9]+)', os.path.basename(f))
            return int(m.group(1)) if m else 99

        new_stages = []
        for pcap_file in sorted(new_pcap_files, key=pcap_key2):
            seqs = analyze_pcap_protocols(pcap_file)
            for src, data in seqs.items():
                seq = data['ports']
                mixed = data.get('mixed_ports', {})
                UDP_HINTS = {22,53,69}
                if not mixed or all(v=='tcp' for v in mixed.values()):
                    mixed = {p: 'udp' if p in UDP_HINTS else 'tcp' for p in seq}
                if verbose: found(f"New sequence: {' → '.join(map(str, seq))}")
                new_stages.append({'ports': seq, 'mixed_ports': mixed})

        if not new_stages: break
        chain_stages.extend(new_stages)
        if verbose: info(f"Executing {len(chain_stages)}-stage chain...")

        chain_result = execute_chain(host, chain_stages, baseline, verbose=verbose)
        chain_opened = chain_result[0] if isinstance(chain_result, tuple) else chain_result
        if chain_opened:
            all_opened |= chain_opened
            current_opened = chain_opened
            for p in chain_opened:
                if p in [80,443,8080,8443]: http_ports.add(p)
        else: break

    session['opened_ports'] = all_opened

    # ── PHASE 6: REPORT ────────────────────────────────────
    if verbose: phase("PHASE 6 — REPORT")
    print_report(host, session, verbose=verbose)

def run_recon(args):
    host = args.target
    verbose = not args.quiet
    if verbose: phase("RECON MODE")
    tcp = full_tcp_scan(host, verbose=verbose)
    udp = udp_scan(host, verbose=verbose)
    session = {
        'tcp_ports': tcp, 'udp_ports': udp,
        'downloaded_pcaps': [], 'opened_ports': set()
    }
    print_report(host, session, verbose=verbose)

# ─── DECODE MODE ──────────────────────────────────────────
def run_decode(args):
    host = args.target
    verbose = not args.quiet
    if verbose: phase("DECODE MODE")

    if args.decode:
        if verbose: info(f"Input: {Y}{args.decode[:80]}{RESET}")
        results = try_all_decoders(args.decode)
    elif args.file:
        results = decode_file(args.file)
    elif args.url:
        results = decode_url(args.url)
    else:
        error("Provide --decode, --file, or --url")
        return

    if not results:
        warn("No sequences found.")
        return

    success(f"Found {len(results)} result(s):")
    print()
    for method, decoded, ports in results:
        found(f"Method  : {method}")
        found(f"Decoded : {decoded[:100]}")
        found(f"Ports   : {' → '.join(map(str, ports))}")
        print()

    if not host:
        return

    method, decoded, sequence = results[0]
    info(f"Executing sequence from {M}{method}{RESET}...")
    baseline = take_snapshot(host, "before knock", verbose=verbose)
    print()

    new_ports = try_knock_all(host, sequence, args, baseline, verbose)
    if new_ports:
        suggest_next_steps(host, new_ports, verbose=verbose)

def try_knock_all(host, sequence, args, baseline, verbose,
                  mixed_ports=None, prereknock_fn=None):
    """Try all knock methods until one works."""
    methods = []

    if mixed_ports and any(v == 'udp' for v in mixed_ports.values()):
        methods.append(('Simultaneous MIXED',
            lambda: simultaneous_knock(host, sequence, 'tcp',
                                      mixed_ports=mixed_ports,
                                      verbose=verbose)))
        methods.append(('Sequential MIXED',
            lambda: sequential_knock(host, sequence, 'tcp',
                                    args.delay,
                                    mixed_ports=mixed_ports,
                                    verbose=verbose)))

    methods += [
        ('Simultaneous TCP',
         lambda: simultaneous_knock(host, sequence, 'tcp', verbose=verbose)),
        ('Sequential TCP',
         lambda: sequential_knock(host, sequence, 'tcp',
                                 args.delay, verbose=verbose)),
        ('Simultaneous UDP',
         lambda: simultaneous_knock(host, sequence, 'udp', verbose=verbose)),
        ('Sequential UDP',
         lambda: sequential_knock(host, sequence, 'udp',
                                 args.delay, verbose=verbose)),
        ('NC knock',
         lambda: nc_knock(host, sequence, verbose=verbose)),
    ]

    # Add heuristic mixed protocol variations
    from modules.knocker import try_mixed_heuristics
    for mixed in try_mixed_heuristics(host, sequence):
        label = "Mixed " + "+".join(
            f"{p}:{proto.upper()}" for p, proto in mixed.items()
        )
        # Capture mixed in closure
        def make_mixed_fn(m):
            return lambda: simultaneous_knock(host, sequence, 'tcp',
                                              mixed_ports=m, verbose=verbose)
        methods.append((label, make_mixed_fn(mixed)))

    for name, fn in methods:
        if verbose: info(f"Method: {M}{name}{RESET}")
        # Re-knock prerequisite sequence before each attempt
        if prereknock_fn:
            prereknock_fn()
            time.sleep(0.3)
        fn()
        opened = verify_knock(host, baseline, wait=1.5, verbose=verbose)
        if opened:
            return opened
        if verbose: warn(f"{name} — no change")
        baseline = take_snapshot(host, verbose=False)

    return set()

# ─── SEQUENCE MODE ────────────────────────────────────────
def run_sequence(args):
    host = args.target
    verbose = not args.quiet
    if verbose: phase("SEQUENCE MODE")

    try:
        sequence = [int(p.strip()) for p in args.sequence.split(',')]
    except ValueError:
        error("Invalid sequence. Use: --sequence 7000,8000,9000")
        return

    baseline = take_snapshot(host, "before knock", verbose=verbose)
    print()

    if args.simultaneous:
        simultaneous_knock(host, sequence,
                          protocol=args.protocol,
                          repeat=args.repeat,
                          verbose=verbose)
    elif args.nc:
        nc_knock(host, sequence, verbose=verbose)
    else:
        sequential_knock(host, sequence,
                        protocol=args.protocol,
                        delay=args.delay,
                        repeat=args.repeat,
                        verbose=verbose)

    opened = verify_knock(host, baseline, verbose=verbose)
    if opened:
        suggest_next_steps(host, opened, verbose=verbose)

# ─── BRUTEFORCE MODE ──────────────────────────────────────
def run_bruteforce(args):
    host = args.target
    verbose = not args.quiet
    fast = args.fast
    if verbose: phase("SMART BRUTEFORCE MODE")

    baseline = take_snapshot(host, "baseline", verbose=verbose)
    print()

    def check_new():
        current = quick_scan_ports(host, CHECK_PORTS,
                                   timeout=0.3, threads=300)
        return current - baseline

    seq, proto, mode, opened = smart_bruteforce(
        host, check_new, delay=args.delay,
        verbose=verbose, fast=fast
    )

    if seq:
        print()
        suggest_next_steps(host, opened, verbose=verbose)
        print()
        info(f"Re-run: {C}python3 knocksense.py -t {host} "
             f"--sequence {','.join(map(str, seq))} "
             f"--protocol {proto}{RESET}")

# ─── PCAP MODE ────────────────────────────────────────────
def run_pcap(args):
    host = args.target
    verbose = not args.quiet
    if verbose: phase("PCAP ANALYSIS MODE")

    if verbose: info(f"Analyzing: {Y}{args.pcap}{RESET}")

    sequences = analyze_pcap_protocols(args.pcap)

    if not sequences:
        warn("No knock sequences found in PCAP.")
        return

    print()
    success(f"Found {len(sequences)} source(s):")
    print()

    for src_ip, data in sequences.items():
        found(f"Source   : {src_ip}")
        found(f"Protocol : {data['protocol'].upper()}")
        found(f"Sequence : {' → '.join(map(str, data['ports']))}")
        if data.get('mixed_ports'):
            for port, proto in data['mixed_ports'].items():
                step(f"  Port {port}: {proto.upper()}")
        print()

    if not host:
        return

    # Find best sequence and execute
    for src_ip, data in sequences.items():
        seq = data['ports']
        mixed = data.get('mixed_ports', {})
        proto = data.get('protocol', 'tcp')

        if len(seq) < 1:
            continue

        baseline = take_snapshot(host, "before knock", verbose=verbose)
        print()

        opened = try_knock_all(
            host, seq, args, baseline, verbose,
            mixed_ports=mixed if proto == 'mixed' else None
        )

        if opened:
            suggest_next_steps(host, opened, verbose=verbose)
            break

# ─── SNIFF MODE ───────────────────────────────────────────
def run_sniff(args):
    host = args.target
    verbose = not args.quiet
    if verbose: phase("LIVE SNIFF MODE")

    try:
        from scapy.all import sniff, IP, TCP, UDP
    except ImportError:
        error("scapy not installed. Run: pip install scapy --break-system-packages")
        return

    if os.geteuid() != 0:
        error("Sniff mode requires root/sudo.")
        return

    iface = args.interface or auto_detect_interface(host)
    duration = args.duration or 30
    if verbose:
        info(f"Interface: {Y}{iface}{RESET} | Duration: {Y}{duration}s{RESET}")
        info("Waiting for knock patterns...")
        print()

    sequences = {}

    def pkt_handler(pkt):
        if IP in pkt and pkt[IP].dst == host:
            src = pkt[IP].src
            if TCP in pkt and pkt[TCP].flags == 'S':
                dport, proto = pkt[TCP].dport, 'tcp'
            elif UDP in pkt:
                dport, proto = pkt[UDP].dport, 'udp'
            else:
                return
            if src not in sequences:
                sequences[src] = {'ports': [], 'protocol': proto,
                                  'mixed_ports': {}}
            if dport not in sequences[src]['ports']:
                sequences[src]['ports'].append(dport)
                sequences[src]['mixed_ports'][dport] = proto
                if verbose:
                    print(f"  {G}[knock]{RESET} {src} → "
                          f"port {Y}{dport}{RESET} ({proto.upper()})")

    sniff(iface=iface, prn=pkt_handler,
          timeout=duration, filter=f"dst host {host}")

    if sequences:
        print()
        success("Patterns detected:")
        for src, data in sequences.items():
            found(f"{src} → {' → '.join(map(str, data['ports']))} "
                  f"({data['protocol'].upper()})")

        for src, data in sequences.items():
            if len(data['ports']) >= 1:
                baseline = take_snapshot(host, "before knock",
                                         verbose=verbose)
                mixed = data['mixed_ports']
                proto = data['protocol']
                opened = try_knock_all(
                    host, data['ports'], args, baseline, verbose,
                    mixed_ports=mixed if proto == 'mixed' else None
                )
                if opened:
                    suggest_next_steps(host, opened, verbose=verbose)
                break

# ─── DISCOVER MODE ────────────────────────────────────────
def run_discover(args):
    verbose = not args.quiet
    phase("NETWORK DISCOVERY MODE")

    live = discover_hosts(args.subnet, verbose=verbose)

    if not live:
        warn("No live hosts found.")
        return

    print()
    success(f"Found {len(live)} live host(s):")
    for i, h in enumerate(live):
        print(f"  {C}[{i+1}]{RESET} {Y}{h}{RESET}")
    print()

    if len(live) == 1:
        target = live[0]
        if verbose: info(f"Auto-selecting: {Y}{target}{RESET}")
    else:
        print(f"{C}[?]{RESET} Select target [1-{len(live)}]: ", end='')
        try:
            choice = int(input().strip()) - 1
            target = live[choice]
        except (ValueError, IndexError):
            error("Invalid selection.")
            return

    args.target = target
    run_auto(args)

# ─── ARGUMENT PARSER ──────────────────────────────────────
def parse_args():
    parser = argparse.ArgumentParser(
        description="KnockSense v1.0 — Smart Port Knock Analyzer & Executor",
        formatter_class=argparse.RawTextHelpFormatter,
        epilog=f"""
{C}Examples:{RESET}
  {G}Full auto{RESET}       python3 knocksense.py -t 10.0.0.6 --auto
  {G}Fast auto{RESET}       python3 knocksense.py -t 10.0.0.6 --auto --fast
  {G}Quiet auto{RESET}      python3 knocksense.py -t 10.0.0.6 --auto --quiet
  {G}Discover{RESET}        python3 knocksense.py --discover
  {G}PCAP{RESET}            python3 knocksense.py -t 10.0.0.6 --pcap capture.pcap
  {G}Decode string{RESET}   python3 knocksense.py -t 10.0.0.6 --decode "+++[->+..]"
  {G}Decode file{RESET}     python3 knocksense.py -t 10.0.0.6 --file config.txt
  {G}Decode URL{RESET}      python3 knocksense.py -t 10.0.0.6 --url http://t/hint
  {G}Sequence{RESET}        python3 knocksense.py -t 10.0.0.6 --sequence 7000,8000,9000
  {G}Simultaneous{RESET}    python3 knocksense.py -t 10.0.0.6 --sequence 7000,8000,9000 --simultaneous
  {G}NC knock{RESET}        python3 knocksense.py -t 10.0.0.6 --sequence 7000,8000,9000 --nc
  {G}Bruteforce{RESET}      python3 knocksense.py -t 10.0.0.6 --bruteforce
  {G}Recon only{RESET}      python3 knocksense.py -t 10.0.0.6 --recon
  {G}Sniff{RESET}           sudo python3 knocksense.py -t 10.0.0.6 --sniff
        """
    )

    parser.add_argument('--target', '-t', help='Target IP address')

    # Modes
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument('--auto', action='store_true',
                      help='Full auto: scan→extract→decode→knock→chain→report')
    mode.add_argument('--discover', action='store_true',
                      help='Discover hosts then auto-attack')
    mode.add_argument('--recon', action='store_true',
                      help='Recon only: TCP+UDP scan')
    mode.add_argument('--decode', '-d',
                      help='Decode string (Base64/Brainfuck/Morse/Hex...)')
    mode.add_argument('--file', '-f',
                      help='Decode contents of a file')
    mode.add_argument('--url', '-u',
                      help='Fetch URL and decode content')
    mode.add_argument('--sequence', '-s',
                      help='Execute known sequence: 7000,8000,9000')
    mode.add_argument('--bruteforce', '-b', action='store_true',
                      help='Smart bruteforce all sequences')
    mode.add_argument('--pcap', '-p',
                      help='Extract and execute sequence from PCAP')
    mode.add_argument('--sniff', action='store_true',
                      help='Live sniff mode (requires root)')

    # Speed & output
    parser.add_argument('--fast', action='store_true',
                        help='Fast mode: TCP only, simultaneous only, fewer checks')
    parser.add_argument('--thorough', action='store_true',
                        help='Thorough mode: try everything, never give up')
    parser.add_argument('--quiet', '-q', action='store_true',
                        help='Quiet: show key findings only')
    parser.add_argument('--verbose', '-v', action='store_true',
                        help='Verbose: show every step (default)')

    # Knock options
    parser.add_argument('--simultaneous', action='store_true',
                        help='Force simultaneous knock')
    parser.add_argument('--nc', action='store_true',
                        help='Use nc (netcat) for knocking')
    parser.add_argument('--protocol', choices=['tcp','udp'], default='tcp',
                        help='Protocol: tcp (default) or udp')
    parser.add_argument('--delay', type=float, default=0.3,
                        help='Delay between knocks (default: 0.3s)')
    parser.add_argument('--repeat', type=int, default=1,
                        help='Knock each port N times (default: 1)')

    # Discovery
    parser.add_argument('--subnet',
                        help='Subnet for discovery (auto-detected if not set)')

    # Sniff
    parser.add_argument('--interface', '--iface', '-i', dest='interface',
                        help='Interface for sniff (auto-detected)')
    parser.add_argument('--duration', type=int, default=30,
                        help='Sniff duration in seconds (default: 30)')

    return parser.parse_args()

# ─── MAIN ─────────────────────────────────────────────────
def main():
    print(BANNER)
    args = parse_args()

    if not args.discover and not args.target:
        error("--target / -t required (except for --discover)")
        sys.exit(1)

    if args.target:
        print(f"  {DIM}Target   : {args.target}{RESET}")

    mode_str = 'FAST' if args.fast else ('THOROUGH' if args.thorough else 'NORMAL')
    out_str = 'QUIET' if args.quiet else 'VERBOSE'
    print(f"  {DIM}Mode     : {mode_str} | Output: {out_str}{RESET}")
    print(f"  {DIM}Delay    : {args.delay}s | Repeat: {args.repeat}x{RESET}")
    print()
    print(f"  {R}For authorized security testing only!{RESET}")
    print()

    if args.auto:           run_auto(args)
    elif args.discover:     run_discover(args)
    elif args.recon:        run_recon(args)
    elif args.decode:       run_decode(args)
    elif args.file:         run_decode(args)
    elif args.url:          run_decode(args)
    elif args.sequence:     run_sequence(args)
    elif args.bruteforce:   run_bruteforce(args)
    elif args.pcap:         run_pcap(args)
    elif args.sniff:        run_sniff(args)

if __name__ == "__main__":
    main()