GTI_GERENCIADOR_FISCAL/src/gti_discovery/read_only.py

"""Offline, auxiliary SQL sanity checker -- NOT the security boundary.

IMPORTANT (Mission 00D architectural decision): this module is kept only
as a defense-in-depth / offline diagnostic tool. It is deliberately no
longer wired into the live read path -- ``gti_discovery.catalog`` never
calls ``validate_read_only`` and exposes no ``execute(sql)``-shaped API
at all, so arbitrary caller-provided SQL text can never reach the
Firebird driver in the first place. A regex/lexical SQL "blacklist" can
never be a complete or reliable security boundary against a real SQL
grammar (this file's own history proves it -- see git log / Mission 00D
report for the quoted-identifier bypass and the CAST/aggregate false
rejections that motivated the redesign). It is retained here, fixed and
tested, purely so it can still be used as an optional, non-authoritative
sanity check if some future caller-facing tool ever needs one -- it must
never again be trusted as the sole gate before execution.
"""

import re
from typing import Any


_SIDE_EFFECT_SQL = re.compile(
    r"\b(?:INSERT|UPDATE|DELETE|CREATE|ALTER|DROP|TRUNCATE|EXECUTE)\b",
    re.IGNORECASE,
)
_FIRST_KEYWORD = re.compile(r"\A(SELECT|WITH)\b", re.IGNORECASE)
_SIDE_EFFECTING_SELECT = re.compile(
    r"(?:\bFROM\s+(?:[A-Z][\w$]*\s*\.\s*)*[A-Z][\w$]*\s*\(|"
    r"\bGEN_ID\s*\(|\bNEXT\s+VALUE\s+FOR\b|\bWITH\s+LOCK\b|"
    r"\bFOR\s+UPDATE\b)",
    re.IGNORECASE,
)
_CALLABLE = re.compile(
    r"\b(?P<qualified>(?:[A-Z][\w$]*\s*\.\s*)*[A-Z][\w$]*)\s*\(",
    re.IGNORECASE,
)
# A bare quoted identifier immediately followed by "(" -- e.g.
# "SOME_PROCEDURE"(1,2) -- used to be invisible to _CALLABLE once quoted
# text was masked to spaces by _scan_sql/_normalize_outside_quotes. That
# was the real bypass: masking hid the callable name instead of still
# flagging "there was a callable here". This pattern reads the ORIGINAL
# (unmasked) SQL to catch that shape explicitly.
_QUOTED_CALLABLE = re.compile(r'"(?:[^"]|"")*"\s*\(')
_DETERMINISTIC_CALLABLES = {
    "ABS", "AS", "AVG", "CAST", "CEILING", "CHAR_LENGTH", "COALESCE", "COUNT",
    "EXTRACT", "FLOOR", "IIF", "LOWER", "MAX", "MIN", "NULLIF", "ROUND",
    "SUBSTRING", "SUM", "TRIM", "UPPER",
}
def _scan_sql(sql: str) -> tuple[str, int]:
    """Return SQL with quoted literals/identifiers masked and semicolon count."""
    masked: list[str] = []
    semicolons = 0
    index = 0
    while index < len(sql):
        char = sql[index]
        if char in ("'", '"'):
            quote = char
            masked.append(" ")
            index += 1
            while index < len(sql):
                masked.append(" ")
                if sql[index] == quote:
                    if index + 1 < len(sql) and sql[index + 1] == quote:
                        masked.append(" ")
                        index += 2
                        continue
                    index += 1
                    break
                index += 1
            else:
                raise ValueError("unterminated quoted SQL value")
            continue
        if sql.startswith("--", index):
            raise ValueError("SQL comments are not allowed")
        if sql.startswith("/*", index):
            raise ValueError("SQL comments are not allowed")
        if char == ";":
            semicolons += 1
        masked.append(char)
        index += 1
    return "".join(masked), semicolons


def _normalize_outside_quotes(sql: str) -> str:
    normalized: list[str] = []
    pending_space = False
    index = 0
    while index < len(sql):
        char = sql[index]
        if char in ("'", '"'):
            if pending_space and normalized:
                normalized.append(" ")
            pending_space = False
            quote = char
            normalized.append(char)
            index += 1
            while index < len(sql):
                normalized.append(sql[index])
                if sql[index] == quote:
                    if index + 1 < len(sql) and sql[index + 1] == quote:
                        normalized.append(sql[index + 1])
                        index += 2
                        continue
                    index += 1
                    break
                index += 1
            continue
        if char.isspace():
            if normalized:
                pending_space = True
        else:
            if pending_space and normalized:
                normalized.append(" ")
            pending_space = False
            normalized.append(char)
        index += 1
    return "".join(normalized).rstrip()


def validate_read_only(sql: str) -> str:
    """Normalize and validate that *sql* is one safe, read-only query."""
    if not isinstance(sql, str):
        raise ValueError("SQL must be a string")

    candidate = sql.strip()
    if candidate.endswith(";"):
        candidate = candidate[:-1].rstrip()
    if not candidate:
        raise ValueError("SQL query cannot be empty")

    masked, semicolons = _scan_sql(candidate)
    if semicolons:
        raise ValueError("multiple SQL statements are not allowed")
    normalized = _normalize_outside_quotes(candidate)
    normalized_masked = " ".join(masked.split())
    if not _FIRST_KEYWORD.match(normalized_masked):
        raise ValueError("only SELECT and WITH queries are allowed")
    if _SIDE_EFFECT_SQL.search(normalized_masked):
        raise ValueError("SQL has a prohibited side effect")
    if _SIDE_EFFECTING_SELECT.search(normalized_masked):
        raise ValueError("SQL contains a known side-effecting construct")
    # A quoted identifier used as a callable (e.g. "SOME_PROC"(1)) is
    # invisible to _CALLABLE once masking hides its text -- check the
    # unmasked, normalized SQL directly so it can never bypass the
    # callable allowlist just by being quoted (the original bypass this
    # module was rewritten to close, see module docstring).
    if _QUOTED_CALLABLE.search(normalized):
        raise ValueError("SQL contains an unapproved quoted callable")
    for match in _CALLABLE.finditer(normalized_masked):
        name = match.group("qualified").rsplit(".", 1)[-1].strip().upper()
        if name not in _DETERMINISTIC_CALLABLES:
            raise ValueError("SQL contains an unapproved callable")
    return normalized