jiang
fa188af0b1
Reorganize WNDB by responsibility and remove legacy scheme endpoints.\n\nRoute analysis and time-series access through project pools, preserve transactional realtime replacement, and refresh GIS materialized views after writes.\n\nAdd database architecture documentation, live pooling coverage, API contract updates, and executable container verification.\n\nBREAKING CHANGE: legacy scheme APIs and flat app.native.wndb module imports are removed.
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
import ast
|
|
from pathlib import Path
|
|
|
|
from app.infra.db.postgresql import scada_assets
|
|
from app.native.wndb.core.database import ChangeSet, sql_literal
|
|
from app.native.wndb.model import controls, junctions, patterns
|
|
|
|
|
|
def test_get_junction_binds_untrusted_identifier(monkeypatch) -> None:
|
|
calls: list[tuple[str, str, tuple[str, ...]]] = []
|
|
malicious_id = "J-1'; DELETE FROM junctions; --"
|
|
|
|
def fake_try_read(name, statement, params):
|
|
calls.append((name, statement, params))
|
|
return None
|
|
|
|
monkeypatch.setattr(junctions, "try_read", fake_try_read)
|
|
|
|
assert junctions.get_junction("project_a", malicious_id) == {}
|
|
assert len(calls) == 1
|
|
name, statement, params = calls[0]
|
|
assert name == "project_a"
|
|
assert "network.junctions" in statement
|
|
assert "WHERE n.id = %s" in statement
|
|
assert malicious_id not in statement
|
|
assert params == (malicious_id,)
|
|
|
|
|
|
def test_get_all_junctions_reads_materialized_view(monkeypatch) -> None:
|
|
statements: list[str] = []
|
|
|
|
def fake_read_all(_name, statement):
|
|
statements.append(statement)
|
|
return []
|
|
|
|
monkeypatch.setattr(junctions, "read_all", fake_read_all)
|
|
monkeypatch.setattr(junctions, "get_all_node_links", lambda _name: {})
|
|
|
|
assert junctions.get_all_junctions("project_a") == []
|
|
assert "FROM gis.junctions" in statements[0]
|
|
|
|
|
|
def test_get_all_scada_info_reads_materialized_view(monkeypatch) -> None:
|
|
statements: list[str] = []
|
|
|
|
def fake_read_all(_name, statement):
|
|
statements.append(statement)
|
|
return []
|
|
|
|
monkeypatch.setattr(scada_assets, "read_all", fake_read_all)
|
|
|
|
assert scada_assets.get_all_scada_info("project_a") == []
|
|
assert "FROM gis.scada_devices" in statements[0]
|
|
|
|
|
|
def test_sql_literal_keeps_attacker_text_inside_one_postgres_literal() -> None:
|
|
malicious = "x'); DROP SCHEMA network CASCADE; --"
|
|
|
|
assert sql_literal("O'Brien") == "'O''Brien'"
|
|
assert sql_literal(malicious) == "'x''); DROP SCHEMA network CASCADE; --'"
|
|
assert sql_literal(None) == "NULL"
|
|
|
|
|
|
def test_pattern_command_quotes_malicious_identifier() -> None:
|
|
malicious = "x'); DROP SCHEMA network CASCADE; --"
|
|
change = ChangeSet({"id": malicious, "factors": [1.0]})
|
|
|
|
command = patterns._add_pattern("project_a", change).sql
|
|
|
|
assert f"values ({sql_literal(malicious)})" in command
|
|
assert "values ('x'); DROP SCHEMA" not in command
|
|
|
|
|
|
def test_inp_control_quotes_embedded_apostrophe() -> None:
|
|
command = controls.inp_in_control("LINK P-1 STATUS 'OPEN'; DELETE")
|
|
|
|
assert "''OPEN''" in command
|
|
assert "STATUS 'OPEN'; DELETE')" not in command
|
|
|
|
|
|
def test_wndb_sql_fstrings_do_not_quote_formatted_values_directly() -> None:
|
|
root = Path(__file__).resolve().parents[2] / "app" / "native" / "wndb"
|
|
violations: list[str] = []
|
|
for path in root.rglob("*.py"):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.JoinedStr):
|
|
continue
|
|
static_text = "".join(
|
|
value.value
|
|
for value in node.values
|
|
if isinstance(value, ast.Constant) and isinstance(value.value, str)
|
|
).lower()
|
|
if not any(
|
|
keyword in static_text
|
|
for keyword in ("select ", "insert into", "update ", "delete from")
|
|
):
|
|
continue
|
|
for index, value in enumerate(node.values):
|
|
if not isinstance(value, ast.FormattedValue):
|
|
continue
|
|
before = node.values[index - 1] if index else None
|
|
after = node.values[index + 1] if index + 1 < len(node.values) else None
|
|
left_quote = (
|
|
isinstance(before, ast.Constant)
|
|
and isinstance(before.value, str)
|
|
and before.value.endswith("'")
|
|
)
|
|
right_quote = (
|
|
isinstance(after, ast.Constant)
|
|
and isinstance(after.value, str)
|
|
and after.value.startswith("'")
|
|
)
|
|
if left_quote and right_quote:
|
|
violations.append(f"{path.name}:{node.lineno}")
|
|
|
|
assert violations == []
|