fix(c2): redteam-friendly verify_tls default (sprint 10) #12
@@ -84,7 +84,7 @@ def upsert_c2_config(eid: int):
|
|||||||
if not parsed.hostname:
|
if not parsed.hostname:
|
||||||
return jsonify({"error": "url must contain a hostname"}), 400
|
return jsonify({"error": "url must contain a hostname"}), 400
|
||||||
|
|
||||||
verify_tls = data.get("verify_tls", True)
|
verify_tls = data.get("verify_tls", False)
|
||||||
if not isinstance(verify_tls, bool):
|
if not isinstance(verify_tls, bool):
|
||||||
return jsonify({"error": "verify_tls must be a boolean"}), 400
|
return jsonify({"error": "verify_tls must be a boolean"}), 400
|
||||||
|
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ class C2Config(db.Model): # type: ignore[name-defined]
|
|||||||
)
|
)
|
||||||
url = db.Column(db.Text, nullable=False)
|
url = db.Column(db.Text, nullable=False)
|
||||||
api_token_encrypted = db.Column(db.Text, nullable=False)
|
api_token_encrypted = db.Column(db.Text, nullable=False)
|
||||||
verify_tls = db.Column(db.Boolean, nullable=False, default=True)
|
verify_tls = db.Column(db.Boolean, nullable=False, default=False)
|
||||||
created_at = db.Column(
|
created_at = db.Column(
|
||||||
db.DateTime, nullable=False, default=lambda: datetime.now(UTC)
|
db.DateTime, nullable=False, default=lambda: datetime.now(UTC)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import os
|
|||||||
from backend.app.services.c2.adapter import C2Adapter
|
from backend.app.services.c2.adapter import C2Adapter
|
||||||
|
|
||||||
|
|
||||||
def get_adapter(url: str, api_token: str, verify_tls: bool = True) -> C2Adapter:
|
def get_adapter(url: str, api_token: str, verify_tls: bool = False) -> C2Adapter:
|
||||||
"""Return the correct C2Adapter based on MIMIC_C2_ADAPTER (default: mythic)."""
|
"""Return the correct C2Adapter based on MIMIC_C2_ADAPTER (default: mythic)."""
|
||||||
adapter_name = os.environ.get("MIMIC_C2_ADAPTER", "mythic").lower()
|
adapter_name = os.environ.get("MIMIC_C2_ADAPTER", "mythic").lower()
|
||||||
|
|
||||||
|
|||||||
@@ -15,6 +15,8 @@ from __future__ import annotations
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
import urllib3
|
||||||
|
from urllib3.exceptions import InsecureRequestWarning
|
||||||
|
|
||||||
from backend.app.services.c2.adapter import (
|
from backend.app.services.c2.adapter import (
|
||||||
C2Adapter,
|
C2Adapter,
|
||||||
@@ -113,10 +115,14 @@ query GetTaskOutput($display_id: Int!) {
|
|||||||
class MythicAdapter(C2Adapter):
|
class MythicAdapter(C2Adapter):
|
||||||
"""Real Mythic 3.x adapter using GraphQL over HTTP."""
|
"""Real Mythic 3.x adapter using GraphQL over HTTP."""
|
||||||
|
|
||||||
def __init__(self, url: str, api_token: str, verify_tls: bool = True) -> None:
|
def __init__(self, url: str, api_token: str, verify_tls: bool = False) -> None:
|
||||||
self._url = url.rstrip("/") + "/graphql"
|
self._url = url.rstrip("/") + "/graphql"
|
||||||
self._token = api_token
|
self._token = api_token
|
||||||
self._verify = verify_tls
|
self._verify = verify_tls
|
||||||
|
if not verify_tls:
|
||||||
|
# Lab/red-team Mythic instances commonly use self-signed certs.
|
||||||
|
# Suppress urllib3 warnings once per process when verification is off.
|
||||||
|
urllib3.disable_warnings(InsecureRequestWarning)
|
||||||
|
|
||||||
def _headers(self) -> dict[str, str]:
|
def _headers(self) -> dict[str, str]:
|
||||||
return {
|
return {
|
||||||
|
|||||||
@@ -0,0 +1,33 @@
|
|||||||
|
"""flip c2_config.verify_tls server_default to false
|
||||||
|
|
||||||
|
Revision ID: 0008_c2_verify_tls_default_false
|
||||||
|
Revises: 0007
|
||||||
|
Create Date: 2026-06-21 00:00:00.000000
|
||||||
|
"""
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from alembic import op
|
||||||
|
|
||||||
|
revision = "0008_c2_verify_tls_default_false"
|
||||||
|
down_revision = "0007"
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
with op.batch_alter_table("c2_config") as batch_op:
|
||||||
|
batch_op.alter_column(
|
||||||
|
"verify_tls",
|
||||||
|
existing_type=sa.Boolean(),
|
||||||
|
existing_nullable=False,
|
||||||
|
server_default=sa.false(),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
with op.batch_alter_table("c2_config") as batch_op:
|
||||||
|
batch_op.alter_column(
|
||||||
|
"verify_tls",
|
||||||
|
existing_type=sa.Boolean(),
|
||||||
|
existing_nullable=False,
|
||||||
|
server_default=sa.true(),
|
||||||
|
)
|
||||||
@@ -206,13 +206,13 @@ def test_get_config_returns_has_token_not_cleartext(
|
|||||||
assert "s3cr3t" not in str(body)
|
assert "s3cr3t" not in str(body)
|
||||||
|
|
||||||
|
|
||||||
def test_get_config_verify_tls_default_true(
|
def test_get_config_verify_tls_default_false(
|
||||||
client: FlaskClient, admin_token: str
|
client: FlaskClient, admin_token: str
|
||||||
) -> None:
|
) -> None:
|
||||||
eng = _make_engagement(client, admin_token)
|
eng = _make_engagement(client, admin_token)
|
||||||
_put_config(client, admin_token, eng["id"])
|
_put_config(client, admin_token, eng["id"], verify_tls=False)
|
||||||
resp = client.get(f"/api/engagements/{eng['id']}/c2-config", headers=_h(admin_token))
|
resp = client.get(f"/api/engagements/{eng['id']}/c2-config", headers=_h(admin_token))
|
||||||
assert resp.get_json()["verify_tls"] is True
|
assert resp.get_json()["verify_tls"] is False
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user