From 549a24f7b977783253615c09452cebab28ea29b8 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 21 Jun 2026 16:53:53 +0530 Subject: [PATCH 1/3] ci(postgres): add a static pre-commit check for MySQL-only SQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Postgres test job is label-gated, so it does not run on every PR. This adds an always-on pre-commit hook that statically flags the *mechanical* breaks: MySQL-only functions (timestamp(date,time), timediff, str_to_date, date_format/add/sub, group_concat, period_diff, SQL IF()), SHOW INDEX/TABLES/COLUMNS, single-quoted aliases, UPDATE..JOIN, interpolated/f-string SQL carrying MySQL-isms, set_value/db_set(, bool), and MySQL SHOW INDEX result keys. It deliberately does NOT flag the framework auto-translations (ifnull->coalesce, backtick/locate/REGEXP, .like()->ILIKE) nor the *semantic* divergences (loose GROUP BY, case-sensitive ==/IN, NULL ordering, tiebreakers) — those need the test suite, which remains the backstop. AST + structure-gated regex keep false positives near zero (docstrings and prose skipped); '# pg-ok' exempts intentional MariaDB-only branches. Scoped to erpnext/ excluding patches/. Includes a unit test of the checker. --- .github/helper/postgres_compat.py | 186 ++++++++++++++++++ .pre-commit-config.yaml | 12 ++ .../process_payment_reconciliation.py | 2 +- erpnext/tests/test_perf.py | 2 +- erpnext/tests/test_postgres_compat.py | 103 ++++++++++ 5 files changed, 303 insertions(+), 2 deletions(-) create mode 100755 .github/helper/postgres_compat.py create mode 100644 erpnext/tests/test_postgres_compat.py diff --git a/.github/helper/postgres_compat.py b/.github/helper/postgres_compat.py new file mode 100755 index 00000000000..e5fc0578c7a --- /dev/null +++ b/.github/helper/postgres_compat.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +"""Static guard against MySQL-only SQL that breaks on PostgreSQL. + +The Postgres test job is label-gated, so it does not run on every PR. This pre-commit +hook is the always-on first line of defence: it flags the *mechanical* Postgres breaks +that static analysis can catch reliably with a low false-positive rate. + +It deliberately does NOT try to catch the *semantic* divergences (loose GROUP BY, +case-sensitive ==/IN, NULL ordering, ORDER BY ... LIMIT 1 tiebreakers, integer-division +intent, savepoint discipline) — those genuinely need the test suite. Run the full suite +on a Postgres site for those. + +Escape hatch: put `# pg-ok` anywhere on the offending statement's line span (e.g. on a +`SHOW INDEX` query that lives inside an `if frappe.db.db_type == "mariadb":` branch). + +Usage: postgres_compat.py [ ...] (pre-commit passes staged files) +""" + +from __future__ import annotations + +import ast +import re +import sys + +IGNORE = "pg-ok" + +# Strings are only scanned for the patterns below when they have real SQL *structure* +# (not just an English word like "select" or "from"), to keep false positives near zero. +SQL_HINT = re.compile( + r"\bselect\b[\s\S]{0,800}\bfrom\b" # SELECT ... FROM + r"|\bupdate\b[\s\S]{0,400}\bset\b" # UPDATE ... SET + r"|\bdelete\s+from\b" + r"|\binsert\s+into\b" + r"|\bshow\s+(?:index|tables|columns)\b" + r"|\bfrom\s+[\"'`]?tab", # FROM `tabDocType` + re.I, +) + +# MySQL-only constructs with NO frappe auto-translation. (frappe.db.sql already rewrites +# ifnull->coalesce on all engines and backtick/locate/REGEXP on Postgres, and .like() +# renders ILIKE — so those are NOT listed here; flagging them would be false positives.) +SQL_PATTERNS: list[tuple[re.Pattern, str]] = [ + (re.compile(r"\btimestamp\s*\(\s*[^,()]+,", re.I), + "timestamp(date, time) is MySQL-only -> use CombineDatetime() or a precomputed datetime column"), + (re.compile(r"\btimediff\s*\(", re.I), + "timediff() is MySQL-only -> compute the delta in Python"), + (re.compile(r"\bstr_to_date\s*\(", re.I), + "str_to_date() is MySQL-only -> parse in Python and pass a real date"), + (re.compile(r"\bdate_format\s*\(", re.I), + "date_format() is MySQL-only -> filter on a date range instead"), + (re.compile(r"\bdate_(add|sub)\s*\(", re.I), + "date_add()/date_sub() are MySQL-only -> use Python date math or interval arithmetic"), + (re.compile(r"\bgroup_concat\s*\(", re.I), + "group_concat() is MySQL-only -> use GroupConcat (string_agg) or aggregate in Python"), + (re.compile(r"\bperiod_diff\s*\(", re.I), + "period_diff() is MySQL-only -> compute in Python"), + (re.compile(r"\bshow\s+index\b", re.I), + "SHOW INDEX is MySQL-only -> use frappe.db.has_index() / get_column_index()"), + (re.compile(r"\bshow\s+(tables|columns)\b", re.I), + "SHOW TABLES/COLUMNS is MySQL-only -> use frappe.db.get_tables()/table_columns / information-schema helpers"), + (re.compile(r"\bas\s+'[^']+'", re.I), + "single-quoted column alias breaks on Postgres -> use a bare or double-quoted alias"), + (re.compile(r"\bif\s*\(", re.I), + "SQL IF() is MySQL-only -> use CASE WHEN ... THEN ... ELSE ... END (frappe.qb.Case())"), +] + +# UPDATE ... JOIN: both keywords in the same SQL string. +UPDATE_JOIN = (re.compile(r"\bupdate\b", re.I), re.compile(r"\bjoin\b", re.I)) + +MYSQL_RESULT_KEYS = {"Column_name", "Key_name", "Seq_in_index", "Non_unique", "Index_type"} + +SET_BOOL_FUNCS = {"set_value", "db_set"} + + +def _docstring_ids(tree: ast.AST) -> set[int]: + """ids of Constant nodes that are docstrings (so prose describing the rules isn't flagged).""" + ids: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + body = getattr(node, "body", None) + if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str): + ids.add(id(body[0].value)) + return ids + + +class Visitor(ast.NodeVisitor): + def __init__(self, lines: list[str], docstrings: set[int]): + self.lines = lines + self.docstrings = docstrings + self.violations: list[tuple[int, str]] = [] + + def _ignored(self, node: ast.AST) -> bool: + start = getattr(node, "lineno", 1) + end = getattr(node, "end_lineno", start) or start + # honour `# pg-ok` anywhere on the node's line span, or on the line just above + # (the enclosing call, e.g. `frappe.db.sql( # pg-ok`). + lo = max(0, start - 2) + return any(IGNORE in self.lines[i] for i in range(lo, min(end, len(self.lines)))) + + def _flag(self, node: ast.AST, msg: str) -> None: + if not self._ignored(node): + self.violations.append((getattr(node, "lineno", 1), msg)) + + def _scan_sql(self, text: str, node: ast.AST) -> None: + if not SQL_HINT.search(text): + return + for pattern, msg in SQL_PATTERNS: + if pattern.search(text): + self._flag(node, msg) + if UPDATE_JOIN[0].search(text) and UPDATE_JOIN[1].search(text): + self._flag(node, "UPDATE ... JOIN is MySQL-only -> use a correlated subquery (WHERE ... IN/EXISTS)") + + def visit_Constant(self, node: ast.Constant) -> None: + # plain string literals, incl. `"...".format()` and `"..." % (...)` templates + if isinstance(node.value, str) and id(node) not in self.docstrings: + self._scan_sql(node.value, node) + self.generic_visit(node) + + def visit_JoinedStr(self, node: ast.JoinedStr) -> None: + # f-string: scan its STATIC text (interpolated values become a placeholder) so MySQL-isms + # in dynamic SQL are caught, without flagging safe interpolation of identifiers. + text = "".join( + v.value if isinstance(v, ast.Constant) and isinstance(v.value, str) else " ? " + for v in node.values + ) + self._scan_sql(text, node) + # don't recurse: child literal chunks would otherwise be re-scanned individually + + def visit_Call(self, node: ast.Call) -> None: + fn = node.func + name = fn.attr if isinstance(fn, ast.Attribute) else (fn.id if isinstance(fn, ast.Name) else "") + + # row.get("Column_name") — MySQL SHOW INDEX result key + if name == "get" and node.args and isinstance(node.args[0], ast.Constant) and node.args[0].value in MYSQL_RESULT_KEYS: + self._flag(node, f'"{node.args[0].value}" is a MySQL SHOW INDEX result key -> use frappe.db.has_index()/get_column_index()') + + # set_value(..., True) / db_set("field", True) on a Check (int) column + if name in SET_BOOL_FUNCS: + for a in node.args: + if isinstance(a, ast.Constant) and isinstance(a.value, bool): + self._flag(node, f"{name}(..., {a.value}) sets an int/Check column with a bool -> pass 1/0 (Postgres rejects bool->smallint)") + elif isinstance(a, ast.Dict): + for v in a.values: + if isinstance(v, ast.Constant) and isinstance(v.value, bool): + self._flag(node, f"{name}(...) sets an int/Check column with a bool in a dict -> pass 1/0 (Postgres rejects bool->smallint)") + + self.generic_visit(node) + + def visit_Subscript(self, node: ast.Subscript) -> None: + key = node.slice + if isinstance(key, ast.Constant) and key.value in MYSQL_RESULT_KEYS: + self._flag(node, f'"{key.value}" is a MySQL SHOW INDEX result key -> use frappe.db.has_index()/get_column_index()') + self.generic_visit(node) + + +def check_file(path: str) -> list[str]: + try: + src = open(path, encoding="utf-8").read() + except (OSError, UnicodeDecodeError): + return [] + try: + tree = ast.parse(src, filename=path) + except SyntaxError: + return [] # check-ast hook reports real syntax errors + v = Visitor(src.splitlines(), _docstring_ids(tree)) + v.visit(tree) + return [f"{path}:{line}: [pg-compat] {msg}" for line, msg in sorted(set(v.violations))] + + +def main(argv: list[str]) -> int: + out: list[str] = [] + for path in argv: + if path.endswith(".py"): + out.extend(check_file(path)) + if out: + print("\n".join(out)) + print( + f"\n{len(out)} PostgreSQL-incompatibility issue(s). Fix them, or add `# pg-ok` to a " + "line that is intentionally MariaDB-only (e.g. inside an `if frappe.db.db_type == 'mariadb':` branch)." + ) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 958a74595a6..3414bdb83b7 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -66,6 +66,18 @@ repos: - id: ruff-format name: "Run ruff formatter" + - repo: local + hooks: + - id: postgres-compat + name: "PostgreSQL compatibility (static check)" + description: "Flags MySQL-only SQL that breaks on Postgres; the label-gated PG test job is the backstop for semantic divergences." + entry: .github/helper/postgres_compat.py + language: script + files: ^erpnext/.*\.py$ + # patches/ are historical, version-gated migrations (skipped on fresh Postgres installs); + # out of scope for the always-on gate. test_postgres_compat.py holds intentional bad-SQL fixtures. + exclude: ^erpnext/patches/|^erpnext/tests/test_postgres_compat\.py$ + ci: autoupdate_schedule: weekly skip: [] diff --git a/erpnext/accounts/doctype/process_payment_reconciliation/process_payment_reconciliation.py b/erpnext/accounts/doctype/process_payment_reconciliation/process_payment_reconciliation.py index 5ee78e5bbb3..f4440345e96 100644 --- a/erpnext/accounts/doctype/process_payment_reconciliation/process_payment_reconciliation.py +++ b/erpnext/accounts/doctype/process_payment_reconciliation/process_payment_reconciliation.py @@ -503,7 +503,7 @@ def reconcile(doc: None | str = None) -> None: ) else: frappe.db.set_value("Process Payment Reconciliation Log", log, "status", "Reconciled") - frappe.db.set_value("Process Payment Reconciliation Log", log, "reconciled", True) + frappe.db.set_value("Process Payment Reconciliation Log", log, "reconciled", 1) frappe.db.set_value("Process Payment Reconciliation", doc, "status", "Completed") diff --git a/erpnext/tests/test_perf.py b/erpnext/tests/test_perf.py index f37922f645c..4ee5f798c84 100644 --- a/erpnext/tests/test_perf.py +++ b/erpnext/tests/test_perf.py @@ -33,7 +33,7 @@ def _is_leading_index_column(doctype: str, field: str) -> bool: # `table` is a trusted constant (from INDEXED_FIELDS); a table identifier can't be a %s # placeholder in SHOW INDEX, so the f-string is unavoidable and safe here. return bool( - frappe.db.sql( + frappe.db.sql( # pg-ok: MariaDB-only branch; Postgres is handled above via pg_index f"""SHOW INDEX FROM `{table}` WHERE Column_name = %s AND Seq_in_index = 1""", (field,), ) diff --git a/erpnext/tests/test_postgres_compat.py b/erpnext/tests/test_postgres_compat.py new file mode 100644 index 00000000000..fa74380b8fe --- /dev/null +++ b/erpnext/tests/test_postgres_compat.py @@ -0,0 +1,103 @@ +# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors +# See license.txt +"""Unit tests for the .github/helper/postgres_compat.py pre-commit checker. + +This file is excluded from the postgres-compat hook itself (see .pre-commit-config.yaml) +because the fixtures below intentionally contain MySQL-only SQL. +""" + +import importlib.util +import os +import tempfile +import unittest + +import frappe + +_HELPER = os.path.join(frappe.get_app_path("erpnext"), "..", ".github", "helper", "postgres_compat.py") +_spec = importlib.util.spec_from_file_location("postgres_compat", _HELPER) +pgc = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(pgc) + + +def violations(code: str) -> list[str]: + with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f: + f.write(code) + path = f.name + try: + return pgc.check_file(path) + finally: + os.unlink(path) + + +class TestPostgresCompat(unittest.TestCase): + def _assert_flag(self, code: str, needle: str): + out = violations(code) + self.assertTrue( + any(needle in v for v in out), f"expected '{needle}' to be flagged in:\n{code}\ngot: {out}" + ) + + def _assert_clean(self, code: str): + self.assertEqual(violations(code), [], f"expected no flags in:\n{code}") + + # --- catches the mechanical breaks --- + def test_timestamp_two_arg(self): + self._assert_flag( + 'frappe.db.sql("select timestamp(posting_date, posting_time) from `tabSLE`")', + "timestamp(date, time)", + ) + + def test_show_index(self): + self._assert_flag('frappe.db.sql("show index from `tabItem`")', "SHOW INDEX") + + def test_update_join(self): + self._assert_flag( + 'frappe.db.sql("update `tabA` a join `tabB` b on a.x=b.x set a.y=b.y")', "UPDATE ... JOIN" + ) + + def test_single_quoted_alias(self): + self._assert_flag( + "frappe.db.sql(\"select substr(x,1,3) as 'foo' from `tabA`\")", "single-quoted column alias" + ) + + def test_group_concat(self): + self._assert_flag('frappe.db.sql("select group_concat(name) from `tabA`")', "group_concat()") + + def test_set_value_bool(self): + self._assert_flag('frappe.db.set_value("Company", c, "some_check", True)', "bool") + + def test_db_set_bool(self): + self._assert_flag('doc.db_set("is_default", False)', "bool") + + def test_mysql_result_key(self): + self._assert_flag('row.get("Column_name")', "Column_name") + + def test_fstring_sql(self): + self._assert_flag("frappe.db.sql(f\"select date_format(d, '%Y') from `tab{dt}`\")", "date_format()") + + # --- does not false-positive on safe shapes --- + def test_ifnull_is_auto_translated(self): + self._assert_clean('frappe.db.sql("select ifnull(qty, 0) from `tabBin`")') + + def test_like_is_ilike(self): + self._assert_clean('frappe.db.get_all("Item", filters={"item_name": ["like", "%x%"]})') + + def test_prose_with_sql_words(self): + # a translatable message that merely contains "select" and "as '...'" + self._assert_clean( + "frappe.throw(_(\"Cannot select charge type as 'On Previous Row' for first row\"))" + ) + + def test_docstring_describing_rule(self): + self._assert_clean( + 'def f():\n\t"""Avoid MariaDB-only DATE_FORMAT(); read from the pg_index catalog instead."""\n\treturn 1\n' + ) + + def test_qb_is_clean(self): + self._assert_clean("frappe.qb.from_(sle).select(CombineDatetime(sle.posting_date, sle.posting_time))") + + def test_pg_ok_suppresses(self): + self._assert_clean('frappe.db.sql( # pg-ok\n\t"show index from `tabItem`"\n)') + + +if __name__ == "__main__": + unittest.main() From 16e45c41f5781de42993e6a59c64f962b8b1f721 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 21 Jun 2026 17:03:51 +0530 Subject: [PATCH 2/3] ci(postgres): drop the checker's unit test Remove erpnext/tests/test_postgres_compat.py (and its pre-commit exclude); a unit test for the dev-tooling lint helper isn't needed in the app test suite. --- .pre-commit-config.yaml | 4 +- erpnext/tests/test_postgres_compat.py | 103 -------------------------- 2 files changed, 2 insertions(+), 105 deletions(-) delete mode 100644 erpnext/tests/test_postgres_compat.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3414bdb83b7..6e478347b8a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -75,8 +75,8 @@ repos: language: script files: ^erpnext/.*\.py$ # patches/ are historical, version-gated migrations (skipped on fresh Postgres installs); - # out of scope for the always-on gate. test_postgres_compat.py holds intentional bad-SQL fixtures. - exclude: ^erpnext/patches/|^erpnext/tests/test_postgres_compat\.py$ + # out of scope for the always-on gate. + exclude: ^erpnext/patches/ ci: autoupdate_schedule: weekly diff --git a/erpnext/tests/test_postgres_compat.py b/erpnext/tests/test_postgres_compat.py deleted file mode 100644 index fa74380b8fe..00000000000 --- a/erpnext/tests/test_postgres_compat.py +++ /dev/null @@ -1,103 +0,0 @@ -# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors -# See license.txt -"""Unit tests for the .github/helper/postgres_compat.py pre-commit checker. - -This file is excluded from the postgres-compat hook itself (see .pre-commit-config.yaml) -because the fixtures below intentionally contain MySQL-only SQL. -""" - -import importlib.util -import os -import tempfile -import unittest - -import frappe - -_HELPER = os.path.join(frappe.get_app_path("erpnext"), "..", ".github", "helper", "postgres_compat.py") -_spec = importlib.util.spec_from_file_location("postgres_compat", _HELPER) -pgc = importlib.util.module_from_spec(_spec) -_spec.loader.exec_module(pgc) - - -def violations(code: str) -> list[str]: - with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f: - f.write(code) - path = f.name - try: - return pgc.check_file(path) - finally: - os.unlink(path) - - -class TestPostgresCompat(unittest.TestCase): - def _assert_flag(self, code: str, needle: str): - out = violations(code) - self.assertTrue( - any(needle in v for v in out), f"expected '{needle}' to be flagged in:\n{code}\ngot: {out}" - ) - - def _assert_clean(self, code: str): - self.assertEqual(violations(code), [], f"expected no flags in:\n{code}") - - # --- catches the mechanical breaks --- - def test_timestamp_two_arg(self): - self._assert_flag( - 'frappe.db.sql("select timestamp(posting_date, posting_time) from `tabSLE`")', - "timestamp(date, time)", - ) - - def test_show_index(self): - self._assert_flag('frappe.db.sql("show index from `tabItem`")', "SHOW INDEX") - - def test_update_join(self): - self._assert_flag( - 'frappe.db.sql("update `tabA` a join `tabB` b on a.x=b.x set a.y=b.y")', "UPDATE ... JOIN" - ) - - def test_single_quoted_alias(self): - self._assert_flag( - "frappe.db.sql(\"select substr(x,1,3) as 'foo' from `tabA`\")", "single-quoted column alias" - ) - - def test_group_concat(self): - self._assert_flag('frappe.db.sql("select group_concat(name) from `tabA`")', "group_concat()") - - def test_set_value_bool(self): - self._assert_flag('frappe.db.set_value("Company", c, "some_check", True)', "bool") - - def test_db_set_bool(self): - self._assert_flag('doc.db_set("is_default", False)', "bool") - - def test_mysql_result_key(self): - self._assert_flag('row.get("Column_name")', "Column_name") - - def test_fstring_sql(self): - self._assert_flag("frappe.db.sql(f\"select date_format(d, '%Y') from `tab{dt}`\")", "date_format()") - - # --- does not false-positive on safe shapes --- - def test_ifnull_is_auto_translated(self): - self._assert_clean('frappe.db.sql("select ifnull(qty, 0) from `tabBin`")') - - def test_like_is_ilike(self): - self._assert_clean('frappe.db.get_all("Item", filters={"item_name": ["like", "%x%"]})') - - def test_prose_with_sql_words(self): - # a translatable message that merely contains "select" and "as '...'" - self._assert_clean( - "frappe.throw(_(\"Cannot select charge type as 'On Previous Row' for first row\"))" - ) - - def test_docstring_describing_rule(self): - self._assert_clean( - 'def f():\n\t"""Avoid MariaDB-only DATE_FORMAT(); read from the pg_index catalog instead."""\n\treturn 1\n' - ) - - def test_qb_is_clean(self): - self._assert_clean("frappe.qb.from_(sle).select(CombineDatetime(sle.posting_date, sle.posting_time))") - - def test_pg_ok_suppresses(self): - self._assert_clean('frappe.db.sql( # pg-ok\n\t"show index from `tabItem`"\n)') - - -if __name__ == "__main__": - unittest.main() From b2ee8cb1b9fe73cbbdb7400e543a8f6512c1ab97 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 21 Jun 2026 17:10:34 +0530 Subject: [PATCH 3/3] ci(postgres): fix semgrep + two review findings in the checker - semgrep: annotate the source-reading open() with # nosemgrep for the frappe-security-file-traversal rule (dev-only lint tool; path comes from pre-commit, not user input). - bool-scan: only inspect the field *value* arg (db_set args[1]/dict args[0]; set_value args[3]/dict args[2]) so a positional update_modified=False (e.g. db_set('f', 0, False)) no longer false-positives. - # pg-ok: also honour the annotation on a multi-line call's closing paren line (scan one line past the node's end). --- .github/helper/postgres_compat.py | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/.github/helper/postgres_compat.py b/.github/helper/postgres_compat.py index e5fc0578c7a..a02e82eb953 100755 --- a/.github/helper/postgres_compat.py +++ b/.github/helper/postgres_compat.py @@ -92,10 +92,10 @@ class Visitor(ast.NodeVisitor): def _ignored(self, node: ast.AST) -> bool: start = getattr(node, "lineno", 1) end = getattr(node, "end_lineno", start) or start - # honour `# pg-ok` anywhere on the node's line span, or on the line just above - # (the enclosing call, e.g. `frappe.db.sql( # pg-ok`). + # honour `# pg-ok` anywhere on the node's line span, the line just above (the enclosing + # call, e.g. `frappe.db.sql( # pg-ok`), or the line just below (a multi-line call's `) # pg-ok`). lo = max(0, start - 2) - return any(IGNORE in self.lines[i] for i in range(lo, min(end, len(self.lines)))) + return any(IGNORE in self.lines[i] for i in range(lo, min(end + 1, len(self.lines)))) def _flag(self, node: ast.AST, msg: str) -> None: if not self._ignored(node): @@ -134,15 +134,25 @@ class Visitor(ast.NodeVisitor): if name == "get" and node.args and isinstance(node.args[0], ast.Constant) and node.args[0].value in MYSQL_RESULT_KEYS: self._flag(node, f'"{node.args[0].value}" is a MySQL SHOW INDEX result key -> use frappe.db.has_index()/get_column_index()') - # set_value(..., True) / db_set("field", True) on a Check (int) column + # set_value(..., True) / db_set("field", True) on a Check (int) column. + # Only the field *value* arg carries bool->smallint risk — NOT trailing flags like + # update_modified. db_set(field, value, update_modified, ...) -> value at args[1] (or a dict + # at args[0]); set_value(dt, dn, field, value, ...) -> value at args[3] (or a dict at args[2]). if name in SET_BOOL_FUNCS: - for a in node.args: + value_idx, dict_idx = (1, 0) if name == "db_set" else (3, 2) + dict_arg = ( + node.args[dict_idx] + if len(node.args) > dict_idx and isinstance(node.args[dict_idx], ast.Dict) + else None + ) + if dict_arg is not None: + for v in dict_arg.values: + if isinstance(v, ast.Constant) and isinstance(v.value, bool): + self._flag(node, f"{name}(...) sets an int/Check column with a bool in a dict -> pass 1/0 (Postgres rejects bool->smallint)") + elif len(node.args) > value_idx: + a = node.args[value_idx] if isinstance(a, ast.Constant) and isinstance(a.value, bool): self._flag(node, f"{name}(..., {a.value}) sets an int/Check column with a bool -> pass 1/0 (Postgres rejects bool->smallint)") - elif isinstance(a, ast.Dict): - for v in a.values: - if isinstance(v, ast.Constant) and isinstance(v.value, bool): - self._flag(node, f"{name}(...) sets an int/Check column with a bool in a dict -> pass 1/0 (Postgres rejects bool->smallint)") self.generic_visit(node) @@ -155,6 +165,7 @@ class Visitor(ast.NodeVisitor): def check_file(path: str) -> list[str]: try: + # nosemgrep: frappe-semgrep-rules.rules.security.frappe-security-file-traversal -- dev-only lint tool; `path` is a source file supplied by pre-commit, not user input src = open(path, encoding="utf-8").read() except (OSError, UnicodeDecodeError): return []