Detect & Break Steganography

import numpy as np
from PIL import Image
from scipy import stats
import matplotlib.pyplot as plt
import sys

def get_pixel_values(image_path: str, channel: int = 0) -> np.ndarray:
    """Extract pixel values for a single channel (0=R, 1=G, 2=B)."""
    img = Image.open(image_path).convert('RGB')
    arr = np.array(img)
    return arr[:, :, channel].flatten()

def chi_square_lsb(values: np.ndarray) -> tuple[float, float]:
    """
    Perform the chi-square attack on a 1-D array of pixel values.
    Returns (chi2_statistic, p_value).
    High p-value (close to 1.0) suggests steganography.
    """
    # Count occurrences of each value 0..255
    counts = np.bincount(values, minlength=256)

    # Build PoV (Pair of Values) lists
    # observed: actual counts of each pair member
    # expected: mean of the pair (what LSB embedding produces)
    observed = []
    expected = []

    for k in range(128):
        n0 = counts[2 * k]       # count of value 2k
        n1 = counts[2 * k + 1]  # count of value 2k+1
        total = n0 + n1
        if total == 0:
            continue
        mean = total / 2.0
        observed.extend([n0, n1])
        expected.extend([mean, mean])

    observed = np.array(observed, dtype=float)
    expected = np.array(expected, dtype=float)

    # Remove pairs where expected == 0 (avoids division by zero)
    mask = expected > 0
    chi2 = np.sum((observed[mask] - expected[mask]) ** 2 / expected[mask])
    dof = mask.sum() - 1
    p_value = 1 - stats.chi2.cdf(chi2, dof)

    return chi2, p_value

def sliding_window_chi2(values: np.ndarray, window: int = 512) -> np.ndarray:
    """
    Run chi-square over sliding windows of pixel values.
    Returns array of p-values — shows where in the image embedding starts.
    Crucial for visualising partial embedding (not all images are fully embedded).
    """
    p_values = []
    for i in range(0, len(values) - window, window // 2):
        segment = values[i:i + window]
        _, p = chi_square_lsb(segment)
        p_values.append(p)
    return np.array(p_values)

def analyse_and_plot(clean_path: str, stego_path: str):
    """Full analysis with side-by-side sliding window plots."""
    fig, axes = plt.subplots(2, 2, figsize=(14, 8))
    fig.patch.set_facecolor('#0d1117')

    for ax in axes.flat:
        ax.set_facecolor('#161b22')
        ax.tick_params(colors='#8b949e')
        for spine in ax.spines.values(): spine.set_color('#30363d')

    for idx, (path, label) in enumerate([(clean_path, 'CLEAN'),
                                             (stego_path, 'STEGO')]):
        values = get_pixel_values(path, channel=0)  # Red channel
        chi2_global, p_global = chi_square_lsb(values)
        p_values_sliding = sliding_window_chi2(values)

        # Top row: value pair distribution
        counts = np.bincount(values, minlength=256)
        pairs = [(counts[2*k], counts[2*k+1]) for k in range(64)]
        evens = [p[0] for p in pairs]
        odds  = [p[1] for p in pairs]
        x = range(64)

        axes[0][idx].bar(x, evens, width=0.4, label='2k', color='#00d8a0', alpha=0.7)
        axes[0][idx].bar([i+0.4 for i in x], odds, width=0.4, label='2k+1',
                           color='#e05252', alpha=0.7)
        axes[0][idx].set_title(f'{label} — Pair Counts (lower 64 pairs)\n'
                                  f'χ²={chi2_global:.1f}  p={p_global:.4f}',
                                  color='#e6edf3', fontsize=10)
        axes[0][idx].legend(facecolor='#21262d', labelcolor='#8b949e', fontsize=8)

        # Bottom row: sliding window p-value chart
        color = '#e05252' if idx == 1 else '#00d8a0'
        axes[1][idx].plot(p_values_sliding, color=color, linewidth=1.5)
        axes[1][idx].axhline(0.95, color='#d29922', linestyle='--',
                                alpha=0.6, label='p=0.95 threshold')
        axes[1][idx].set_ylim(0, 1)
        axes[1][idx].set_title(f'{label} — Sliding Window p-values',
                                  color='#e6edf3', fontsize=10)
        axes[1][idx].set_xlabel('Window index', color='#8b949e')
        axes[1][idx].set_ylabel('p-value', color='#8b949e')
        axes[1][idx].legend(facecolor='#21262d', labelcolor='#8b949e', fontsize=8)

    plt.tight_layout(pad=2.0)
    plt.savefig('chi_square_analysis.png', dpi=150, bbox_inches='tight',
                facecolor='#0d1117')
    plt.show()
    print("\nPlot saved: chi_square_analysis.png")

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print("Usage: python chi_square_attack.py clean.png stego.png")
        sys.exit(1)
    analyse_and_plot(sys.argv[1], sys.argv[2])

import jpegio as jio
import numpy as np
import matplotlib.pyplot as plt
from collections import Counter
import sys

def extract_dct_coefficients(jpeg_path: str) -> np.ndarray:
    """
    Read raw quantised DCT coefficients directly from JPEG file.
    Returns a flattened array of all AC coefficients (excludes DC at position [0,0]).
    """
    struct = jio.read(jpeg_path)
    coeffs = []
    for channel_coeffs in struct.coef_arrays:
        h, w = channel_coeffs.shape
        # Process each 8x8 block
        for row in range(0, h, 8):
            for col in range(0, w, 8):
                block = channel_coeffs[row:row+8, col:col+8]
                # Flatten block, skip DC coefficient at [0,0]
                ac = block.flatten()[1:]
                coeffs.extend(ac.tolist())
    return np.array(coeffs)

def pov_histogram(coeffs: np.ndarray, rng: int = 10) -> dict:
    """
    Build Pair of Values (PoV) histogram of DCT coefficients.
    For JSteg detection: compare frequency of +k vs -k.
    Natural images have |f(+k)| ≈ |f(-k)| but not exactly equal.
    JSteg embedding equalises them — specifically at k=1 and k=2.
    """
    counts = Counter(coeffs.tolist())
    pov = {}
    for k in range(1, rng + 1):
        pov[k] = {
            'positive': counts.get(k, 0),
            'negative': counts.get(-k, 0),
            'ratio': counts.get(k, 1) / max(counts.get(-k, 1), 1)
        }
    return pov

def jsteg_detection_score(coeffs: np.ndarray) -> float:
    """
    Simple JSteg detection heuristic.
    Returns score 0..1 where >0.7 suggests JSteg embedding.
    Based on the deviation of f(1)/f(-1) from expected natural ratio.
    Natural images: ratio typically 0.95-1.05
    JSteg images: ratio approaches 1.00 (equalised)
    """
    counts = Counter(coeffs.tolist())
    f_pos1 = counts.get(1, 0)
    f_neg1 = counts.get(-1, 0)
    f_pos2 = counts.get(2, 0)
    f_neg2 = counts.get(-2, 0)

    if f_neg1 == 0 or f_neg2 == 0:
        return 0.0

    ratio1 = f_pos1 / f_neg1
    ratio2 = f_pos2 / f_neg2

    # Natural images: these ratios have specific biases
    # JSteg drives them toward 1.0 independently
    # Score based on how close both ratios are to 1.0
    deviation1 = abs(ratio1 - 1.0)
    deviation2 = abs(ratio2 - 1.0)

    # Lower deviation = more suspicious
    score = 1.0 - min((deviation1 + deviation2) / 2.0, 1.0)
    return round(score, 4)

def analyse_jpeg(path: str):
    """Full DCT analysis with visualisation."""
    print(f"\n{'='*50}")
    print(f"Analysing: {path}")
    print(f"{'='*50}")

    coeffs = extract_dct_coefficients(path)
    pov = pov_histogram(coeffs, rng=12)
    score = jsteg_detection_score(coeffs)

    print(f"Total AC coefficients analysed: {len(coeffs):,}")
    print(f"\nPoV Histogram (|k| vs |-k| counts):")
    print(f"{'k':>4}  {'f(+k)':>8}  {'f(-k)':>8}  {'ratio':>8}")
    print("-" * 36)
    for k, v in pov.items():
        flag = " ◄ suspicious" if abs(v['ratio']-1.0) < 0.02 and k <= 3 else ""
        print(f"{k:>4}  {v['positive']:>8,}  {v['negative']:>8,}  {v['ratio']:>8.4f}{flag}")

    print(f"\nJSteg detection score: {score:.4f}")
    if score > 0.85:
        print("→ HIGH confidence: JSteg steganography likely present")
    elif score > 0.65:
        print("→ MODERATE confidence: anomalies detected, further analysis needed")
    else:
        print("→ LOW confidence: image likely clean")

    # Plot DCT coefficient distribution
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
    fig.patch.set_facecolor('#0d1117')

    for ax in [ax1, ax2]:
        ax.set_facecolor('#161b22')
        ax.tick_params(colors='#8b949e')
        for s in ax.spines.values(): s.set_color('#30363d')

    # Left: overall DCT histogram (trimmed to ±30)
    trimmed = coeffs[(abs(coeffs) <= 30)]
    ax1.hist(trimmed, bins=range(-30, 31), color='#00d8a0', alpha=0.8, edgecolor='none')
    ax1.set_title('DCT Coefficient Distribution (±30)', color='#e6edf3')
    ax1.set_xlabel('DCT coefficient value', color='#8b949e')

    # Right: PoV bar chart for k=1..8
    ks = list(pov.keys())[:8]
    pos_counts = [pov[k]['positive'] for k in ks]
    neg_counts = [pov[k]['negative'] for k in ks]
    x = np.arange(len(ks))
    ax2.bar(x - 0.2, pos_counts, 0.4, label='f(+k)', color='#00d8a0', alpha=0.8)
    ax2.bar(x + 0.2, neg_counts, 0.4, label='f(-k)', color='#e05252', alpha=0.8)
    ax2.set_xticks(x); ax2.set_xticklabels([f'k={k}' for k in ks])
    ax2.set_title('PoV Histogram: f(+k) vs f(-k)', color='#e6edf3')
    ax2.legend(facecolor='#21262d', labelcolor='#8b949e')

    plt.tight_layout()
    outname = path.replace('.jpg', '_dct.png').replace('.jpeg', '_dct.png')
    plt.savefig(outname, dpi=150, bbox_inches='tight', facecolor='#0d1117')
    print(f"Plot saved: {outname}")

if __name__ == '__main__':
    for arg in sys.argv[1:]:
        analyse_jpeg(arg)

"""
Covert channel: encodes a message in the IP ID field.
Run as root. Use only in isolated lab environments.

Protocol:
  - Each packet carries 1 byte of the message in the IP ID field (low 8 bits)
  - High byte of IP ID is used as a sequence number (0-255)
  - Termination: IP ID = 0xFFFF signals end of message
"""
from scapy.all import IP, ICMP, send, conf
import time
import sys

conf.verb = 0  # suppress Scapy output

def encode_message(message: str, dst: str, delay: float = 0.1):
    """
    Encode a string message in IP ID fields of ICMP packets.
    Each byte of the message becomes the low byte of the IP ID.
    """
    encoded = message.encode('utf-8')
    print(f"Sending {len(encoded)} bytes to {dst}...")

    for seq, byte_val in enumerate(encoded):
        # Encode: high byte = sequence number, low byte = message byte
        ip_id = ((seq % 256) << 8) | byte_val
        pkt = IP(dst=dst, id=ip_id) / ICMP()
        send(pkt)
        time.sleep(delay)

    # Termination packet
    send(IP(dst=dst, id=0xFFFF) / ICMP())
    print("Transmission complete. Termination packet sent.")

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print("Usage: python3 covert_sender.py <dst_ip> <message>")
        sys.exit(1)
    encode_message(sys.argv[2], sys.argv[1])

"""
Receiver: reconstructs the message from IP ID fields.
Run as root on the destination machine.
"""
from scapy.all import sniff, IP, ICMP
import sys

message_bytes = {}

def process_packet(pkt):
    if IP in pkt and ICMP in pkt:
        ip_id = pkt[IP].id

        if ip_id == 0xFFFF:
            # Termination — reconstruct message
            if message_bytes:
                sorted_bytes = [message_bytes[k] for k in sorted(message_bytes)]
                msg = bytes(sorted_bytes).decode('utf-8', errors='replace')
                print(f"\n[+] Covert message received ({len(message_bytes)} bytes):")
                print(f"    {msg}")
                message_bytes.clear()
            return

        seq = (ip_id >> 8) & 0xFF     # high byte = sequence
        byte_val = ip_id & 0xFF        # low byte = message byte
        message_bytes[seq] = byte_val
        print(f"[>] Packet {seq}: byte={byte_val} ('{chr(byte_val) if 32 <= byte_val < 127 else '?'}')",
              end='\r')

print("[*] Listening for covert channel packets (Ctrl+C to stop)...")
sniff(filter="icmp", prn=process_packet, store=0)

"""
Detect IP ID-based covert channels in a PCAP file.
Analyses IP ID entropy and sequential patterns.
"""
from scapy.all import rdpcap, IP
import numpy as np
from collections import defaultdict
import sys

def analyse_ip_ids(pcap_path: str):
    pkts = rdpcap(pcap_path)
    ip_ids_by_src = defaultdict(list)

    for pkt in pkts:
        if IP in pkt:
            ip_ids_by_src[pkt[IP].src].append(pkt[IP].id)

    print(f"{'Source IP':<20} {'Pkts':>5} {'Entropy':>8} {'Max jump':>10} {'Verdict'}")
    print("-"*65)

    for src, ids in ip_ids_by_src.items():
        if len(ids) < 5:
            continue

        ids_arr = np.array(ids)
        # Entropy of IP ID values (higher = more random = suspicious)
        _, cnts = np.unique(ids_arr, return_counts=True)
        probs = cnts / cnts.sum()
        entropy = -np.sum(probs * np.log2(probs + 1e-10))

        # Max jump between consecutive IDs
        diffs = np.abs(np.diff(ids_arr.astype(int)))
        max_jump = diffs.max() if len(diffs) > 0 else 0

        verdict = 'SUSPICIOUS' if entropy > 6.0 and max_jump > 500 else 'normal'
        print(f"{src:<20} {len(ids):>5} {entropy:>8.3f} {max_jump:>10} {verdict}")

if __name__ == '__main__':
    analyse_ip_ids(sys.argv[1])

"""
Automated PDF forensics: scans for steganographic indicators.
Outputs a structured report of suspicious findings.
"""
import re
import sys
import zlib
import base64
from pathlib import Path

def read_pdf_raw(path: str) -> bytes:
    with open(path, 'rb') as f:
        return f.read()

def count_eof_markers(data: bytes) -> int:
    """Count %%EOF markers — multiple EOFs = incremental updates."""
    return data.count(b'%%EOF')

def find_white_text_indicators(data: bytes) -> list:
    """Find PDF colour operators that set white fill (potential invisible text)."""
    findings = []
    # Decompress FlateDecode streams first
    streams = re.findall(rb'stream\r?\n(.*?)\r?\nendstream', data, re.DOTALL)
    for i, stream in enumerate(streams):
        try:
            decoded = zlib.decompress(stream)
        except:
            decoded = stream
        # "1 1 1 rg" = white RGB fill in PDF content stream
        if b'1 1 1 rg' in decoded or b'1 1 1 RG' in decoded:
            findings.append(f"Stream {i}: white fill colour operator found")
        if b'0 0 0 0 k' in decoded:
            findings.append(f"Stream {i}: white CMYK fill found")
    return findings

def find_suspicious_keywords(data: bytes) -> dict:
    """Count suspicious PDF keywords."""
    keywords = [
        b'/EmbeddedFile', b'/ObjStm', b'/JavaScript',
        b'/JS', b'/OCG', b'/XFA', b'/AA', b'/OpenAction'
    ]
    return {kw.decode(): data.count(kw) for kw in keywords}

def find_base64_in_metadata(data: bytes) -> list:
    """Look for base64-encoded strings in Info dictionary fields."""
    findings = []
    # Rough regex for Info dict values that look like base64
    matches = re.findall(rbr'\(([A-Za-z0-9+/]{20,}={0,2})\)', data)
    for m in matches:
        try:
            decoded = base64.b64decode(m)
            if all(32 <= b < 127 for b in decoded[:20]):
                findings.append(f"Possible base64 payload: {decoded[:60]}...")
        except:
            pass
    return findings

def analyse_pdf(path: str):
    """Run all checks and print a forensic report."""
    data = read_pdf_raw(path)
    print(f"\n{'='*55}")
    print(f"PDF FORENSIC REPORT: {Path(path).name}")
    print(f"{'='*55}")
    print(f"File size    : {len(data):,} bytes")
    print(f"%%EOF markers: {count_eof_markers(data)}",
          "← MULTIPLE (incremental updates present)"
          if count_eof_markers(data) > 1 else "")

    kw = find_suspicious_keywords(data)
    print("\nSuspicious keyword counts:")
    for k, v in kw.items():
        flag = " ◄ FLAGGED" if v > 0 else ""
        print(f"  {k:<20} {v:>3}{flag}")

    white = find_white_text_indicators(data)
    if white:
        print(f"\nWhite text indicators found:")
        for w in white: print(f"  ◄ {w}")
    else:
        print("\nNo white text indicators found.")

    b64 = find_base64_in_metadata(data)
    if b64:
        print(f"\nPossible base64 payloads in string fields:")
        for b in b64[:5]: print(f"  ◄ {b}")

if __name__ == '__main__':
    analyse_pdf(sys.argv[1])

"""
Video steganography detector.
Analyses LSB entropy and frame-difference metrics per frame.
Plots anomalies to identify stego frames.

Install: pip install opencv-python numpy matplotlib --break-system-packages --break-system-packages
"""
import cv2
import numpy as np
import matplotlib.pyplot as plt
from scipy.stats import entropy as scipy_entropy
import sys

def lsb_entropy(frame: np.ndarray) -> float:
    """
    Compute Shannon entropy of the LSB plane of a frame.
    Clean frames: ~1.0 (near-random LSBs from natural image detail).
    Stego frames: approaches 1.0 very precisely (truly random embedded bits).
    The signature is *consistency* — stego frames cluster tightly at high entropy.
    """
    lsbs = frame & 1  # extract least significant bits
    lsb_flat = lsbs.flatten()
    ones = np.sum(lsb_flat)
    zeros = len(lsb_flat) - ones
    total = len(lsb_flat)
    if total == 0: return 0.0
    p1 = ones / total
    p0 = zeros / total
    if p0 == 0 or p1 == 0: return 0.0
    return -p0 * np.log2(p0) - p1 * np.log2(p1)

def frame_difference(f1: np.ndarray, f2: np.ndarray) -> float:
    """Mean absolute difference between consecutive frames (motion metric)."""
    return np.mean(np.abs(f1.astype(int) - f2.astype(int)))

def analyse_video(video_path: str, sample_rate: int = 1):
    """
    Analyse video for steganographic anomalies.
    sample_rate: analyse every Nth frame (1 = every frame, 5 = every 5th).
    """
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        print(f"Error: cannot open {video_path}")
        return

    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    print(f"Video: {video_path}")
    print(f"Frames: {total_frames}  FPS: {fps:.1f}  Duration: {total_frames/fps:.1f}s")
    print(f"Analysing every {sample_rate} frame(s)...")

    entropies = []
    diffs = []
    frame_indices = []
    prev_frame = None
    frame_num = 0

    while True:
        ret, frame = cap.read()
        if not ret: break

        if frame_num % sample_rate == 0:
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            ent = lsb_entropy(gray)
            entropies.append(ent)
            frame_indices.append(frame_num)

            if prev_frame is not None:
                diff = frame_difference(gray, prev_frame)
                diffs.append(diff)
            else:
                diffs.append(0)
            prev_frame = gray

        frame_num += 1
        if frame_num % 100 == 0:
            print(f"  Processed {frame_num}/{total_frames} frames...", end='\r')

    cap.release()
    print(f"\nComplete. Analysed {len(entropies)} frames.")

    # Detect anomalous frames: entropy > mean + 2*std
    ent_arr = np.array(entropies)
    threshold = ent_arr.mean() + 2 * ent_arr.std()
    flagged = [frame_indices[i] for i, e in enumerate(entropies) if e > threshold]

    print(f"\nEntropy stats: mean={ent_arr.mean():.4f}  std={ent_arr.std():.4f}")
    print(f"Detection threshold: {threshold:.4f}")
    if flagged:
        print(f"Flagged frames: {flagged[:20]}{'...' if len(flagged)>20 else ''}")
        print(f"Total flagged: {len(flagged)} of {len(entropies)}")
    else:
        print("No anomalous frames detected.")

    # Plot
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 6), sharex=True)
    fig.patch.set_facecolor('#0d1117')
    for ax in [ax1, ax2]:
        ax.set_facecolor('#161b22')
        ax.tick_params(colors='#8b949e')
        for s in ax.spines.values(): s.set_color('#30363d')

    ax1.plot(frame_indices, entropies, color='#00d8a0', linewidth=0.8, label='LSB entropy')
    ax1.axhline(threshold, color='#e05252', linestyle='--', alpha=0.7, label=f'threshold ({threshold:.3f})')
    for f in flagged:
        ax1.axvline(f, color='#d29922', alpha=0.3, linewidth=1)
    ax1.set_ylabel('LSB entropy', color='#8b949e')
    ax1.set_title(f'Video Steganalysis: {video_path}', color='#e6edf3')
    ax1.legend(facecolor='#21262d', labelcolor='#8b949e', fontsize=9)

    ax2.plot(frame_indices, diffs, color='#58a6ff', linewidth=0.8, label='Frame difference')
    ax2.set_ylabel('Frame difference', color='#8b949e')
    ax2.set_xlabel('Frame number', color='#8b949e')
    ax2.legend(facecolor='#21262d', labelcolor='#8b949e', fontsize=9)

    plt.tight_layout()
    plt.savefig('video_analysis.png', dpi=150, bbox_inches='tight', facecolor='#0d1117')
    plt.show()

if __name__ == '__main__':
    path = sys.argv[1] if len(sys.argv) > 1 else 'test_video.mp4'
    rate = int(sys.argv[2]) if len(sys.argv) > 2 else 1
    analyse_video(path, rate)

"""
Embed a message into specific frames of a video using LSB steganography.
For detection testing only.

Usage: python video_embed.py input.mp4 output.mp4 "secret message" 10 20 30
       (embed message into frames 10, 20, 30)
"""
import cv2
import numpy as np
import sys

def embed_in_frame(frame: np.ndarray, message: str) -> np.ndarray:
    """Embed message bytes into LSB of frame pixels (blue channel)."""
    data = message.encode() + b'\x00'  # null terminator
    bits = ''.join(format(b, '08b') for b in data)
    flat = frame[:, :, 0].flatten().copy()
    if len(bits) > len(flat):
        raise ValueError("Message too long for frame")
    for i, bit in enumerate(bits):
        flat[i] = (flat[i] & ~1) | int(bit)
    result = frame.copy()
    result[:, :, 0] = flat.reshape(frame.shape[:2])
    return result

def embed_video(src: str, dst: str, message: str, target_frames: list):
    cap = cv2.VideoCapture(src)
    fps = cap.get(cv2.CAP_PROP_FPS)
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(dst, fourcc, fps, (w, h))
    frame_num = 0
    target_set = set(target_frames)
    embedded = []

    while True:
        ret, frame = cap.read()
        if not ret: break
        if frame_num in target_set:
            frame = embed_in_frame(frame, message)
            embedded.append(frame_num)
        out.write(frame)
        frame_num += 1

    cap.release(); out.release()
    print(f"Embedded in frames: {embedded}")
    print(f"Output: {dst}")

if __name__ == '__main__':
    src, dst, msg = sys.argv[1], sys.argv[2], sys.argv[3]
    frames = [int(x) for x in sys.argv[4:]]
    embed_video(src, dst, msg, frames)