from __future__ import annotations import csv import json import sqlite3 from pathlib import Path from jarca_ledger_mcp import service from jarca_ledger_mcp.provenance_policy import ( EXPOSURE_LEVEL_DEBUG_GATED, EXPOSURE_LEVEL_MINIMAL, EXPOSURE_LEVEL_OPERATOR_SUMMARY, sanitize_provenance_for_exposure, sanitize_provenance, sanitize_provenance_envelope, ) from jarca_ledger_mcp.kb_bank_pdf import KbPdfTransaction from jarca_ledger_mcp.seoulpay_xlsx import SeoulPayTransaction from jarca_ledger_mcp.toss_bank_pdf import TossBankTransaction from jarca_ledger_mcp.toss_securities_pdf import TossSecuritiesTransaction SECRET_SOURCE_PATH = "/very/secret/home/alice/Documents/private/statement.pdf" SOURCE_HASH = "abcdef1234567890fedcba0987654321abcdef1234567890fedcba0987654321" def _expected_filename(file_path: str, source_hash: str = SOURCE_HASH) -> str: return f"{Path(file_path).name}#{source_hash[:12]}" def _stored_batch_and_payloads(db_path: str) -> tuple[str, list[dict[str, object]]]: with sqlite3.connect(db_path) as conn: batch_filename = conn.execute("SELECT filename FROM import_batches ORDER BY imported_at DESC LIMIT 1").fetchone()[0] payload_rows = conn.execute("SELECT raw_payload FROM transactions ORDER BY created_at").fetchall() return batch_filename, [json.loads(row[0]) for row in payload_rows] def test_preview_ledger_csv_import_redacts_full_path_and_raw_columns(tmp_path): csv_path = tmp_path / "import-preview.csv" with csv_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=["date", "amount", "merchant", "secret_note", "local_path"]) writer.writeheader() writer.writerow( { "date": "2026-06-18", "amount": "42000", "merchant": "테스트상점", "secret_note": "do not leak this raw note", "local_path": "/Users/alice/Desktop/private.csv", } ) result = service.preview_ledger_csv_import(str(csv_path)) assert result["success"] is True assert result["filename"] == _expected_filename(str(csv_path), result["source_hash"]) assert result["preview"] == [{"row_number": 1, "date": "2026-06-18", "amount": "42000", "merchant": "테스트상점"}] rendered = json.dumps(result, ensure_ascii=False) assert str(tmp_path) not in rendered assert "secret_note" not in rendered assert "local_path" not in rendered def test_preview_and_import_kb_bank_pdf_sanitize_path_and_payload(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": SECRET_SOURCE_PATH, "source_hash": SOURCE_HASH, "page_count": 4, "metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456}, "transactions": [ KbPdfTransaction( page=7, transaction_datetime="2026-06-18T09:00:00", transaction_date="2026-06-18", raw_summary="적요-민감메모", counterparty="테스트상대", withdrawal=42000, deposit=0, balance_after=123456, remittance_memo="숨기고싶은메모", branch="강남지점", ) ], "validation": {"ok": True, "errors": []}, } monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed) preview = service.preview_kb_bank_pdf_import("ignored.pdf", "pw", db_path=db_path) imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path) assert preview["success"] is True assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH) assert SECRET_SOURCE_PATH not in json.dumps(preview, ensure_ascii=False) assert imported["success"] is True batch_filename, payloads = _stored_batch_and_payloads(db_path) assert batch_filename == _expected_filename(SECRET_SOURCE_PATH) assert payloads == [{"page": 7}] def test_preview_and_import_seoulpay_xlsx_sanitize_path_and_payload(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": "/very/secret/home/alice/Documents/private/seoulpay.xlsx", "source_hash": SOURCE_HASH, "sheet_name": "결제내역", "metadata": {"holder_name": "홍길동", "period": "20260601~20260620"}, "transactions": [ SeoulPayTransaction( row_index=11, channel="서울페이", transaction_no="TRX-001", transaction_datetime="2026-06-18 18:27:46", transaction_date="2026-06-18", merchant="땡겨요", business_no="2028102637", merchant_id="AF2402815444", transaction_kind="온라인 결제", payment_kind="결제", amount=42000, supply_amount=38182, service_fee=0, vat=3818, tax_status="과세", ) ], "validation": {"ok": True, "parsed_rows": 1, "errors": []}, } monkeypatch.setattr(service, "parse_seoulpay_xlsx", lambda file_path, password="": parsed) preview = service.preview_seoulpay_xlsx_import("ignored.xlsx", db_path=db_path) imported = service.import_seoulpay_xlsx("ignored.xlsx", db_path=db_path) assert preview["success"] is True assert preview["file_path"] == _expected_filename(parsed["file_path"]) assert parsed["file_path"] not in json.dumps(preview, ensure_ascii=False) assert imported["success"] is True batch_filename, payloads = _stored_batch_and_payloads(db_path) assert batch_filename == _expected_filename(parsed["file_path"]) assert payloads == [{"row_index": 11}] def test_preview_and_import_toss_bank_pdf_sanitize_path_and_payload(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": SECRET_SOURCE_PATH, "source_hash": SOURCE_HASH, "page_count": 3, "metadata": {"holder_name": "홍길동", "account_name": "토스 통장", "account_number_masked": "100-1234-5678", "final_balance": 820000}, "transactions": [ TossBankTransaction( row_index=9, transaction_datetime="2026-06-18 09:00:00", transaction_date="2026-06-18", action="이체", counterparty="테스트상대", amount=12000, type="expense", balance_after=820000, raw_summary="상대방 요약", raw_lines=[SECRET_SOURCE_PATH, "ssh private key"], ) ], "validation": {"ok": True, "parsed_rows": 1, "errors": []}, } monkeypatch.setattr(service, "parse_toss_bank_pdf", lambda file_path: parsed) preview = service.preview_toss_bank_pdf_import("ignored.pdf", db_path=db_path) imported = service.import_toss_bank_pdf("ignored.pdf", db_path=db_path) assert preview["success"] is True assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH) rendered = json.dumps(preview, ensure_ascii=False) assert SECRET_SOURCE_PATH not in rendered assert "raw_lines" not in rendered assert imported["success"] is True batch_filename, payloads = _stored_batch_and_payloads(db_path) assert batch_filename == _expected_filename(SECRET_SOURCE_PATH) assert payloads == [{"row_index": 9}] def test_preview_and_import_toss_securities_pdf_sanitize_path_and_payload(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": SECRET_SOURCE_PATH, "source_hash": SOURCE_HASH, "page_count": 5, "metadata": {"holder_name": "홍길동", "account_name": "토스증권 계좌", "account_number_masked": "123-01-048829"}, "transactions": [ TossSecuritiesTransaction( row_index=5, section="USD", transaction_date="2026-06-18", action="외화증권배당금입금", name="ACME(A123)", code="A123", amount=4300, type="income", quantity=1.5, unit_price=25.2, fee=120, tax=33, balance_after=910000, raw_numbers=["1.5", "25.2", "4300"], raw_block=[SECRET_SOURCE_PATH, "api_token=secret"], ) ], "validation": {"ok": True, "parsed_rows": 1, "errors": []}, } monkeypatch.setattr(service, "parse_toss_securities_pdf", lambda file_path: parsed) preview = service.preview_toss_securities_pdf_import("ignored.pdf", db_path=db_path) imported = service.import_toss_securities_pdf("ignored.pdf", db_path=db_path) assert preview["success"] is True assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH) rendered = json.dumps(preview, ensure_ascii=False) assert SECRET_SOURCE_PATH not in rendered assert "raw_block" not in rendered assert imported["success"] is True batch_filename, payloads = _stored_batch_and_payloads(db_path) assert batch_filename == _expected_filename(SECRET_SOURCE_PATH) assert payloads == [{"row_index": 5, "section": "USD", "quantity": 1.5, "unit_price": 25.2, "fee": 120, "tax": 33}] def test_sanitize_provenance_by_exposure_levels_filters_sensitive_values(): envelope = { "source": "kb_bank_pdf", "source_hash": SOURCE_HASH, "file_path": SECRET_SOURCE_PATH, "filename": SECRET_SOURCE_PATH, "parser": "kb", "import_batch_id": "batch-1", "external_id": "ext-1", "dedupe_key": "dedupe-1", "row_number": 7, "page": 3, "sheet": "A1", "validation": {"ok": True, "errors": []}, "raw_payload": { "row_index": 7, "page": 3, "action": "transfer", "raw_text": "raw statement text", "password": "p@ss", "counterparty": "김대리", "secret": "should-hide", "transaction_path": "/very/sensitive/path.csv", "raw_document": "full text dump", }, "debug": { "parser_diag": "ok", "raw_text": "should-be-hidden", "raw_payload_dump": "hide-this", "token": "token-001", "message": "keep? no", "notes": ["safe", "path=/tmp/should-not-leak", {"secret": "still-hidden", "message": "ok"}], }, "local_path": "/another/secret/path", } minimal = sanitize_provenance(envelope, exposure_level=EXPOSURE_LEVEL_MINIMAL) assert minimal["source"] == "kb_bank_pdf" assert minimal["parser"] == "kb" assert minimal["source_hash"] == SOURCE_HASH assert minimal["source_hash_prefix"] == SOURCE_HASH[:12] assert "file_path" not in minimal assert "filename" not in minimal assert "raw_payload" not in minimal assert "debug" not in minimal rendered = json.dumps(minimal, ensure_ascii=False) assert SECRET_SOURCE_PATH not in rendered assert "/tmp/should-not-leak" not in rendered assert "password" not in rendered assert "token" not in rendered operator_summary = sanitize_provenance( envelope, exposure_level=EXPOSURE_LEVEL_OPERATOR_SUMMARY, ) assert operator_summary["source"] == "kb_bank_pdf" assert operator_summary["parser"] == "kb" assert operator_summary["filename"] == _expected_filename(SECRET_SOURCE_PATH) assert operator_summary["source_hash_prefix"] == SOURCE_HASH[:12] assert operator_summary["validation"] == {"ok": True, "errors": []} assert "raw_payload" not in operator_summary assert operator_summary["filename"] == _expected_filename(SECRET_SOURCE_PATH) assert "raw_text" not in json.dumps(operator_summary, ensure_ascii=False) debug = sanitize_provenance( envelope, exposure_level=EXPOSURE_LEVEL_DEBUG_GATED, ) assert debug["filename"] == _expected_filename(SECRET_SOURCE_PATH) assert debug["source_hash_prefix"] == SOURCE_HASH[:12] assert "raw_payload" in debug debug_payload = debug["raw_payload"] assert debug_payload == { "row_index": 7, "page": 3, "action": "transfer", "counterparty": "김대리", } assert "password" not in json.dumps(debug, ensure_ascii=False) assert "raw_text" not in json.dumps(debug, ensure_ascii=False) assert "token" not in json.dumps(debug, ensure_ascii=False) assert "raw_document" not in json.dumps(debug, ensure_ascii=False) def test_sanitize_provenance_envelope_alias_matches_core_sanitizer(): envelope = { "source": "toss_securities_pdf", "source_hash": SOURCE_HASH, "file_path": SECRET_SOURCE_PATH, "raw_payload": '{"row_index":1,"page":2,"secret":"x"}', } assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_MINIMAL) == sanitize_provenance( envelope, EXPOSURE_LEVEL_MINIMAL, ) assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY) == sanitize_provenance( envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY, ) assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_DEBUG_GATED) == sanitize_provenance( envelope, EXPOSURE_LEVEL_DEBUG_GATED, ) assert sanitize_provenance_for_exposure(envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY) == sanitize_provenance( envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY, ) def test_list_ledger_import_batches_defaults_to_minimal_provenance(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": SECRET_SOURCE_PATH, "source_hash": SOURCE_HASH, "page_count": 4, "metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456}, "transactions": [ KbPdfTransaction( page=7, transaction_datetime="2026-06-18T09:00:00", transaction_date="2026-06-18", raw_summary="적요-민감메모", counterparty="테스트상대", withdrawal=42000, deposit=0, balance_after=123456, remittance_memo="숨기고싶은메모", branch="강남지점", ) ], "validation": {"ok": True, "errors": []}, } monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed) imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path) assert imported["success"] is True listed = service.list_ledger_import_batches(source="kb_bank_pdf", db_path=db_path) assert listed["success"] is True assert listed["count"] == 1 batch = listed["import_batches"][0] assert batch["memo"] == "" assert batch["filename"] == _expected_filename(SECRET_SOURCE_PATH) assert batch["provenance"] == { "source": "kb_bank_pdf", "parser": "kb_bank_pdf", "import_batch_id": imported["import_batch_id"], "source_hash": SOURCE_HASH, "source_hash_prefix": SOURCE_HASH[:12], } rendered = json.dumps(batch, ensure_ascii=False) assert SECRET_SOURCE_PATH not in rendered assert "account_number_masked" not in rendered assert "validation" not in rendered def test_list_ledger_import_batches_operator_summary_exposes_masked_filename_and_validation(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": "/very/secret/home/alice/Documents/private/seoulpay.xlsx", "source_hash": SOURCE_HASH, "sheet_name": "결제내역", "metadata": {"holder_name": "홍길동", "period": "20260601~20260620"}, "transactions": [ SeoulPayTransaction( row_index=11, channel="서울페이", transaction_no="TRX-001", transaction_datetime="2026-06-18 18:27:46", transaction_date="2026-06-18", merchant="땡겨요", business_no="2028102637", merchant_id="AF2402815444", transaction_kind="온라인 결제", payment_kind="결제", amount=42000, supply_amount=38182, service_fee=0, vat=3818, tax_status="과세", ) ], "validation": {"ok": True, "parsed_rows": 1, "errors": []}, } monkeypatch.setattr(service, "parse_seoulpay_xlsx", lambda file_path, password="": parsed) imported = service.import_seoulpay_xlsx("ignored.xlsx", db_path=db_path) assert imported["success"] is True listed = service.list_ledger_import_batches( source="seoulpay_xlsx", provenance_exposure=EXPOSURE_LEVEL_OPERATOR_SUMMARY, db_path=db_path, ) assert listed["success"] is True batch = listed["import_batches"][0] assert batch["memo"] == "" assert batch["provenance"] == { "source": "seoulpay_xlsx", "parser": "seoulpay_xlsx", "filename": _expected_filename(parsed["file_path"]), "import_batch_id": imported["import_batch_id"], "sheet_name": "결제내역", "source_hash": SOURCE_HASH, "source_hash_prefix": SOURCE_HASH[:12], "validation": {"ok": True, "parsed_rows": 1, "errors": []}, } rendered = json.dumps(batch, ensure_ascii=False) assert parsed["file_path"] not in rendered assert "holder_name" not in rendered assert "period" not in rendered def test_list_ledger_import_batches_requires_explicit_gate_for_debug(monkeypatch, tmp_path): db_path = str(tmp_path / "ledger.sqlite3") parsed = { "file_path": SECRET_SOURCE_PATH, "source_hash": SOURCE_HASH, "page_count": 1, "metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456}, "validation": {"ok": True, "errors": []}, "transactions": [ KbPdfTransaction( page=7, transaction_datetime="2026-06-18T09:00:00", transaction_date="2026-06-18", raw_summary="적요-민감메모", counterparty="테스트상대", withdrawal=42000, deposit=0, balance_after=123456, remittance_memo="숨기고싶은메모", branch="강남지점", ) ], } monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed) imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path) assert imported["success"] is True denied = service.list_ledger_import_batches( source="kb_bank_pdf", provenance_exposure=EXPOSURE_LEVEL_DEBUG_GATED, db_path=db_path, ) assert denied == { "success": False, "error_code": "forbidden", "error": "debug-gated provenance requires explicit debug_provenance_gate=True", } allowed = service.list_ledger_import_batches( source="kb_bank_pdf", provenance_exposure=EXPOSURE_LEVEL_DEBUG_GATED, debug_provenance_gate=True, db_path=db_path, ) assert allowed["success"] is True assert allowed["import_batches"][0]["provenance"]["filename"] == _expected_filename(SECRET_SOURCE_PATH) assert "debug" not in allowed["import_batches"][0]["provenance"] def test_mcp_list_ledger_import_batches_passes_provenance_exposure(monkeypatch): from jarca_ledger_mcp.adapters.mcp import server captured = {} def fake_list_ledger_import_batches(source, include_rolled_back, limit, provenance_exposure="minimal", debug_provenance_gate=False): captured.update( source=source, include_rolled_back=include_rolled_back, limit=limit, provenance_exposure=provenance_exposure, debug_provenance_gate=debug_provenance_gate, ) return {"success": True, "import_batches": [], "count": 0} monkeypatch.setattr(server.contracts, "list_ledger_import_batches", fake_list_ledger_import_batches) result = server.list_ledger_import_batches(source="kb_bank_pdf", include_rolled_back=False, limit=5) assert result["success"] is True assert captured == { "source": "kb_bank_pdf", "include_rolled_back": False, "limit": 5, "provenance_exposure": EXPOSURE_LEVEL_MINIMAL, "debug_provenance_gate": False, } result = server.list_ledger_import_batches( source="seoulpay_xlsx", include_rolled_back=True, limit=3, provenance_exposure=EXPOSURE_LEVEL_OPERATOR_SUMMARY, debug_provenance_gate=True, ) assert result["success"] is True assert captured == { "source": "seoulpay_xlsx", "include_rolled_back": True, "limit": 3, "provenance_exposure": EXPOSURE_LEVEL_OPERATOR_SUMMARY, "debug_provenance_gate": True, }