From 53ce20956d9085c5d9a8e61770886863a38ac08c Mon Sep 17 00:00:00 2001 From: gitea-impl-agent Date: Wed, 14 Oct 2026 04:44:28 +0000 Subject: [PATCH] Reject malformed persisted settlement header evidence Preserve safe shared error envelopes and read-only legacy defaults. Add exact/latest evidence-integrity regressions for service, gateway and MCP consumers. --- .../application/monthly_settlement_service.py | 2 + tests/test_comprehensive_settlement.py | 75 +++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/jarca_ledger_mcp/application/monthly_settlement_service.py b/jarca_ledger_mcp/application/monthly_settlement_service.py index 0c083e2..de90063 100644 --- a/jarca_ledger_mcp/application/monthly_settlement_service.py +++ b/jarca_ledger_mcp/application/monthly_settlement_service.py @@ -205,6 +205,8 @@ def get_monthly_settlement_snapshot(month: str = "", run_id: str = "", db_path: if selected_month and basis_date[:7] != selected_month: return fail("validation_error", "run does not match settlement month") legacy = True + if metadata.get("evidence_integrity_error"): + return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=rid) keys = set(run.keys()) if run else set() def field(name): return run[name] if run and name in keys else None from .settlement_cashflow_service import _AGGREGATION_VERSION diff --git a/tests/test_comprehensive_settlement.py b/tests/test_comprehensive_settlement.py index 4cc8c7b..d06c416 100644 --- a/tests/test_comprehensive_settlement.py +++ b/tests/test_comprehensive_settlement.py @@ -610,3 +610,78 @@ def test_saved_evidence_privacy_contract_filters_keys_not_allowlisted_scalar_val contract = (Path(__file__).resolve().parents[1] / "docs/interface-contract.md").read_text() assert "allowlisted scalar values are preserved" in contract assert "not a content-based secret redactor" in contract + + +@pytest.mark.parametrize("field", ["excluded_evidence", "carried_evidence", "applied_evidence"]) +@pytest.mark.parametrize("corrupt", ["malformed_json", "wrong_json_object"]) +@pytest.mark.parametrize("lookup", ["exact", "latest"]) +def test_saved_reader_rejects_corrupt_header_evidence_without_writes(tmp_path, monkeypatch, field, corrupt, lookup): + path = dbfile(tmp_path) + saved = ok(close_draft(path, new_draft(path))) + marker = "SYNTHETIC_CORRUPT_EVIDENCE_MARKER" + payload = "{" + marker if corrupt == "malformed_json" else json.dumps({"private": marker}) + with sqlite3.connect(path) as conn: + columns = {row[1] for row in conn.execute("PRAGMA table_info(asset_snapshot_runs)")} + assert field in columns, f"required settlement evidence column missing: {field}" + conn.execute(f"UPDATE asset_snapshot_runs SET {field}=? WHERE id=?", (payload, saved["run_id"])) + conn.commit() + before = list(conn.iterdump()) + + query = {"run_id": saved["run_id"]} if lookup == "exact" else {"month": "2026-09"} + result = service.get_monthly_settlement_snapshot(db_path=path, **query) + expected = {"success": False, "error_code": "data_integrity_error", "error": "snapshot evidence is malformed", "snapshot_run_id": saved["run_id"]} + assert result == expected + from jarca_ledger_mcp.adapters.streamlit import settlement_gateway + gateway = settlement_gateway.get_monthly_settlement_snapshot(db_path=path, **query) + assert gateway == result + from jarca_ledger_mcp.adapters.mcp import server + monkeypatch.setenv("JARCA_LEDGER_DB_PATH", path) + tool = server.mcp._tool_manager._tools["get_monthly_settlement_snapshot"] + assert tool.fn(**query) == result + assert service.get_asset_snapshot_report("2026-09-30", db_path=path) == expected + assert marker not in json.dumps(result) + assert not ({"run_id", "totals", "income_total", "source_snapshot", "cashflow_items", "excluded_items", "carried_items", "applied_evidence"} & result.keys()) + with sqlite3.connect(path) as conn: + assert list(conn.iterdump()) == before + + +@pytest.mark.parametrize("field", ["excluded_evidence", "carried_evidence", "applied_evidence"]) +def test_saved_reader_accepts_empty_header_evidence_without_changing_legacy_api_shapes(tmp_path, field): + path = dbfile(tmp_path) + saved = ok(close_draft(path, new_draft(path))) + with sqlite3.connect(path) as conn: + columns = {row[1] for row in conn.execute("PRAGMA table_info(asset_snapshot_runs)")} + assert field in columns, f"required settlement evidence column missing: {field}" + conn.execute(f"UPDATE asset_snapshot_runs SET {field}='[]' WHERE id=?", (saved["run_id"],)) + conn.commit() + with sqlite3.connect(path) as conn: + before = list(conn.iterdump()) + exact = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path)) + latest = ok(service.get_monthly_settlement_snapshot(month="2026-09", db_path=path)) + assert exact == latest + public_field = {"excluded_evidence": "excluded_items", "carried_evidence": "carried_items", "applied_evidence": "applied_evidence"}[field] + assert exact[public_field] == [] + assert exact["source"] == "settlement_snapshot" and exact["is_comprehensive"] is True + cashflow = ok(service.get_monthly_cashflow_report("2026-09", db_path=path)) + assert "run_id" not in cashflow and "totals" not in cashflow and "cashflow_items" not in cashflow + with sqlite3.connect(path) as conn: + assert list(conn.iterdump()) == before + + +def test_schema_eleven_legacy_reader_returns_empty_evidence_without_writes(tmp_path): + from test_legacy_snapshot_migration import _legacy_closed_db + path, _ = _legacy_closed_db(tmp_path) + with sqlite3.connect(path) as conn: + assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0 + before = list(conn.iterdump()) + exact = ok(service.get_monthly_settlement_snapshot(run_id="legacy-run-correction", db_path=path)) + latest = ok(service.get_monthly_settlement_snapshot(month="2026-06", db_path=path)) + # Headerless legacy correction metadata already differs by lookup mode; + # this rework preserves that separate existing behavior rather than changing it. + for result in (exact, latest): + assert result["legacy"] is True and result["is_comprehensive"] is False + assert result["source_snapshot"] == {} + assert result["excluded_items"] == [] and result["carried_items"] == [] and result["applied_evidence"] == [] + assert all(value is None for value in result["totals"].values()) + with sqlite3.connect(path) as conn: + assert list(conn.iterdump()) == before