Deploying the Mirai Console

Begin by obtaining the Mirai Console Loader (MCL) to establish the base environment:

# Download and execute the installer
curl -LO https://github.com/iTXTech/mcl-installer/releases/latest/download/mcl-installer
chmod +x mcl-installer
./mcl-installer

# Initialize the console
./mcl

Configuring the HTTP Adapter

Install the mirai-api-http plugin to expose RESTful endpoints:

./mcl --update-package net.mamoe:mirai-api-http --channel stable-v2 --type plugin

Modify config/net.mamoe.mirai-api-http/setting.yml to enable external access:

adapters:
  - http

enableVerify: true
verifyKey: your-secret-key-here

debug: false
singleMode: false
cacheSize: 4096

adapterSettings:
  http:
    host: 0.0.0.0
    port: 7777
    cors: ["*"]
    unreadQueueMaxSize: 100

Note: When containerizing with Docker, bind to 0.0.0.0 instead of localhost to allow external connections. The verifyKey serves as the authentication credential for subsequent API calls.

Authentication Components

Modern QQ protocols require signature servers and login solvers to bypass security challenges.

Login Solver: Place the mirai-login-solver-sakura JAR into the plugins/ directory. This component exposes port 22333 for mobile device verification during authentication.

Sign Server (QSign): Download the d62ddce release of QSign and deploy it alongside the console:

• Copy the plugin JAR to plugins/
• Extract txlib/ to the console root
• Update the protocol version to 8.9.90 in the configuration

Launching the Bot Instance

Start the console and authenticate using the Android Pad protocol to avoid rate limiting:

./mcl

Inside the console:

/login <qq_number> <password> ANDROID_PAD

After successful authentication, configure automatic login persistence to eliminate manual intervention on restart.

Python Client Implementation

The following client handles session management and message dispatching:

import requests
from typing import Optional, Dict, Any, List
import time

class MiraiBotClient:
    def __init__(self, gateway: str, auth_token: str, bot_id: str):
        self.gateway = gateway.rstrip('/')
        self.auth_token = auth_token
        self.bot_id = bot_id
        self.session_token: Optional[str] = None
        self._establish_session()
        
    def _establish_session(self) -> None:
        """Authenticate and bind session to bot instance"""
        auth_endpoint = f"{self.gateway}/verify"
        response = requests.post(auth_endpoint, json={"verifyKey": self.auth_token})
        response.raise_for_status()
        self.session_token = response.json()["session"]
        
        bind_endpoint = f"{self.gateway}/bind"
        requests.post(bind_endpoint, json={
            "sessionKey": self.session_token,
            "qq": self.bot_id
        })
    
    def _refresh_if_expired(self) -> None:
        """Mirai sessions expire after 30 minutes of inactivity"""
        check_url = f"{self.gateway}/sessionInfo?sessionKey={self.session_token}"
        resp = requests.get(check_url)
        if resp.json().get("code") == 3:
            self._establish_session()
    
    def fetch_group_roster(self) -> List[Dict[str, Any]]:
        self._refresh_if_expired()
        url = f"{self.gateway}/groupList?sessionKey={self.session_token}"
        return requests.get(url).json()
    
    def dispatch_group_message(self, group_id: int, content: str) -> Dict[str, Any]:
        self._refresh_if_expired()
        payload = {
            "sessionKey": self.session_token,
            "target": group_id,
            "messageChain": [{"type": "Plain", "text": content}]
        }
        url = f"{self.gateway}/sendGroupMessage"
        return requests.post(url, json=payload).json()


if __name__ == "__main__":
    client = MiraiBotClient(
        gateway="http://localhost:7777",
        auth_token="your-secret-key-here",
        bot_id="123456789"
    )
    print(client.fetch_group_roster())

GZCTF Event Broadcasting

Integrate with the competition platform's notification API to relay events:

import time
import logging
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Optional
import requests
from dateutil import parser
from mirai_client import MiraiBotClient

# Configuration constants
POLLING_INTERVAL = 2  # Seconds between API checks
GATEWAY_URL = "http://localhost:7777"
AUTH_SECRET = "your-secret-key-here"
BOT_ACCOUNT = "123456789"
TARGET_GROUP = "987654321"
CTF_API_ENDPOINT = "http://ctf-platform.local/api/game/4/notices"

# Notification templates
MESSAGE_TEMPLATES = {
    'Normal': "📢 **Announcement**\nContent: {content}\nTimestamp: {time}",
    'NewChallenge': "🆕 **New Challenge**\nChallenge: {challenge}\nReleased: {time}",
    'NewHint': "💡 **Hint Released**\nChallenge: {challenge} has new hints available\nTime: {time}",
    'FirstBlood': "🩸 **First Blood**\nCongratulations {player} for solving {challenge}!\nTime: {time}",
    'SecondBlood': "🥈 **Second Blood**\n{player} claimed second solve on {challenge}\nTime: {time}",
    'ThirdBlood': "🥉 **Third Blood**\n{player} secured third place on {challenge}\nTime: {time}"
}

class CTFNotificationRelay:
    def __init__(self):
        self.bot = MiraiBotClient(GATEWAY_URL, AUTH_SECRET, BOT_ACCOUNT)
        self.last_sequence = 0
        self.timezone_offset = timezone(timedelta(hours=8))
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger("CTF-Relay")
        
    def normalize_timestamp(self, iso_string: str) -> str:
        """Convert ISO timestamp to localized readable format"""
        parsed = parser.isoparse(iso_string.replace('Z', '+00:00'))
        local_time = parsed.astimezone(self.timezone_offset)
        return local_time.strftime("%Y-%m-%d %H:%M:%S")
    
    def format_notification(self, event: Dict) -> Optional[str]:
        event_type = event.get("type")
        if event_type not in MESSAGE_TEMPLATES:
            return None
            
        values = event.get("values", [])
        timestamp = self.normalize_timestamp(event.get("time", ""))
        
        if event_type == "Normal":
            return MESSAGE_TEMPLATES[event_type].format(content=values[0], time=timestamp)
        else:
            return MESSAGE_TEMPLATES[event_type].format(
                player=values[0], 
                challenge=values[1], 
                time=timestamp
            )
    
    def run(self):
        """Main event loop"""
        self.logger.info("Initializing relay service...")
        
        # Bootstrap: fetch existing to set baseline
        try:
            history = requests.get(CTF_API_ENDPOINT).json()
            if history:
                self.last_sequence = max(n["id"] for n in history)
        except Exception as e:
            self.logger.error(f"Bootstrap failed: {e}")
            return
            
        self.logger.info(f"Monitoring from sequence ID: {self.last_sequence}")
        
        while True:
            try:
                response = requests.get(CTF_API_ENDPOINT)
                response.raise_for_status()
                notifications = sorted(response.json(), key=lambda x: x["id"])
                
                for notice in notifications:
                    if notice["id"] > self.last_sequence:
                        message = self.format_notification(notice)
                        if message:
                            self.logger.info(f"Relaying notification #{notice['id']}")
                            self.bot.dispatch_group_message(int(TARGET_GROUP), message)
                            self.last_sequence = notice["id"]
                            
                time.sleep(POLLING_INTERVAL)
                
            except KeyboardInterrupt:
                self.logger.info("Shutting down relay...")
                break
            except Exception as e:
                self.logger.error(f"Polling error: {e}")
                time.sleep(POLLING_INTERVAL)


if __name__ == "__main__":
    relay = CTFNotificationRelay()
    relay.run()