#!/usr/bin/env python3
"""sqli_prescan.py — deterministic SQL-injection pre-scan.

ARION service artifact. Scans source files for query-construction
patterns that are checkable offline, line-wise — no server, no
runtime, no network, no DB schema.

Findings (severity):
  SQLI-QUERY-INTERP    HIGH   user-controllable text interpolated,
                        concatenated, or format-expanded directly
                        into a query-execution call: py f-strings /
                        % / .format / + concat, JS template literals
                        and + concat, PHP "$var" / . concat, Ruby
                        #{}, Go + concat / fmt.Sprintf, Java +
                        concat / String.format, C# $"" / + concat /
                        string.Format
  SQLI-RAW-UNSAFE      HIGH   same interpolation into an explicitly
                        raw/unsafe API — the footgun surface: raw(),
                        $queryRaw/$executeRaw (Prisma), knex.raw,
                        sequelize.query, find_by_sql, unprepared,
                        FromSqlRaw, ExecuteSqlRaw, connection.execute
  SQLI-IDENT-INTERP    MEDIUM interpolation into identifier/order
                        positions (orderBy/order/groupBy/column/table
                        name/sort direction) — cannot be parameterized
                        and is still injectable when attacker-controlled

Severity downgrades:
  LOW    the interpolated expression is wrapped in a sanitizer on the
         same line (intval/floatval/(int)/escape_string/->escape/
         ->quote/pg_escape), or the file lives under a test/spec/
         fixture/mock path.

Suppression (no finding):
  - parameterized calls: execute("...%s", (v,)), query("...?", [v]),
    "... :name" + bind dict, $1/@p placeholders — only flagged when an
    interpolation trigger is present on the call.
  - full-line comments; /* */ block-comment bodies.
  - query strings stored in a variable and passed by name (the flag
    lands on the line where the sink receives the dangerous argument —
    a plain identifier argument is not flagged).

Multi-line calls: when a sink call opens with unbalanced parens, up to
6 following lines are inspected and the finding is attributed to the
sink line. Covers f'''...''' / f\"\"\"...\"\"\" bodies ({name} inside
a pending f-string block is flagged).

Targets: .js .jsx .ts .tsx .mjs .cjs .py .rb .go .java .kt .php .cs
  .ex .exs .rs source files.

Honest scope: PRE-SCAN, not an audit. It cannot prove taint — a flag
means "this line builds SQL from non-literal parts, review where the
value comes from". Queries assembled across many lines via a builder
object, ORM-safe APIs, and stored procedures are out of scope. A clean
report is not proof the codebase is injection-free.

Usage:
  sqli_prescan.py <file-or-dir> [--json]
  sqli_prescan.py --selftest
Stdlib only. Exit 0 = no HIGH/MEDIUM findings, 1 = findings, 2 = usage.
"""
import json
import os
import re
import sys

SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "INFO": 3}

TARGET_EXT = (
    ".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".py", ".rb", ".go",
    ".java", ".kt", ".php", ".cs", ".rs", ".ex", ".exs",
)

SKIP_DIRS = {
    "node_modules", ".git", "vendor", "dist", "build", "target",
    "__pycache__", ".venv", "venv", "coverage", ".next", "out",
}

# --- sinks ------------------------------------------------------------------

# Raw-SQL APIs — interpolation here is the canonical footgun.
RAW_SINK_RE = re.compile(
    r"(?i)(\$queryRaw|\$executeRaw|knex\.raw|sequelize\.query|"
    r"find_by_sql|unprepared|select_all|FromSqlRaw|ExecuteSqlRaw|"
    r"ExecuteSqlCommand|connection\.execute|\.raw\s*\(|\.extra\s*\()"
)

# Identifier/structure positions — MEDIUM (can't be bound as params).
IDENT_SINK_RE = re.compile(
    r"(?i)(orderBy|order_by|->orderBy|\.order\s*\(|groupBy|group_by|"
    r"\.sort\s*\(|\.column\s*\(|\.table\s*\(|\.pluck\s*\()")

# General query-execution verbs — safe to match anywhere.
STRONG_SINK_RE = re.compile(
    r"(?i)\b(execute|executemany|executescript|execute_query|"
    r"executeQuery|executeUpdate|createQuery|createNativeQuery|"
    r"mysql_query|mysqli_query|pg_query|pg_send_query|sqlite_query|"
    r"mssql_query|odbc_exec|queryRow|executeSql|ExecuteReader|"
    r"ExecuteNonQuery|ExecuteScalar|SqlCommand)\s*\(")

# ORM predicate APIs — flagged only when an interpolation trigger is
# present (parameterized/hashes forms produce no trigger).
GENERIC_SINK_RE = re.compile(
    r"\.(where|having|find_by|joins|delete_all|update_all)\s*\(")
# ORM verbs that are also common array/DOM APIs in JS/TS — only
# counted as sinks in .py/.rb/.php/.ex/.exs files.
GENERIC_STRICT_RE = re.compile(
    r"\.(select|filter|exists|order|group|get_by)\s*\(")
GENERIC_STRICT_EXT = {".py", ".rb", ".php", ".ex", ".exs"}

# Generic verbs that need a db-ish receiver to count as a sink.
DBCTX_SINK_RE = re.compile(
    r"(?i)\b(db|conn|connection|pool|client|knex|sequelize|prisma|pdo|"
    r"dbh|sql|repo|session|engine|cursor|cur|stmt|statement|database|"
    r"em|entityManager|jdbcTemplate|tx|trx|qb)\s*[.:\->]+\s*"
    r"(query|exec|raw|all|get|run|prepare|where|select|stream|each|"
    r"many|one|none|any|first)\s*\(")

# --- interpolation triggers --------------------------------------------------

# Python/Ruby-style in-string interpolation
TRIG_FSTR1 = re.compile(r"\bf['\"][^'\"\n]*\{[A-Za-z_]")     # f"..{x}.."
TRIG_FSTR3 = re.compile(r"\bf('''|\"\"\")")                 # f'''.. opens
TRIG_RB = re.compile(r"#\{[^}]{1,80}\}")                    # "..#{x}.."
# JS template literal
TRIG_TL = re.compile(r"`[^`\n]*\$\{")
TRIG_TL_BARE = re.compile(r"\$\{")                          # inside pending
# PHP double-quoted "$var" / "{$var}" — php files only
TRIG_PHPSTR = re.compile(r'"[^"\n]*\{\$|"[^"\n]*\$[A-Za-z_]\w*')
# C# interpolated string $"...{x}..."
TRIG_CS = re.compile(r'\$"[^"\n]*\{[^}\n]')
# String concatenation: ".." + x / x + ".." (php also ".." . $x)
TRIG_CAT_L = re.compile(r"['\"`][^'\"`\n]*['\"`]\s*\+\s*[A-Za-z_$]")
TRIG_CAT_R = re.compile(r"[A-Za-z_$\w)\]]\s*\+\s*['\"`]")
TRIG_CAT_PHP = re.compile(
    r"['\"][^'\"\n]*['\"]\s*\.\s*\$|\$[\w]+\s*\.\s*['\"]")
# Format-expansion sinks
TRIG_FMT = re.compile(
    r"(%s\b[^'\"\n]*['\"]\s*%|['\"]\s*%\s*[\(\w]|\.format\s*\(|"
    r"fmt\.Sprintf|sprintf\s*\(|String\.format|string\.Format|"
    r"str_replace\s*\(|pstmt|\.bind_param)")

# Generic brace used only inside a pending f-string block
TRIG_BRACE = re.compile(r"\{[A-Za-z_][\w.\[\]()]{0,60}\}")

# identifier-ish interpolated names -> HIGH downgraded to MEDIUM
# (identifier position can't be parameterized; still needs a
# human check on where the value comes from)
IDENT_NAME_RE = re.compile(
    r"(?i)^(col(?:umn)?s?|table|tbl|field|order|sort|dir|direction|"
    r"schema|key|name|alias|kind|bucket|index|db|database|attr|"
    r"group|grouping|partition|shard|tenant|pk|fk|asc|desc|"
    r"db_name|table_name|column_name|order_by|sort_by|group_by)$")
EXPR_RE = re.compile(r"\$\{([^}]*)\}|\{([A-Za-z_][\w.]*)\}")
SIMPLE_IDENT_RE = re.compile(r"^[A-Za-z_]\w*$")

# same-line sanitizer wrappers -> downgrade to LOW
ESCAPE_RE = re.compile(
    r"(?i)(mysql_real_escape|mysqli_real_escape|sqlite_escape|"
    r"escape_string|->escape\s*\(|\.escape\s*\(|->quote\s*\(|"
    r"\bintval\s*\(|floatval\s*\(|\(int\)\s*\$|abs\s*\(\s*int|"
    r"pg_escape|->quoteIdentifier|cast\s*\()")

COMMENT_LINE = {
    ".py": re.compile(r"^\s*#"), ".rb": re.compile(r"^\s*#"),
    ".ex": re.compile(r"^\s*#"), ".exs": re.compile(r"^\s*#"),
}

TEST_PATH_RE = re.compile(
    r"(?i)(^|/)(test|tests|spec|specs|fixtures?|mocks?|__tests__|"
    r"testdata)(/|$)|(^|/)test_|_test\.|\.test\.|\.spec\.|spec_")


def sev_rank(s):
    return SEV_ORDER.get(s, 9)


class Finding:
    def __init__(self, code, sev, file, line, msg):
        self.code, self.sev, self.file, self.line, self.msg = \
            code, sev, file, line, msg

    def as_dict(self):
        return {"code": self.code, "severity": self.sev,
                "file": self.file, "line": self.line, "msg": self.msg}


def is_comment(line, ext):
    cre = COMMENT_LINE.get(ext)
    if cre and cre.match(line):
        return True
    stripped = line.strip()
    return stripped.startswith(("//", "*", "<!--"))


def test_path(rel):
    return bool(TEST_PATH_RE.search("/" + rel.replace(os.sep, "/")))


def sink_kind(line, ext):
    """Return 'raw' | 'ident' | 'query' | None for a sink-bearing line."""
    if RAW_SINK_RE.search(line):
        return "raw"
    if IDENT_SINK_RE.search(line):
        return "ident"
    if STRONG_SINK_RE.search(line) or GENERIC_SINK_RE.search(line) \
            or DBCTX_SINK_RE.search(line):
        return "query"
    if ext in GENERIC_STRICT_EXT and GENERIC_STRICT_RE.search(line):
        return "query"
    return None


def triggers_for(line, ext):
    """List of trigger names present on a single line."""
    out = []
    if TRIG_FSTR1.search(line) or TRIG_FSTR3.search(line):
        out.append("fstring")
    if ext == ".rb" and TRIG_RB.search(line):
        out.append("ruby-interp")
    if TRIG_TL.search(line):
        out.append("template-literal")
    if ext == ".php" and TRIG_PHPSTR.search(line):
        out.append("php-interp")
    if ext in (".cs",) and TRIG_CS.search(line):
        out.append("csharp-interp")
    if TRIG_CAT_L.search(line) or TRIG_CAT_R.search(line):
        out.append("concat")
    if ext == ".php" and TRIG_CAT_PHP.search(line):
        out.append("php-concat")
    if TRIG_FMT.search(line):
        out.append("format")
    return out


def parens_open(line):
    depth = 0
    instr = None
    esc = False
    for ch in line:
        if esc:
            esc = False
            continue
        if ch == "\\":
            esc = True
            continue
        if instr:
            if ch == instr:
                instr = None
            continue
        if ch in "\"'`":
            instr = ch
        elif ch == "(":
            depth += 1
        elif ch == ")":
            depth -= 1
    return depth > 0


def scan_file(path, rel):
    findings = []
    try:
        lines = open(path, "r", encoding="utf-8", errors="replace") \
            .read().splitlines()
    except OSError:
        return findings

    ext = os.path.splitext(path)[1].lower()
    low_path = test_path(rel)
    in_block_comment = False

    # pending-sink state: a query call opened but parens not yet closed
    pending = None  # dict(kind=, line=, fstr3=bool)

    def emit(kind, ln, trig, text):
        code = {"raw": "SQLI-RAW-UNSAFE",
                "ident": "SQLI-IDENT-INTERP",
                "query": "SQLI-QUERY-INTERP"}[kind]
        sev = {"raw": "HIGH", "ident": "MEDIUM", "query": "HIGH"}[kind]
        if sev == "HIGH":
            # identifier-position interpolation (e.g. ${column}) can't
            # be parameterized but is usually an allowlist — MEDIUM
            exprs = [a or b for a, b in EXPR_RE.findall(text)]
            if exprs and all(
                    SIMPLE_IDENT_RE.match(e) and IDENT_NAME_RE.match(e)
                    for e in exprs):
                sev = "MEDIUM"
        if low_path or ESCAPE_RE.search(text):
            sev = "LOW"
        msg = {
            "raw": "raw-SQL API receives %s-built text — these calls "
                   "skip parameterization entirely" % trig,
            "ident": "%s into identifier/order position — cannot be "
                     "parameterized; must be allowlisted" % trig,
            "query": "query call built with %s — if any part is "
                     "user-influenced this is injectable; use bound "
                     "parameters" % trig,
        }[kind]
        findings.append(Finding(code, sev, rel, ln, msg))

    for i, line in enumerate(lines, 1):
        # block comment tracking (c-family)
        if in_block_comment:
            if "*/" in line:
                in_block_comment = False
            continue
        if "/*" in line:
            # crude: ignore single-line /* ... */
            if "*/" not in line[line.index("/*") + 2:]:
                in_block_comment = True
                continue

        if is_comment(line, ext):
            continue

        kind = sink_kind(line, ext)
        trigs = triggers_for(line, ext)

        if pending is not None:
            p = pending
            p["n"] += 1
            # trigger inside the still-open call?
            more = trigs + (["brace"] if p.get("fstr3")
                            and TRIG_BRACE.search(line) else [])
            if p.get("fstr3") is False and TRIG_FSTR3.search(line):
                p["fstr3"] = True
            if more:
                emit(p["kind"], p["line"], "+".join(more) or "interp",
                     line)
                pending = None
            elif not parens_open(line) and p["n"] > 0:
                # call closed without a trigger
                pending = None
            elif p["n"] >= 6:
                pending = None
            continue

        if kind and trigs:
            emit(kind, i, "+".join(trigs), line)
        elif kind and parens_open(line):
            pending = {"kind": kind, "line": i, "n": 0,
                       "fstr3": bool(TRIG_FSTR3.search(line))}

    return findings


def find_targets(root):
    if os.path.isfile(root):
        return [root]
    out = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fn in filenames:
            if fn.lower().endswith(TARGET_EXT):
                out.append(os.path.join(dirpath, fn))
    return sorted(out)


def render(findings, targets):
    if not findings:
        return ("SQLI pre-scan: %d file(s) scanned, 0 findings. "
                "Clean report != proof of safety (pre-scan scope)."
                % len(targets))
    findings = sorted(findings, key=lambda f:
                      (sev_rank(f.sev), f.file, f.line))
    out = ["SQLI pre-scan: %d file(s), %d finding(s)"
           % (len(targets), len(findings)), ""]
    cur = None
    for f in findings:
        if f.sev != cur:
            cur = f.sev
            out.append("== %s ==" % cur)
        out.append("%s:%d  [%s] %s" % (f.file, f.line, f.code, f.msg))
    out.append("")
    out.append("Pre-scan only: flags non-literal SQL construction at "
               "call sites. Confirm the source of the interpolated "
               "value before treating any flag as exploitable.")
    return "\n".join(out)


SELFTEST = [
    # --- positive vectors ---
    ("a.py", 'cur.execute(f"SELECT * FROM u WHERE id={uid}")\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.py", 'cur.execute("SELECT * FROM u WHERE id=" + uid)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.py", 'cur.execute("SELECT {}".format(col))\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.py", 'cur.execute("SELECT * FROM u WHERE id=%s" % uid)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.py", 'cur.execute(\n    f"SELECT {col} FROM t"\n)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.js", 'db.query(`SELECT * FROM u WHERE id=${uid}`)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.js", 'db.query("SELECT * FROM u WHERE id=" + uid)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.ts", 'const r = await prisma.$queryRaw`SELECT ${x}`\n',
     {"SQLI-RAW-UNSAFE"}),
    ("a.js", 'knex.raw("select * from t where a=" + a)\n',
     {"SQLI-RAW-UNSAFE"}),
    ("a.php", 'mysql_query("SELECT * FROM u WHERE id=$id");\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.php", '$db->query("SELECT * FROM u WHERE id=" . $id);\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.rb", 'User.where("name = \'#{n}\'")\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.go", 'rows, _ := db.Query("SELECT * FROM u WHERE id=" + id)\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.go", 'db.Query(fmt.Sprintf("SELECT * FROM u WHERE id=%d", id))\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.java",
     'st.executeQuery("SELECT * FROM u WHERE id=" + id);\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.cs", 'new SqlCommand($"SELECT * FROM u WHERE id={id}")\n',
     {"SQLI-QUERY-INTERP"}),
    ("a.cs", 'db.Database.ExecuteSqlRaw("DELETE FROM t WHERE x=" + x)\n',
     {"SQLI-RAW-UNSAFE"}),
    ("a.py", 'q = "SELECT * FROM t ORDER BY " + col\n'
     'cur.execute(q)\n', set()),  # var-arg not flagged line-wise
    ("a.py", 'items.order(f"{col} {direction}")\n',
     {"SQLI-IDENT-INTERP"}),
    ("a.js", 'qb.orderBy(`${col} ${dir}`)\n',
     {"SQLI-IDENT-INTERP"}),
    # --- downgrades ---
    ("a.php", 'mysql_query("SELECT * FROM u WHERE id=" . intval($id));\n',
     {"SQLI-QUERY-INTERP"}),  # still flags, LOW tier
    ("tests/test_a.py",
     'cur.execute(f"SELECT * FROM u WHERE id={uid}")\n',
     {"SQLI-QUERY-INTERP"}),  # LOW via test path
    # --- negative vectors ---
    ("a.py", 'cur.execute("SELECT * FROM u WHERE id=%s", (uid,))\n',
     set()),
    ("a.js", 'db.query("SELECT * FROM u WHERE id=?", [uid])\n', set()),
    ("a.py", 'session.execute(text("SELECT :x"), {"x": v})\n', set()),
    ("a.py", '# cur.execute(f"SELECT * FROM u WHERE id={uid}")\n',
     set()),
    ("a.js", 'const q = "SELECT * FROM t"; db.query(q)\n', set()),
    ("a.js", '// db.query(`SELECT ${x}`)\n', set()),
    ("a.js", '/* db.query(`SELECT ${x}`)\n   more */\n', set()),
    ("a.py", 'def f(x):\n    return x\n', set()),
    ("a.js", 'res.json({ok: true})\n', set()),
    ("a.cs", 'new SqlCommand("SELECT * FROM u WHERE id=@id", c)\n',
     set()),
    # JS array/DOM verbs are not ORM sinks
    ("a.tsx", "class={['b', v ? `b-${v}` : ''].filter(Boolean).join(' ')}\n",
     set()),
    ("a.js", 'items.select(`${x}`)\n', set()),
    ("a.js", 'arr.exists(`${x}`)\n', set()),
    # identifier-name interpolation downgrades HIGH -> MEDIUM (same code)
    ("a.ts", 'await pool.query(`insert into t (id, ${column}) v ($1,$2)`)\n',
     {"SQLI-QUERY-INTERP"}),
    # knex-style where with interpolation still counts in JS
    ("a.js", 'knex.where(`id = ${x}`)\n', {"SQLI-QUERY-INTERP"}),
]


def run_selftest():
    import tempfile
    fails = []
    for i, (name, src, expect) in enumerate(SELFTEST):
        with tempfile.TemporaryDirectory() as td:
            p = os.path.join(td, name)
            os.makedirs(os.path.dirname(p), exist_ok=True)
            open(p, "w").write(src)
            got = {f.code for f in scan_file(p, name)}
            missing = expect - got
            extra = got - expect
            if missing or extra:
                fails.append(("v%d %s" % (i, name),
                              "missing=%s extra=%s" % (missing, extra)))
    if fails:
        for f_ in fails:
            print("SELFTEST FAIL", f_)
        return 1
    print("SELFTEST PASS (%d vectors)" % len(SELFTEST))
    return 0


def main():
    args = [a for a in sys.argv[1:] if a != "--json"]
    want_json = "--json" in sys.argv
    if "--selftest" in sys.argv:
        return run_selftest()
    if not args:
        print(__doc__)
        return 2
    root = args[0]
    targets = find_targets(root)
    findings = []
    for p in targets:
        rel = os.path.relpath(p, root) if os.path.isdir(root) \
            else os.path.basename(p)
        findings += scan_file(p, rel)
    if want_json:
        print(json.dumps([f.as_dict() for f in findings], indent=1))
    else:
        print(render(findings, targets))
    return 1 if any(f.sev in ("HIGH", "MEDIUM") for f in findings) else 0


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