verae-mcp-timestamping-server/tests/test_server.py

"""In-memory MCP Client against hash-only tools."""

import base64

import pytest
from mcp import Client

from verae_mcp_timestamping_server.global_client import GlobalClient
from verae_mcp_timestamping_server.guard import FORBIDDEN_CONTENT_KEYS
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger
from verae_mcp_timestamping_server.server import build_mcp

pytestmark = pytest.mark.anyio


def _error_text(result) -> str:
    parts = []
    for block in result.content or []:
        text = getattr(block, "text", None)
        if text:
            parts.append(text)
    return "\n".join(parts)


async def test_stamp_tool_returns_receipts():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp, raise_exceptions=True) as client:
        result = await client.call_tool(
            "stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
        )
        assert result.is_error is not True
        data = result.structured_content
        assert data is not None
        receipts = data["receipts"]
        assert receipts[0]["hash"] == SHA256_EMPTY
        assert receipts[0]["scope"] == "local"


async def test_stamp_and_hash_report_schemas_forbid_content_bytes():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp, raise_exceptions=True) as client:
        listed = await client.list_tools()
        by_name = {t.name: t for t in listed.tools}
        for name in ("stamp", "hash_report", "verify"):
            tool = by_name[name]
            props = tool.input_schema["properties"]
            for key in FORBIDDEN_CONTENT_KEYS:
                assert key not in props, f"{name} schema still lists {key}"
            assert tool.input_schema.get("additionalProperties") is False
        assert "sha256" in by_name["stamp"].input_schema["properties"]
        assert "sha256" in by_name["hash_report"].input_schema["properties"]


async def test_stamp_rejects_content_bytes_at_runtime():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp) as client:
        r1 = await client.call_tool(
            "stamp",
            {
                "sha256": SHA256_EMPTY,
                "scope": "local",
                "content": "this is file bytes",
            },
        )
        assert r1.is_error is True
        assert "content/bytes paths are rejected" in _error_text(r1)

        r2 = await client.call_tool(
            "stamp",
            {
                "sha256": SHA256_EMPTY,
                "bytes": base64.b64encode(b"payload").decode(),
            },
        )
        assert r2.is_error is True
        assert "content/bytes paths are rejected" in _error_text(r2)

        r3 = await client.call_tool(
            "hash_report",
            {"sha256": SHA256_EMPTY, "file": "/tmp/secret.bin"},
        )
        assert r3.is_error is True
        assert "content/bytes paths are rejected" in _error_text(r3)


async def test_stamp_rejects_unknown_extra_keys():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp) as client:
        result = await client.call_tool(
            "stamp",
            {"sha256": SHA256_EMPTY, "scope": "local", "not_a_real_field": 1},
        )
        assert result.is_error is True
        assert "Extra inputs are not permitted" in _error_text(result)


async def test_stamp_rejects_non_digest_masquerading_as_sha256():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp) as client:
        long_b64 = base64.b64encode(b"x" * 200).decode()
        multiline = "line1\nline2\nline3"
        for bad in ("not-a-hash", "", long_b64, multiline, "a" * 128):
            result = await client.call_tool(
                "stamp", {"sha256": bad, "scope": "local"}
            )
            assert result.is_error is True, bad
            assert "sha256" in _error_text(result).lower()


async def test_verify_and_hash_report_tools():
    mcp = build_mcp(MemoryLedger())
    async with Client(mcp, raise_exceptions=True) as client:
        stamped = await client.call_tool(
            "stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
        )
        receipts = stamped.structured_content["receipts"]
        receipt_id = receipts[0]["receipt_id"]

        verified = await client.call_tool(
            "verify", {"receipt_id": receipt_id, "want_scope": "local"}
        )
        assert verified.structured_content["valid"] is True

        report = await client.call_tool("hash_report", {"sha256": SHA256_EMPTY})
        items = report.structured_content["receipts"]
        assert items[0]["receipt_id"] == receipt_id


class _ToolHttp:
    def __init__(self):
        self.calls = []

    def request(self, method, url, payload, headers):
        self.calls.append((method, url, payload, headers))

        class Resp:
            status_code = 200
            text = ""
            headers = {"Content-Type": "application/json"}

            def json(self):
                if url.endswith("/auth/login"):
                    return {
                        "token": "tool-token",
                        "expiresAt": "2099-01-01T00:00:00Z",
                        "user": {"id": "u1", "username": "admin", "role": "admin"},
                    }
                if url.endswith("/auth/validate"):
                    return {
                        "valid": True,
                        "username": "admin",
                        "role": "admin",
                        "expiresAt": "2099-01-01T00:00:00Z",
                    }
                return {}

        return Resp()


async def test_login_tool_does_not_return_token(tmp_path):
    http = _ToolHttp()
    held = GlobalClient(
        "https://api.veraetime.net",
        http=http,
        auth_path=tmp_path / "auth.properties",
    )
    mcp = build_mcp(MemoryLedger(), global_client=held)
    async with Client(mcp, raise_exceptions=True) as client:
        listed = await client.list_tools()
        login = next(tool for tool in listed.tools if tool.name == "login")
        assert login.input_schema.get("additionalProperties") is False
        assert "password" in login.input_schema["properties"]
        result = await client.call_tool(
            "login", {"username": "admin", "password": "secret-pass"}
        )
        data = result.structured_content
        assert data["username"] == "admin"
        assert data["role"] == "admin"
        assert "token" not in data
        assert "tool-token" not in str(data)
        session = await client.call_tool("session", {})
        assert session.structured_content["valid"] is True
        assert session.structured_content["role"] == "admin"