Security Skills Atlas
red team

Performing Red Team Operations with Covenant C2

Conduct red team operations using the Covenant C2 framework for authorized adversary simulation, including listener setup, grunt deployment, task execution, and lateral movement tracking.

adversary-simulationc2covenantpenetration-testingred-team
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1021 on the official MITRE ATT&CK siteT1059 on the official MITRE ATT&CK siteT1078 on the official MITRE ATT&CK siteT1190 on the official MITRE ATT&CK siteT1595 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-07 on the official NIST CSF 2.0 siteGV.OV-02 on the official NIST CSF 2.0 siteID.RA-01 on the official NIST CSF 2.0 site

Overview

Covenant is a collaborative .NET C2 framework for red teamers that provides a Swagger-documented REST API for managing listeners, launchers, grunts (agents), and tasks. This skill covers automating Covenant operations through its API for authorized red team engagements: creating HTTP/HTTPS listeners, generating binary and PowerShell launchers, deploying grunts, executing tasks on compromised hosts, and tracking lateral movement.

When to Use

  • When conducting security assessments that involve performing red team with covenant
  • When following incident response procedures for related security events
  • When performing scheduled security testing or auditing activities
  • When validating security controls through hands-on testing

Prerequisites

  • Covenant C2 server deployed (Docker or .NET 6)
  • Python 3.9+ with requests library
  • Covenant API token (obtained via /api/users/login)
  • Written authorization for red team engagement
  • Isolated lab or authorized target environment

Steps

Step 1: Authenticate to Covenant API

Obtain a JWT token by posting credentials to /api/users/login endpoint.

Step 2: Create Listener

Configure an HTTP or HTTPS listener with callback URLs and bind address.

Step 3: Generate Launcher

Create a binary, PowerShell, or MSBuild launcher tied to the listener for grunt deployment.

Step 4: Deploy and Manage Grunts

Monitor grunt callbacks, execute tasks, and collect output from compromised hosts.

Step 5: Document Operations

Generate an operations report documenting all actions, timestamps, and findings.

Expected Output

JSON report with listener configuration, active grunts, executed tasks, and task output for engagement documentation.

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.1 KB

API Reference: Performing Red Team with Covenant C2

Covenant REST API Endpoints

Endpoint Method Purpose
/api/users/login POST Authenticate and get JWT token
/api/listeners GET List all listeners
/api/listeners/http POST Create HTTP listener
/api/grunts GET List all grunts (agents)
/api/grunts/{id}/interact POST Execute task on grunt
/api/grunttasks/{id} GET Get task output
/api/launchers/binary PUT Generate binary launcher
/api/launchers/powershell PUT Generate PowerShell launcher
/api/launchers/msbuild PUT Generate MSBuild launcher

Authentication

POST /api/users/login
{"userName": "admin", "password": "pass"}
 
Response: {"covenantToken": "eyJhbGciOi..."}
Header: Authorization: Bearer <token>

Listener Configuration

Field Type Description
name string Listener display name
bindAddress string IP to bind (0.0.0.0 for all)
bindPort int Port for grunt callbacks
connectAddresses array Callback addresses for grunts
listenerTypeId int 1=HTTP, 2=Bridge

Grunt Status Values

Status Description
Uninitialized Grunt created but not connected
Stage0 Initial callback received
Stage1 Key exchange in progress
Stage2 Fully staged and active
Active Connected and ready for tasks
Lost Missed check-in threshold

Built-in Tasks

Task Description
WhoAmI Current user identity
GetHostname Target hostname
ListDirectory Directory listing
Download Retrieve file from target
Upload Upload file to target
PowerShell Execute PowerShell command
Assembly Load and execute .NET assembly
Mimikatz Credential extraction

References

Scripts 1

agent.py7.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for red team operations with Covenant C2 framework.

Automates Covenant C2 operations through its REST API: listener
management, launcher generation, grunt monitoring, and task
execution for authorized adversary simulation engagements.
"""
# For authorized red team engagements only

import argparse
import json
import os
from datetime import datetime
from pathlib import Path

try:
    import requests
except ImportError:
    requests = None


class CovenantC2Agent:
    """Manages Covenant C2 operations via its REST API."""

    def __init__(self, covenant_url, username, password,
                 output_dir="./covenant_ops"):
        self.base_url = covenant_url.rstrip("/")
        self.token = None
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.operations_log = []
        self._authenticate(username, password)

    def _authenticate(self, username, password):
        """Authenticate and obtain JWT token from Covenant API."""
        if not requests:
            return
        try:
            resp = requests.post(
                f"{self.base_url}/api/users/login",
                json={"userName": username, "password": password},
                verify=False, timeout=10,
            )
            if resp.status_code == 200:
                data = resp.json()
                self.token = data.get("covenantToken") or data.get("token")
                self._log("authenticate", "success", {"user": username})
        except requests.RequestException as e:
            self._log("authenticate", "failed", {"error": str(e)})

    def _api(self, method, path, data=None):
        if not requests or not self.token:
            return None
        try:
            resp = requests.request(
                method, f"{self.base_url}/api{path}",
                headers={"Authorization": f"Bearer {self.token}",
                          "Content-Type": "application/json"},
                json=data, verify=False, timeout=15,
            )
            return resp
        except requests.RequestException:
            return None

    def _log(self, action, status, details=None):
        self.operations_log.append({
            "timestamp": datetime.utcnow().isoformat(),
            "action": action, "status": status,
            "details": details or {},
        })

    def list_listeners(self):
        """List all configured listeners."""
        resp = self._api("GET", "/listeners")
        if resp and resp.status_code == 200:
            listeners = resp.json()
            self._log("list_listeners", "success", {"count": len(listeners)})
            return [{"id": l["id"], "name": l["name"], "status": l["status"],
                     "bindAddress": l.get("bindAddress"),
                     "bindPort": l.get("bindPort"),
                     "listenerType": l.get("listenerType", {}).get("name")}
                    for l in listeners]
        return []

    def create_http_listener(self, name, bind_port=80, connect_addresses=None):
        """Create an HTTP listener for grunt callbacks."""
        listener_data = {
            "name": name,
            "bindAddress": "0.0.0.0",
            "bindPort": bind_port,
            "connectAddresses": connect_addresses or ["0.0.0.0"],
            "listenerTypeId": 1,
            "status": "Active",
        }
        resp = self._api("POST", "/listeners/http", listener_data)
        if resp and resp.status_code in (200, 201):
            result = resp.json()
            self._log("create_listener", "success",
                      {"name": name, "port": bind_port, "id": result.get("id")})
            return result
        self._log("create_listener", "failed",
                  {"status": resp.status_code if resp else 0})
        return None

    def list_grunts(self):
        """List all active grunts (agents)."""
        resp = self._api("GET", "/grunts")
        if resp and resp.status_code == 200:
            grunts = resp.json()
            self._log("list_grunts", "success", {"count": len(grunts)})
            return [{"id": g["id"], "name": g["name"], "status": g["status"],
                     "hostname": g.get("hostname"), "userName": g.get("userName"),
                     "ipAddress": g.get("ipAddress"),
                     "operatingSystem": g.get("operatingSystem"),
                     "integrity": g.get("integrity"),
                     "lastCheckIn": g.get("lastCheckIn")}
                    for g in grunts]
        return []

    def create_launcher(self, listener_id, launcher_type="Binary"):
        """Generate a launcher payload for grunt deployment."""
        resp = self._api("PUT", f"/launchers/{launcher_type.lower()}",
                         {"listenerId": listener_id})
        if resp and resp.status_code == 200:
            launcher = resp.json()
            self._log("create_launcher", "success",
                      {"type": launcher_type, "listener_id": listener_id})
            return {"type": launcher_type, "launcherString": launcher.get("launcherString", "")[:200]}
        return None

    def execute_task(self, grunt_id, task_name, parameters=None):
        """Assign and execute a task on a grunt."""
        task_data = {
            "gruntId": grunt_id,
            "taskName": task_name,
            "parameters": parameters or [],
        }
        resp = self._api("POST", f"/grunts/{grunt_id}/interact", task_data)
        if resp and resp.status_code in (200, 201):
            result = resp.json()
            self._log("execute_task", "success",
                      {"grunt_id": grunt_id, "task": task_name})
            return {"taskId": result.get("id"), "output": result.get("gruntTaskOutput", "")}
        self._log("execute_task", "failed",
                  {"grunt_id": grunt_id, "task": task_name})
        return None

    def get_task_output(self, task_id):
        """Retrieve output from a completed task."""
        resp = self._api("GET", f"/grunttasks/{task_id}")
        if resp and resp.status_code == 200:
            return resp.json().get("gruntTaskOutput", "")
        return None

    def generate_report(self):
        """Generate an operations report for engagement documentation."""
        listeners = self.list_listeners()
        grunts = self.list_grunts()

        report = {
            "report_date": datetime.utcnow().isoformat(),
            "covenant_url": self.base_url,
            "active_listeners": listeners,
            "active_grunts": grunts,
            "operations_log": self.operations_log,
            "total_operations": len(self.operations_log),
        }
        out = self.output_dir / "covenant_ops_report.json"
        with open(out, "w") as f:
            json.dump(report, f, indent=2)
        print(json.dumps(report, indent=2))
        return report


def main():
    parser = argparse.ArgumentParser(
        description="Covenant C2 red team operations agent (authorized use only)"
    )
    parser.add_argument("covenant_url", help="Covenant server URL (e.g. https://10.0.0.5:7443)")
    parser.add_argument("--username", default="admin", help="Covenant username")
    parser.add_argument("--password", required=True, help="Covenant password")
    parser.add_argument("--output-dir", default="./covenant_ops",
                        help="Output directory for ops report")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    agent = CovenantC2Agent(args.covenant_url, args.username, args.password,
                            output_dir=args.output_dir)
    agent.generate_report()


if __name__ == "__main__":
    main()
Keep exploring
Performing Active Directory Forest Trust AttackEnumerate and audit Active Directory forest trust relationships using impacket for SID filtering analysis, trust key extraction, cross-forest SID history abuse detection, and inter-realm Kerberos ticket assessment.Building C2 Infrastructure with Sliver FrameworkBuild and configure a resilient command-and-control infrastructure using BishopFox's Sliver C2 framework with redirectors, HTTPS listeners, and multi-operator support for authorized red team engagements.Conducting Full-Scope Red Team EngagementPlan and execute a comprehensive red team engagement covering reconnaissance through post-exploitation using MITRE ATT&CK-aligned TTPs to evaluate an organization's detection and response capabilities.Conducting Pass-the-Ticket AttackPass-the-Ticket (PtT) is a lateral movement technique that uses stolen Kerberos tickets (TGT or TGS) to authenticate to services without knowing the user's password. By extracting Kerberos tickets from memory (LSASS) on a compromised host, an attacker can inject those tickets into their own session to impersonate the ticket owner and access resources as that user.Conducting Spearphishing Simulation CampaignSpearphishing simulation is a targeted social engineering attack vector used by red teams to gain initial access. Unlike broad phishing campaigns, spearphishing uses OSINT-derived intelligence to craft highly personalized messages targeting specific individuals. This skill covers developing pretexts, building payloads, setting up email infrastructure, executing the campaign, and tracking results.Executing Red Team Engagement PlanningRed team engagement planning is the foundational phase that defines scope, objectives, rules of engagement (ROE), threat model selection, and operational timelines before any offensive testing begins. A well-structured engagement plan ensures the red team simulates realistic adversary behavior while maintaining safety guardrails that prevent unintended business disruption.