184 lines
7.6 KiB
Python
184 lines
7.6 KiB
Python
"""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__
|