GTI_GERENCIADOR_FISCAL/fiscal/tests/test_catalog.py

"""Mission 00D Phase 8: prove arbitrary SQL cannot reach execution through
the closed catalog API, and that legitimate Firebird constructs used by
the project's own internal queries are not rejected.

``FirebirdCatalog`` has no ``execute``/``query``/``raw_execute`` method at
all, so most classic SQL-injection payloads simply have nowhere to go --
these tests prove that structurally (no method exists to try them on) and
prove that the one dynamic operation, ``sample_relation_rows``, rejects
every malicious relation/column value because it is validated against the
catalog itself, never interpreted as SQL text.
"""

import pytest

from gti_discovery.catalog import MAX_SAMPLE_ROWS, FirebirdCatalog, _quote_identifier
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _register_approved_adapter
from gti_discovery.read_only import validate_read_only


class _RecordingAdapter:
    read_only = True

    def __init__(self, rows_by_call=None):
        self.calls: list[tuple[str, tuple]] = []
        self._rows_by_call = rows_by_call or {}

    def execute(self, sql, params=()):
        self.calls.append((sql, params))
        return self._rows_by_call.get(len(self.calls) - 1, [])


_register_approved_adapter(_RecordingAdapter, _ADAPTER_REGISTRATION_TOKEN)


def _catalog_with_relations(relations, fields_by_relation):
    """A FirebirdCatalog whose relations/fields queries return fixture rows."""
    relation_rows = [{"RELATION_NAME": name, "RELATION_ID": i, "SYSTEM_FLAG": 0} for i, name in enumerate(relations)]
    field_rows = [
        {"RELATION_NAME": relation, "FIELD_NAME": field, "FIELD_POSITION": i}
        for relation, fields in fields_by_relation.items()
        for i, field in enumerate(fields)
    ]
    adapter = _RecordingAdapter({0: relation_rows, 1: field_rows})
    catalog = FirebirdCatalog(adapter)
    return catalog, adapter


# --- Phase 8: closed API surface -----------------------------------------

def test_no_arbitrary_sql_method_exists_at_all():
    for forbidden in ("execute", "query", "raw_execute", "execute_statement", "run_sql", "run_query"):
        assert not hasattr(FirebirdCatalog, forbidden)


# --- Phase 8: relation/column must be catalog-known, never free text -----

@pytest.mark.parametrize("malicious_relation", [
    "PRODUCTS\"; DROP TABLE X; --",
    "PRODUCTS' OR '1'='1",
    'PRODUCTS" ("EAN"',
    "PRODUCTS/*comment*/",
    "PRODUCTS UNION SELECT * FROM RDB$USERS",
    "SOME_PROCEDURE(1,2,3)",
    '"SOME_UDF"(1)',
    "PRODUCTS; SELECT * FROM RDB$USERS",
    "(SELECT PASSWORD FROM USERS)",
    "PRODUCTS AS X UNION SELECT 1",
])
def test_sample_relation_rows_rejects_unknown_or_malicious_relation(malicious_relation):
    catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
    with pytest.raises(ValueError):
        catalog.sample_relation_rows(malicious_relation, ["ID"], 1)
    # never reached the driver with a sampling query (only the relations
    # bootstrap call, if any, is allowed to have happened)
    sampling_calls = [c for c in adapter.calls if "FIRST" in c[0].upper()]
    assert sampling_calls == []


@pytest.mark.parametrize("malicious_column", [
    "EAN\"; DROP TABLE X; --",
    "EAN' OR '1'='1",
    "EAN, PASSWORD",
    "EAN) UNION SELECT PASSWORD FROM USERS --",
    "COUNT(*)",
    "(SELECT 1)",
])
def test_sample_relation_rows_rejects_unknown_or_malicious_column(malicious_column):
    catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
    with pytest.raises(ValueError):
        catalog.sample_relation_rows("PRODUCTS", [malicious_column], 1)
    sampling_calls = [c for c in adapter.calls if "FIRST" in c[0].upper()]
    assert sampling_calls == []


def test_sample_relation_rows_accepts_legitimate_relation_and_columns():
    catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
    catalog.sample_relation_rows("PRODUCTS", ["ID", "EAN"], 5)
    sql, params = adapter.calls[-1]
    assert sql == 'SELECT FIRST 5 "ID", "EAN" FROM "PRODUCTS"'
    assert params == ()


@pytest.mark.parametrize("limit", [0, -1, 51, 10 ** 9, "5", 5.0, True])
def test_sample_relation_rows_rejects_bad_limit(limit):
    catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
    with pytest.raises(ValueError):
        catalog.sample_relation_rows("PRODUCTS", ["ID"], limit)


def test_sample_relation_rows_limit_ceiling_is_enforced():
    catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
    catalog.sample_relation_rows("PRODUCTS", ["ID"], MAX_SAMPLE_ROWS)
    with pytest.raises(ValueError):
        catalog.sample_relation_rows("PRODUCTS", ["ID"], MAX_SAMPLE_ROWS + 1)


def test_sample_relation_rows_never_accepts_empty_columns():
    catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
    with pytest.raises(ValueError):
        catalog.sample_relation_rows("PRODUCTS", [], 1)


def test_quote_identifier_doubles_embedded_quotes():
    assert _quote_identifier('WEIRD"NAME') == '"WEIRD""NAME"'


# --- Phase 5: procedures are metadata-only, never executable -------------

def test_get_procedures_returns_metadata_never_a_callable():
    adapter = _RecordingAdapter({0: [{"PROCEDURE_NAME": "SP_RECALC_STOCK", "PROCEDURE_ID": 1}]})
    catalog = FirebirdCatalog(adapter)
    procedures = catalog.get_procedures()
    assert procedures == [{"PROCEDURE_NAME": "SP_RECALC_STOCK", "PROCEDURE_ID": 1}]
    # a procedure name is never itself a relation, so it can never be
    # threaded into sample_relation_rows
    with pytest.raises(ValueError):
        catalog.sample_relation_rows("SP_RECALC_STOCK", ["ID"], 1)


# --- read_only.py kept as an offline tool: bypass tests still apply ------

@pytest.mark.parametrize("payload", [
    '"SOME_PROCEDURE"(1,2,3)',
    'SELECT * FROM "SOME_PROC"(1)',
    "SELECT * FROM PRODUCTS; DROP TABLE PRODUCTS",
    "SELECT * FROM PRODUCTS -- comment",
    "SELECT * FROM PRODUCTS /* comment */",
    "EXECUTE PROCEDURE SP_DANGEROUS",
    "SELECT SOME_UDF(1) FROM RDB$DATABASE",
    "SELECT (SELECT PASSWORD FROM USERS) FROM PRODUCTS",
])
def test_offline_validator_still_rejects_every_bypass_payload(payload):
    with pytest.raises(ValueError):
        validate_read_only(payload)


@pytest.mark.parametrize("legit", [
    "SELECT CAST(X AS INTEGER) FROM PRODUCTS",
    "SELECT COUNT(*) FROM PRODUCTS",
    "SELECT SUM(QTY), MIN(QTY), MAX(QTY), AVG(QTY) FROM STOCK GROUP BY PRODUCT_ID",
    "SELECT COALESCE(X, 0) FROM PRODUCTS",
    "WITH x AS (SELECT 1 AS n FROM RDB$DATABASE) SELECT n FROM x",
    # UNION of two plain read-only SELECTs has no side effect -- the
    # validator correctly accepts it rather than over-blocking.
    "SELECT ID FROM A UNION SELECT ID FROM B",
    # A quoted alias containing reserved-word-looking text or a semicolon
    # is inert -- it can never execute anything, it is just a name; the
    # masking correctly treats quoted content as opaque, not as SQL.
    "SELECT * FROM PRODUCTS AS \"'; DROP TABLE X; --\"",
    # A string literal containing reserved words is data, not SQL --
    # masking must not accidentally treat it as a statement.
    "SELECT 'reserved SELECT INSERT DELETE word inside a string' FROM PRODUCTS",
])
def test_offline_validator_accepts_legitimate_firebird_constructs(legit):
    assert validate_read_only(legit) == legit


def test_offline_validator_is_not_wired_into_the_live_catalog_path():
    """Mission 00D: read_only.validate_read_only must not be imported by
    catalog.py -- it is an optional offline tool, never the runtime gate."""
    import gti_discovery.catalog as catalog_module
    assert "validate_read_only" not in dir(catalog_module)
    assert "read_only" not in catalog_module.__dict__