#!/usr/bin/env python3 """Read-only educational receipt checker. Own application profile, not MCP conformance.""" import json import math import os import re import stat import sys VERSION = "0.1.0" MAX_BYTES = 1048576 MAX_AMOUNT = 2**53 - 1 VERSIONS = {"2026-07-28", "2025-11-25"} CURRENCIES = {"RUB", "USD", "EUR"} IDENTIFIER = re.compile(r"example-[A-Za-z0-9][A-Za-z0-9._-]*\Z") class Rejected(Exception): def __init__(self, outcome, diagnostic, code): self.outcome, self.diagnostic, self.code = outcome, diagnostic, code def reject(outcome, diagnostic, code=2): raise Rejected(outcome, diagnostic, code) def exact_keys(value, required): return type(value) is dict and set(value) == set(required) def identifier(value): return type(value) is str and IDENTIFIER.fullmatch(value) is not None def amount(value): return type(value) is int and 0 < value <= MAX_AMOUNT def rpc_id(value): return type(value) in (int, str) def same_id(left, right): return type(left) is type(right) and left == right def pairs_object(pairs): obj = {} for key, value in pairs: if key in obj: reject("invalid_input", "duplicate_json_key") obj[key] = value return obj def reject_constant(_value): reject("invalid_input", "nonfinite_json_number") def finite_float(value): number = float(value) if not math.isfinite(number): reject("invalid_input", "nonfinite_json_number") return number def read_json(path): """Bounded descriptor-based read; reject links in every path component on POSIX.""" fd = None try: # O_NOFOLLOW and dir_fd are required: never silently weaken this policy. if not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"): reject("input_io_error", "unsupported_file_platform", 4) if not path or "\x00" in path: reject("input_io_error", "unreadable_input", 4) parts = path.split(os.sep) fd = os.open(os.sep if os.path.isabs(path) else ".", os.O_RDONLY | os.O_DIRECTORY) components = [part for part in parts if part not in ("", ".")] if not components: reject("input_io_error", "nonregular_input", 4) for part in components[:-1]: next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) os.close(fd) fd = next_fd next_fd = os.open(components[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd) os.close(fd) fd = next_fd info = os.fstat(fd) if not stat.S_ISREG(info.st_mode): reject("input_io_error", "nonregular_input", 4) if info.st_size > MAX_BYTES: reject("invalid_input", "file_too_large") # A file growing after fstat still cannot exceed the read bound. data = bytearray() while len(data) <= MAX_BYTES: block = os.read(fd, min(65536, MAX_BYTES + 1 - len(data))) if not block: break data.extend(block) if len(data) > MAX_BYTES: reject("invalid_input", "file_too_large") except Rejected: raise except (OSError, ValueError, TypeError, NotImplementedError): reject("input_io_error", "unreadable_input", 4) finally: if fd is not None: os.close(fd) try: decoded = data.decode("utf-8", errors="strict") except UnicodeError: reject("invalid_input", "invalid_utf8") try: return json.loads(decoded, object_pairs_hook=pairs_object, parse_constant=reject_constant, parse_float=finite_float) except Rejected: raise except (ValueError, RecursionError): reject("invalid_input", "invalid_json") def evaluate(document): if not exact_keys(document, ("protocol_version", "expected", "response")): reject("invalid_input", "input_fields") expected = document["expected"] if not exact_keys(expected, ("rpc_id", "order_id", "amount_minor", "currency")): reject("invalid_input", "expected_fields") if not rpc_id(expected["rpc_id"]): reject("invalid_input", "expected_rpc_id_type") if not identifier(expected["order_id"]): reject("invalid_input", "expected_order_id") if not amount(expected["amount_minor"]): reject("invalid_input", "expected_amount_minor") if type(expected["currency"]) is not str or expected["currency"] not in CURRENCIES: reject("invalid_input", "expected_currency") version = document["protocol_version"] if type(version) is not str: reject("invalid_input", "protocol_version_type") if version not in VERSIONS: reject("out_of_scope", "unsupported_protocol_version", 3) envelope = document["response"] if type(envelope) is not dict: reject("invalid_envelope", "envelope_object") if ("result" in envelope) == ("error" in envelope): reject("invalid_envelope", "result_error_exclusivity") channel = "error" if "error" in envelope else "result" if not exact_keys(envelope, ("jsonrpc", "id", channel)): reject("invalid_envelope", "envelope_fields") if type(envelope["jsonrpc"]) is not str or envelope["jsonrpc"] != "2.0": reject("invalid_envelope", "jsonrpc_version") if not rpc_id(envelope["id"]): reject("invalid_envelope", "rpc_id_type") if not same_id(envelope["id"], expected["rpc_id"]): reject("invalid_envelope", "rpc_id_mismatch") if channel == "error": error = envelope["error"] if (type(error) is not dict or not {"code", "message"} <= set(error) or not set(error) <= {"code", "message", "data"} or type(error["code"]) is not int or type(error["message"]) is not str): reject("invalid_envelope", "rpc_error_shape") reject("rpc_error", "jsonrpc_error_channel", 1) result = envelope["result"] if type(result) is not dict: reject("invalid_envelope", "result_object") if "isError" in result and type(result["isError"]) is not bool: reject("invalid_envelope", "isError_type") if "resultType" not in result: if version == "2026-07-28": reject("invalid_envelope", "missing_resultType") else: if type(result["resultType"]) is not str or not result["resultType"]: reject("invalid_envelope", "resultType_type") if result["resultType"] != "complete": reject("out_of_scope", "unsupported_resultType", 3) if result.get("isError") is True: reject("tool_error", "tool_isError", 1) if "structuredContent" not in result: reject("missing_receipt", "missing_structuredContent", 1) receipt = result["structuredContent"] if not exact_keys(receipt, ("refund_id", "order_id", "amount_minor", "currency", "status")): reject("invalid_receipt", "receipt_fields") if not identifier(receipt["refund_id"]) or not identifier(receipt["order_id"]): reject("invalid_receipt", "receipt_identifier") if not amount(receipt["amount_minor"]): reject("invalid_receipt", "receipt_amount_minor") if type(receipt["currency"]) is not str or receipt["currency"] not in CURRENCIES: reject("invalid_receipt", "receipt_currency") if type(receipt["status"]) is not str or receipt["status"] not in {"succeeded", "pending", "failed"}: reject("invalid_receipt", "receipt_status") if any(receipt[key] != expected[key] for key in ("order_id", "amount_minor", "currency")): reject("mismatch", "receipt_expected_mismatch", 1) if receipt["status"] == "pending": reject("pending", "receipt_pending", 1) if receipt["status"] == "failed": reject("failed", "receipt_failed", 1) return "consistent_sample", "educational_expected_match", 0 def main(argv=None): argv = sys.argv[1:] if argv is None else argv try: if argv == ["--help"]: outcome, diagnostic, code = "usage", "check.py INPUT_JSON; see README.md", 0 elif len(argv) != 1 or argv[0].startswith("-"): reject("invalid_input", "cli_arguments") else: outcome, diagnostic, code = evaluate(read_json(argv[0])) except Rejected as error: outcome, diagnostic, code = error.outcome, error.diagnostic, error.code except (Exception, KeyboardInterrupt): # Exception strings, paths, and input content must never reach stdout/stderr. outcome, diagnostic, code = "internal_error", "checker_exception", 2 print(json.dumps({"version": VERSION, "outcome": outcome, "diagnostic": diagnostic, "exit_code": code}, sort_keys=True)) return code if __name__ == "__main__": sys.exit(main())