diff --git a/pyproject.toml b/pyproject.toml index d2fc6bb..177770c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ include = ["revettr*", "revettr_mcp*"] [project] name = "revettr" -version = "0.3.0" +version = "0.4.0" description = "Counterparty risk scoring for agentic commerce" readme = "README.md" license = "MIT" diff --git a/revettr_mcp/server.py b/revettr_mcp/server.py index 1a9595f..5909eb3 100644 --- a/revettr_mcp/server.py +++ b/revettr_mcp/server.py @@ -1,10 +1,11 @@ -"""Revettr MCP Server — counterparty risk scoring as an MCP tool. +"""Revettr MCP Server — counterparty risk scoring as MCP tools. Wraps the Revettr REST API (revettr.com) so MCP clients like -Claude Desktop, Cursor, and Windsurf can call score_counterparty -as a native tool. +Claude Desktop, Cursor, and Windsurf can call scoring and risk +analysis tools natively. """ +import asyncio import ipaddress import math import os @@ -35,17 +36,126 @@ def _validate_url(url: str) -> None: _validate_url(REVETTR_URL) +# --------------------------------------------------------------------------- +# Flag descriptions for explain_risk +# --------------------------------------------------------------------------- + +FLAG_DESCRIPTIONS: dict[str, str] = { + "domain_age_under_30d": ( + "The domain was registered less than 30 days ago, a common indicator " + "of fraudulent or newly created entities." + ), + "domain_age_under_90d": ( + "The domain was registered less than 90 days ago. Newer domains carry " + "higher risk." + ), + "no_mx_records": ( + "The domain has no MX (mail) records, suggesting it may not be a " + "legitimate business domain." + ), + "ssl_invalid": "The domain's SSL certificate is invalid or expired.", + "ssl_missing": "The domain has no SSL certificate.", + "tor_exit_node": ( + "The IP address is a Tor exit node, indicating anonymization." + ), + "known_vpn": "The IP address belongs to a known VPN provider.", + "datacenter_ip": ( + "The IP address belongs to a datacenter rather than a residential or " + "business network." + ), + "wallet_never_transacted": ( + "This wallet has never transacted on-chain, suggesting it was recently " + "created or unused." + ), + "wallet_age_under_7d": "This wallet was created less than 7 days ago.", + "wallet_age_under_30d": "This wallet is less than 30 days old.", + "wallet_age_under_90d": "This wallet is less than 90 days old.", + "low_counterparty_diversity": ( + "This wallet has transacted with very few unique counterparties." + ), + "sanctions_exact_match": ( + "CRITICAL: Exact match on international sanctions lists (OFAC/EU/UN). " + "Do not transact." + ), + "sanctions_high_confidence_match": ( + "High-confidence match on sanctions lists. Treat as sanctioned until " + "verified." + ), + "no_spf_record": ( + "The domain lacks an SPF record for email authentication." + ), + "no_dmarc_record": ( + "The domain lacks a DMARC record for email authentication." + ), +} + mcp = FastMCP( "Revettr", instructions=( "Revettr scores counterparties for agentic commerce. " - "Use score_counterparty before sending money to an unknown entity. " + "Available tools:\n" + "1. score_counterparty — Score a counterparty before sending money. " "Send any combination of domain, ip, wallet_address, or company_name. " - "Score 80-100 = low risk, 60-79 = medium, 30-59 = high, 0-29 = critical." + "Score 80-100 = low risk, 60-79 = medium, 30-59 = high, 0-29 = critical.\n" + "2. is_safe_to_transact — Quick yes/no safety check for a wallet address " + "against a configurable score threshold.\n" + "3. score_batch — Score up to 10 wallets in parallel and return results " + "sorted by score (highest/safest first).\n" + "4. explain_risk — Get human-readable explanations of risk flags for a " + "counterparty, with tier-based recommendations.\n" + "5. health_check — Check Revettr API availability and signal source status." ), ) +# --------------------------------------------------------------------------- +# Shared scoring helper +# --------------------------------------------------------------------------- + + +async def _score_one(body: dict) -> dict: + """Route a single scoring request through x402 payment or direct call.""" + wallet_key = os.getenv("REVETTR_WALLET_KEY") + if wallet_key: + return await _call_with_x402_payment(body, wallet_key) + else: + return await _call_direct(body) + + +# --------------------------------------------------------------------------- +# Validation helpers +# --------------------------------------------------------------------------- + + +def _validate_amount_usd(amount_usd: float | None) -> None: + """Validate amount_usd is finite, positive, and not NaN.""" + if amount_usd is not None: + if not isinstance(amount_usd, (int, float)): + raise ValueError("amount_usd must be a number") + try: + val = float(amount_usd) + except (OverflowError, ValueError): + raise ValueError("amount_usd is too large or not a valid number") + if math.isnan(val) or math.isinf(val): + raise ValueError("amount_usd must be finite (not NaN or inf)") + if val <= 0: + raise ValueError("amount_usd must be greater than 0") + + +def _validate_wallet_address(wallet_address: str) -> None: + """Validate an EVM wallet address.""" + if not re.match(r"^0x[a-fA-F0-9]{40}$", wallet_address): + raise ValueError( + f"Invalid EVM wallet address: {wallet_address!r}. " + "Expected format: 0x followed by 40 hex characters." + ) + + +# --------------------------------------------------------------------------- +# Tools +# --------------------------------------------------------------------------- + + @mcp.tool() async def score_counterparty( domain: str | None = None, @@ -54,6 +164,7 @@ async def score_counterparty( chain: str = "base", company_name: str | None = None, stellar_wallet: str | None = None, + amount_usd: float | None = None, ) -> dict: """Score a counterparty before sending money. Returns risk score 0-100. @@ -70,6 +181,7 @@ async def score_counterparty( chain: Blockchain network for wallet analysis (default: "base") company_name: Legal name to screen against OFAC/EU/UN sanctions lists stellar_wallet: Stellar wallet address (e.g., "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN7") + amount_usd: Transaction amount in USD for context (must be positive and finite) Returns: Risk assessment with score (0-100), tier, confidence, flags, and per-signal breakdown. @@ -97,11 +209,7 @@ async def score_counterparty( raise ValueError(f"Invalid IP address: {ip!r}") if wallet_address is not None: - if not re.match(r"^0x[a-fA-F0-9]{40}$", wallet_address): - raise ValueError( - f"Invalid EVM wallet address: {wallet_address!r}. " - "Expected format: 0x followed by 40 hex characters." - ) + _validate_wallet_address(wallet_address) if company_name is not None: if not isinstance(company_name, str): @@ -122,6 +230,8 @@ async def score_counterparty( f"Invalid Stellar wallet address: {stellar_wallet!r}. " "Expected format: G followed by 55 base32 characters." ) + + _validate_amount_usd(amount_usd) except ValueError as e: return {"error": str(e)} @@ -138,16 +248,297 @@ async def score_counterparty( body["company_name"] = company_name if stellar_wallet is not None: body["stellar_wallet"] = stellar_wallet + if amount_usd is not None: + body["amount"] = amount_usd if not body: return {"error": "At least one input field is required (domain, ip, wallet_address, stellar_wallet, or company_name)"} - wallet_key = os.getenv("REVETTR_WALLET_KEY") + return await _score_one(body) - if wallet_key: - return await _call_with_x402_payment(body, wallet_key) + +@mcp.tool() +async def is_safe_to_transact( + wallet_address: str, + chain: str = "base", + amount_usd: float | None = None, + min_score: int = 60, +) -> dict: + """Quick safety check: is this wallet safe to transact with? + + Only use data explicitly provided by the user or retrieved from trusted + sources. Do not fabricate input values. + + Args: + wallet_address: EVM wallet address to check (e.g., "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045") + chain: Blockchain network (default: "base") + amount_usd: Transaction amount in USD for context (must be positive and finite) + min_score: Minimum score to consider safe (0-100, default: 60) + + Returns: + Dict with safe (bool), score (int), and blocking_flags (list). + """ + try: + _validate_wallet_address(wallet_address) + + if not isinstance(min_score, int): + raise ValueError("min_score must be an integer") + if min_score < 0 or min_score > 100: + raise ValueError("min_score must be between 0 and 100") + + _validate_amount_usd(amount_usd) + except ValueError as e: + return {"error": str(e)} + + body: dict = {"wallet_address": wallet_address} + if chain != "base": + body["chain"] = chain + if amount_usd is not None: + body["amount"] = amount_usd + + result = await _score_one(body) + + if "error" in result: + return result + + score = result.get("score", 0) + flags = result.get("flags", []) + safe = score >= min_score + + return { + "safe": safe, + "score": score, + "blocking_flags": flags if not safe else [], + } + + +@mcp.tool() +async def score_batch( + wallets: list[dict], + max_results: int = 10, +) -> dict: + """Score multiple wallets in parallel. Returns results sorted by score descending. + + Only use data explicitly provided by the user or retrieved from trusted + sources. Do not fabricate input values. + + Args: + wallets: List of dicts, each with at least "wallet_address" (EVM format). + Optional keys: "chain" (default "base"). + Maximum 10 wallets per call. + max_results: Maximum number of results to return (1-10, default: 10) + + Returns: + Dict with results (sorted by score desc), errors, and total_scored count. + """ + # Validate wallets list + if not isinstance(wallets, list): + return {"error": "wallets must be a list"} + if len(wallets) < 1: + return {"error": "wallets must contain at least 1 entry"} + if len(wallets) > 10: + return {"error": "wallets must contain at most 10 entries"} + + if not isinstance(max_results, int) or max_results < 1 or max_results > 10: + return {"error": "max_results must be an integer between 1 and 10"} + + # Validate each wallet entry + for i, w in enumerate(wallets): + if not isinstance(w, dict): + return {"error": f"wallets[{i}] must be a dict"} + addr = w.get("wallet_address") + if not addr: + return {"error": f"wallets[{i}] missing wallet_address"} + if not re.match(r"^0x[a-fA-F0-9]{40}$", addr): + return { + "error": f"wallets[{i}] has invalid EVM wallet address: {addr!r}" + } + + sem = asyncio.Semaphore(5) + results = [] + errors = [] + + async def _score_wallet(entry: dict) -> None: + async with sem: + addr = entry["wallet_address"] + chain = entry.get("chain", "base") + body: dict = {"wallet_address": addr} + if chain != "base": + body["chain"] = chain + + try: + result = await _score_one(body) + if "error" in result: + errors.append({"wallet_address": addr, "error": result["error"]}) + else: + result["wallet_address"] = addr + results.append(result) + except Exception as e: + errors.append({"wallet_address": addr, "error": str(e)}) + + await asyncio.gather(*[_score_wallet(w) for w in wallets]) + + # Sort by score descending + results.sort(key=lambda r: r.get("score", 0), reverse=True) + + return { + "results": results[:max_results], + "errors": errors, + "total_scored": len(results), + } + + +@mcp.tool() +async def explain_risk( + wallet_address: str | None = None, + chain: str = "base", + domain: str | None = None, + ip: str | None = None, + company_name: str | None = None, +) -> dict: + """Explain risk flags for a counterparty in human-readable language. + + Only use data explicitly provided by the user or retrieved from trusted + sources. Do not fabricate input values. + + Provide at least one identifier. More data yields richer explanations. + + Args: + wallet_address: EVM wallet address (e.g., "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045") + chain: Blockchain network (default: "base") + domain: Domain or URL of the counterparty + ip: IP address of the counterparty server + company_name: Legal name to screen against sanctions lists + + Returns: + Dict with summary, risk_factors (human-readable), recommendation, score, and tier. + """ + if all(v is None for v in [wallet_address, domain, ip, company_name]): + return {"error": "At least one identifier is required (wallet_address, domain, ip, or company_name)"} + + try: + if wallet_address is not None: + _validate_wallet_address(wallet_address) + + if domain is not None: + if not isinstance(domain, str): + raise ValueError("domain must be a string") + domain = domain.strip() + hostname = domain + if "://" in domain: + from urllib.parse import urlparse + parsed = urlparse(domain) + hostname = parsed.hostname or domain + if len(hostname) > 253: + raise ValueError("domain hostname exceeds 253-character DNS limit") + if re.search(r"\s", domain): + raise ValueError("domain must not contain whitespace") + + if ip is not None: + try: + ipaddress.ip_address(ip) + except ValueError: + raise ValueError(f"Invalid IP address: {ip!r}") + + if company_name is not None: + if not isinstance(company_name, str): + raise ValueError("company_name must be a string") + company_name = company_name.strip() + if len(company_name) > 200: + raise ValueError("company_name exceeds 200-character limit") + except ValueError as e: + return {"error": str(e)} + + body: dict = {} + if wallet_address is not None: + body["wallet_address"] = wallet_address + if chain != "base": + body["chain"] = chain + if domain is not None: + body["domain"] = domain + if ip is not None: + body["ip"] = ip + if company_name is not None: + body["company_name"] = company_name + + result = await _score_one(body) + + if "error" in result: + return result + + score = result.get("score", 0) + tier = result.get("tier", "unknown") + flags = result.get("flags", []) + + # Map flags to human-readable descriptions + risk_factors = [] + for flag in flags: + if flag in FLAG_DESCRIPTIONS: + risk_factors.append({"flag": flag, "description": FLAG_DESCRIPTIONS[flag]}) + elif flag.startswith("high_risk_country_"): + country = flag.replace("high_risk_country_", "").replace("_", " ").title() + risk_factors.append({ + "flag": flag, + "description": f"Associated with {country}, which is classified as a high-risk jurisdiction.", + }) + else: + risk_factors.append({ + "flag": flag, + "description": f"Risk flag detected: {flag.replace('_', ' ')}.", + }) + + # Tier-based recommendation + recommendations = { + "low": "Safe to proceed.", + "medium": "Proceed with caution.", + "high": "High risk. Consider alternative.", + "critical": "Do not transact.", + } + recommendation = recommendations.get(tier, "Proceed with caution.") + + # Build summary + n_flags = len(risk_factors) + if n_flags == 0: + summary = f"Score {score}/100 ({tier} risk). No specific risk flags detected." else: - return await _call_direct(body) + summary = f"Score {score}/100 ({tier} risk). {n_flags} risk factor{'s' if n_flags != 1 else ''} identified." + + return { + "summary": summary, + "risk_factors": risk_factors, + "recommendation": recommendation, + "score": score, + "tier": tier, + } + + +@mcp.tool() +async def health_check() -> dict: + """Check Revettr API health and signal source availability. + + Returns: + Dict with API health status, or error information on failure. + """ + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: + response = await client.get(f"{REVETTR_URL}/health") + + if response.status_code == 200: + return response.json() + else: + return { + "error": f"Health check returned status {response.status_code}", + "status": response.status_code, + } + except httpx.TimeoutException: + return {"error": "Health check timed out (15s)"} + except Exception as e: + return {"error": f"Health check failed: {str(e)}"} + + +# --------------------------------------------------------------------------- +# HTTP transport helpers +# --------------------------------------------------------------------------- async def _call_with_x402_payment(body: dict, wallet_key: str) -> dict: @@ -184,16 +575,21 @@ async def _call_with_x402_payment(body: dict, wallet_key: str) -> dict: async def _call_direct(body: dict) -> dict: """Call Revettr API without payment (will get 402 if payment required).""" - async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client: - response = await client.post(f"{REVETTR_URL}/v1/score", json=body) + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client: + response = await client.post(f"{REVETTR_URL}/v1/score", json=body) - if response.status_code == 200: - return response.json() - elif response.status_code == 402: - return { - "error": "Payment required. Set REVETTR_WALLET_KEY environment variable with a funded Base wallet private key.", - "docs": "https://revettr.com/docs", - "pricing": "$0.01 USDC per request via x402 on Base", - } - else: - return {"error": f"API returned status {response.status_code}"} + if response.status_code == 200: + return response.json() + elif response.status_code == 402: + return { + "error": "Payment required. Set REVETTR_WALLET_KEY environment variable with a funded Base wallet private key.", + "docs": "https://revettr.com/docs", + "pricing": "$0.01 USDC per request via x402 on Base", + } + else: + return {"error": f"API returned status {response.status_code}"} + except httpx.TimeoutException: + return {"error": "Request timed out (30s)"} + except Exception: + return {"error": "Request failed — an unexpected error occurred"} diff --git a/server.json b/server.json index e346178..224bbea 100644 --- a/server.json +++ b/server.json @@ -6,12 +6,12 @@ "url": "https://github.com/AlexanderLawson17/revettr-python", "source": "github" }, - "version": "0.2.3", + "version": "0.4.0", "packages": [ { "registryType": "pypi", "identifier": "revettr", - "version": "0.2.3", + "version": "0.4.0", "transport": { "type": "stdio" } diff --git a/tests/test_explain_flags.py b/tests/test_explain_flags.py new file mode 100644 index 0000000..1aabb1f --- /dev/null +++ b/tests/test_explain_flags.py @@ -0,0 +1,141 @@ +"""Tests for FLAG_DESCRIPTIONS completeness, formatting, and correctness.""" + +import pathlib +import pytest + + +def _fastmcp_available() -> bool: + try: + import fastmcp # noqa: F401 + return True + except ImportError: + return False + + +pytestmark = pytest.mark.skipif( + not _fastmcp_available(), reason="fastmcp not installed" +) + +# Expected flag keys that must be present +EXPECTED_FLAGS = [ + "domain_age_under_30d", + "domain_age_under_90d", + "no_mx_records", + "ssl_invalid", + "ssl_missing", + "tor_exit_node", + "known_vpn", + "datacenter_ip", + "wallet_never_transacted", + "wallet_age_under_7d", + "wallet_age_under_30d", + "wallet_age_under_90d", + "low_counterparty_diversity", + "sanctions_exact_match", + "sanctions_high_confidence_match", + "no_spf_record", + "no_dmarc_record", +] + + +class TestFlagDescriptionsCompleteness: + """Ensure FLAG_DESCRIPTIONS contains all expected flags.""" + + def test_all_expected_flags_present(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + for flag in EXPECTED_FLAGS: + assert flag in FLAG_DESCRIPTIONS, f"Missing flag: {flag}" + + def test_no_extra_unexpected_flags(self): + """FLAG_DESCRIPTIONS should only contain known flags.""" + from revettr_mcp.server import FLAG_DESCRIPTIONS + + extra = set(FLAG_DESCRIPTIONS.keys()) - set(EXPECTED_FLAGS) + assert extra == set(), f"Unexpected flags in FLAG_DESCRIPTIONS: {extra}" + + def test_flag_count_matches(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert len(FLAG_DESCRIPTIONS) == len(EXPECTED_FLAGS) + + +class TestFlagDescriptionsFormatting: + """Ensure all flag descriptions are well-formed strings.""" + + def test_all_descriptions_are_strings(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + for flag, desc in FLAG_DESCRIPTIONS.items(): + assert isinstance(desc, str), f"{flag}: description is not a string" + + def test_all_descriptions_nonempty(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + for flag, desc in FLAG_DESCRIPTIONS.items(): + assert len(desc.strip()) > 0, f"{flag}: description is empty" + + def test_all_descriptions_end_with_period(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + for flag, desc in FLAG_DESCRIPTIONS.items(): + assert desc.strip().endswith("."), ( + f"{flag}: description does not end with a period: {desc!r}" + ) + + def test_descriptions_are_human_readable_length(self): + """Descriptions should be between 10 and 500 characters.""" + from revettr_mcp.server import FLAG_DESCRIPTIONS + + for flag, desc in FLAG_DESCRIPTIONS.items(): + assert 10 <= len(desc) <= 500, ( + f"{flag}: description length {len(desc)} outside expected range" + ) + + def test_sanctions_flags_are_emphatic(self): + """Sanctions flags should contain strong language.""" + from revettr_mcp.server import FLAG_DESCRIPTIONS + + exact = FLAG_DESCRIPTIONS["sanctions_exact_match"] + assert "CRITICAL" in exact or "Do not" in exact, ( + "sanctions_exact_match should contain urgent language" + ) + + high = FLAG_DESCRIPTIONS["sanctions_high_confidence_match"] + assert "sanctions" in high.lower(), ( + "sanctions_high_confidence_match should mention sanctions" + ) + + +class TestFlagDescriptionsContent: + """Spot-check specific flag descriptions for correctness.""" + + def test_domain_age_30d_mentions_30_days(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "30 days" in FLAG_DESCRIPTIONS["domain_age_under_30d"] + + def test_tor_flag_mentions_tor(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "Tor" in FLAG_DESCRIPTIONS["tor_exit_node"] + + def test_ssl_missing_mentions_ssl(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "SSL" in FLAG_DESCRIPTIONS["ssl_missing"] or "ssl" in FLAG_DESCRIPTIONS["ssl_missing"] + + def test_wallet_never_transacted_mentions_on_chain(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "on-chain" in FLAG_DESCRIPTIONS["wallet_never_transacted"] + + def test_spf_flag_mentions_spf(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "SPF" in FLAG_DESCRIPTIONS["no_spf_record"] + + def test_dmarc_flag_mentions_dmarc(self): + from revettr_mcp.server import FLAG_DESCRIPTIONS + + assert "DMARC" in FLAG_DESCRIPTIONS["no_dmarc_record"] diff --git a/tests/test_mcp_tools.py b/tests/test_mcp_tools.py new file mode 100644 index 0000000..1a7b5a3 --- /dev/null +++ b/tests/test_mcp_tools.py @@ -0,0 +1,469 @@ +"""Tests for MCP server tools: is_safe_to_transact, score_batch, explain_risk, health_check. + +Uses unittest.mock.AsyncMock to patch _score_one so tests run without +network access or fastmcp optional dependency issues. +""" + +import asyncio +import math +import re +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + + +def _fastmcp_available() -> bool: + try: + import fastmcp # noqa: F401 + return True + except ImportError: + return False + + +pytestmark = pytest.mark.skipif( + not _fastmcp_available(), reason="fastmcp not installed" +) + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + +VALID_WALLET = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045" +VALID_WALLET_2 = "0x1234567890abcdef1234567890abcdef12345678" + +MOCK_SCORE_RESPONSE = { + "score": 85, + "tier": "low", + "confidence": 0.9, + "signals_checked": 3, + "flags": [], +} + +MOCK_RISKY_RESPONSE = { + "score": 25, + "tier": "critical", + "confidence": 0.8, + "signals_checked": 4, + "flags": ["sanctions_exact_match", "wallet_never_transacted"], +} + +MOCK_MEDIUM_RESPONSE = { + "score": 65, + "tier": "medium", + "confidence": 0.7, + "signals_checked": 2, + "flags": ["domain_age_under_30d", "no_mx_records"], +} + +MOCK_HIGH_RESPONSE = { + "score": 40, + "tier": "high", + "confidence": 0.6, + "signals_checked": 3, + "flags": ["wallet_age_under_7d", "low_counterparty_diversity"], +} + + +# --------------------------------------------------------------------------- +# score_counterparty — amount_usd parameter +# --------------------------------------------------------------------------- + + +class TestScoreCounterpartyAmountUsd: + """Test the new amount_usd parameter on score_counterparty.""" + + def test_amount_usd_added_to_body(self): + from revettr_mcp.server import score_counterparty + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE + result = asyncio.run( + score_counterparty(wallet_address=VALID_WALLET, amount_usd=100.0) + ) + call_body = mock.call_args[0][0] + assert call_body["amount"] == 100.0 + + def test_amount_usd_nan_rejected(self): + from revettr_mcp.server import score_counterparty + + result = asyncio.run( + score_counterparty(wallet_address=VALID_WALLET, amount_usd=float("nan")) + ) + assert "error" in result + assert "finite" in result["error"] + + def test_amount_usd_inf_rejected(self): + from revettr_mcp.server import score_counterparty + + result = asyncio.run( + score_counterparty(wallet_address=VALID_WALLET, amount_usd=float("inf")) + ) + assert "error" in result + assert "finite" in result["error"] + + def test_amount_usd_negative_rejected(self): + from revettr_mcp.server import score_counterparty + + result = asyncio.run( + score_counterparty(wallet_address=VALID_WALLET, amount_usd=-5.0) + ) + assert "error" in result + assert "greater than 0" in result["error"] + + def test_amount_usd_none_ok(self): + from revettr_mcp.server import score_counterparty + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE + result = asyncio.run( + score_counterparty(wallet_address=VALID_WALLET, amount_usd=None) + ) + call_body = mock.call_args[0][0] + assert "amount" not in call_body + + +# --------------------------------------------------------------------------- +# is_safe_to_transact +# --------------------------------------------------------------------------- + + +class TestIsSafeToTransact: + """Test the is_safe_to_transact tool.""" + + def test_safe_wallet(self): + from revettr_mcp.server import is_safe_to_transact + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE + result = asyncio.run(is_safe_to_transact(wallet_address=VALID_WALLET)) + assert result["safe"] is True + assert result["score"] == 85 + assert result["blocking_flags"] == [] + + def test_unsafe_wallet(self): + from revettr_mcp.server import is_safe_to_transact + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_RISKY_RESPONSE + result = asyncio.run(is_safe_to_transact(wallet_address=VALID_WALLET)) + assert result["safe"] is False + assert result["score"] == 25 + assert len(result["blocking_flags"]) > 0 + + def test_custom_min_score(self): + from revettr_mcp.server import is_safe_to_transact + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_MEDIUM_RESPONSE # score=65 + result = asyncio.run( + is_safe_to_transact(wallet_address=VALID_WALLET, min_score=70) + ) + assert result["safe"] is False + + def test_invalid_wallet_rejected(self): + from revettr_mcp.server import is_safe_to_transact + + result = asyncio.run(is_safe_to_transact(wallet_address="not-a-wallet")) + assert "error" in result + assert "Invalid EVM" in result["error"] + + def test_min_score_out_of_range(self): + from revettr_mcp.server import is_safe_to_transact + + result = asyncio.run( + is_safe_to_transact(wallet_address=VALID_WALLET, min_score=150) + ) + assert "error" in result + assert "between 0 and 100" in result["error"] + + def test_min_score_negative(self): + from revettr_mcp.server import is_safe_to_transact + + result = asyncio.run( + is_safe_to_transact(wallet_address=VALID_WALLET, min_score=-1) + ) + assert "error" in result + + def test_amount_usd_passed_through(self): + from revettr_mcp.server import is_safe_to_transact + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE + asyncio.run( + is_safe_to_transact(wallet_address=VALID_WALLET, amount_usd=50.0) + ) + call_body = mock.call_args[0][0] + assert call_body["amount"] == 50.0 + + def test_api_error_passed_through(self): + from revettr_mcp.server import is_safe_to_transact + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = {"error": "API returned status 500"} + result = asyncio.run(is_safe_to_transact(wallet_address=VALID_WALLET)) + assert "error" in result + + +# --------------------------------------------------------------------------- +# score_batch +# --------------------------------------------------------------------------- + + +class TestScoreBatch: + """Test the score_batch tool.""" + + def test_batch_two_wallets(self): + from revettr_mcp.server import score_batch + + responses = [ + {"score": 90, "tier": "low", "flags": []}, + {"score": 40, "tier": "high", "flags": ["wallet_age_under_7d"]}, + ] + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.side_effect = responses + result = asyncio.run(score_batch(wallets=[ + {"wallet_address": VALID_WALLET}, + {"wallet_address": VALID_WALLET_2}, + ])) + assert result["total_scored"] == 2 + assert len(result["results"]) == 2 + # Sorted descending by score + assert result["results"][0]["score"] >= result["results"][1]["score"] + + def test_empty_wallets_rejected(self): + from revettr_mcp.server import score_batch + + result = asyncio.run(score_batch(wallets=[])) + assert "error" in result + assert "at least 1" in result["error"] + + def test_too_many_wallets_rejected(self): + from revettr_mcp.server import score_batch + + wallets = [{"wallet_address": VALID_WALLET}] * 11 + result = asyncio.run(score_batch(wallets=wallets)) + assert "error" in result + assert "at most 10" in result["error"] + + def test_invalid_wallet_in_batch(self): + from revettr_mcp.server import score_batch + + result = asyncio.run(score_batch(wallets=[ + {"wallet_address": "bad-address"}, + ])) + assert "error" in result + + def test_missing_wallet_address_key(self): + from revettr_mcp.server import score_batch + + result = asyncio.run(score_batch(wallets=[ + {"chain": "base"}, + ])) + assert "error" in result + assert "missing wallet_address" in result["error"] + + def test_partial_failure(self): + """When some wallets score and others error, both buckets are populated.""" + from revettr_mcp.server import score_batch + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.side_effect = [ + {"score": 90, "tier": "low", "flags": []}, + {"error": "API returned status 500"}, + {"score": 60, "tier": "medium", "flags": ["wallet_age_under_30d"]}, + ] + result = asyncio.run(score_batch(wallets=[ + {"wallet_address": VALID_WALLET}, + {"wallet_address": VALID_WALLET_2}, + {"wallet_address": f"0x{'b' * 40}"}, + ])) + assert result["total_scored"] == 2 + assert len(result["results"]) == 2 + assert len(result["errors"]) == 1 + assert result["errors"][0]["wallet_address"] == VALID_WALLET_2 + # Results still sorted by score descending + assert result["results"][0]["score"] >= result["results"][1]["score"] + + def test_max_results_truncation(self): + from revettr_mcp.server import score_batch + + responses = [ + {"score": 90 - i * 10, "tier": "low", "flags": []} + for i in range(3) + ] + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.side_effect = responses + result = asyncio.run(score_batch( + wallets=[ + {"wallet_address": f"0x{'a' * 38}{i:02d}"} for i in range(3) + ], + max_results=2, + )) + assert len(result["results"]) == 2 + + +# --------------------------------------------------------------------------- +# explain_risk +# --------------------------------------------------------------------------- + + +class TestExplainRisk: + """Test the explain_risk tool.""" + + def test_explain_with_known_flags(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_MEDIUM_RESPONSE + result = asyncio.run(explain_risk(domain="example.com")) + assert result["score"] == 65 + assert result["tier"] == "medium" + assert result["recommendation"] == "Proceed with caution." + assert len(result["risk_factors"]) == 2 + # Check that descriptions are filled in + for rf in result["risk_factors"]: + assert "description" in rf + assert len(rf["description"]) > 10 + + def test_explain_critical_tier(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_RISKY_RESPONSE + result = asyncio.run(explain_risk(wallet_address=VALID_WALLET)) + assert result["recommendation"] == "Do not transact." + assert result["tier"] == "critical" + + def test_explain_no_identifiers_rejected(self): + from revettr_mcp.server import explain_risk + + result = asyncio.run(explain_risk()) + assert "error" in result + assert "At least one" in result["error"] + + def test_explain_unknown_flag(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = { + "score": 50, "tier": "high", "flags": ["some_unknown_flag"], + "confidence": 0.5, "signals_checked": 1, + } + result = asyncio.run(explain_risk(domain="example.com")) + rf = result["risk_factors"][0] + assert "some unknown flag" in rf["description"] + + def test_explain_high_risk_country_flag(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = { + "score": 30, "tier": "high", "flags": ["high_risk_country_north_korea"], + "confidence": 0.9, "signals_checked": 1, + } + result = asyncio.run(explain_risk(domain="example.com")) + rf = result["risk_factors"][0] + assert "North Korea" in rf["description"] + + def test_explain_low_tier_recommendation(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE # tier="low" + result = asyncio.run(explain_risk(wallet_address=VALID_WALLET)) + assert result["recommendation"] == "Safe to proceed." + + def test_explain_summary_no_flags(self): + from revettr_mcp.server import explain_risk + + with patch("revettr_mcp.server._score_one", new_callable=AsyncMock) as mock: + mock.return_value = MOCK_SCORE_RESPONSE # flags=[] + result = asyncio.run(explain_risk(wallet_address=VALID_WALLET)) + assert "No specific risk flags" in result["summary"] + + +# --------------------------------------------------------------------------- +# health_check +# --------------------------------------------------------------------------- + + +class TestHealthCheck: + """Test the health_check tool.""" + + def test_health_check_success(self): + from revettr_mcp.server import health_check + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"status": "ok", "version": "0.4.0"} + + with patch("httpx.AsyncClient") as MockClient: + mock_ctx = AsyncMock() + mock_ctx.get.return_value = mock_response + MockClient.return_value.__aenter__ = AsyncMock(return_value=mock_ctx) + MockClient.return_value.__aexit__ = AsyncMock(return_value=False) + + result = asyncio.run(health_check()) + assert result["status"] == "ok" + + def test_health_check_server_error(self): + from revettr_mcp.server import health_check + + mock_response = MagicMock() + mock_response.status_code = 500 + + with patch("httpx.AsyncClient") as MockClient: + mock_ctx = AsyncMock() + mock_ctx.get.return_value = mock_response + MockClient.return_value.__aenter__ = AsyncMock(return_value=mock_ctx) + MockClient.return_value.__aexit__ = AsyncMock(return_value=False) + + result = asyncio.run(health_check()) + assert "error" in result + assert "500" in result["error"] + + def test_health_check_timeout(self): + from revettr_mcp.server import health_check + + with patch("httpx.AsyncClient") as MockClient: + mock_ctx = AsyncMock() + mock_ctx.get.side_effect = httpx.TimeoutException("timed out") + MockClient.return_value.__aenter__ = AsyncMock(return_value=mock_ctx) + MockClient.return_value.__aexit__ = AsyncMock(return_value=False) + + result = asyncio.run(health_check()) + assert "error" in result + assert "timed out" in result["error"].lower() + + +# --------------------------------------------------------------------------- +# _score_one routing +# --------------------------------------------------------------------------- + + +class TestScoreOneRouting: + """Test that _score_one correctly routes to x402 or direct.""" + + def test_score_one_without_wallet_key(self): + from revettr_mcp.server import _score_one + + with patch.dict("os.environ", {}, clear=True): + with patch("revettr_mcp.server._call_direct", new_callable=AsyncMock) as mock_direct: + mock_direct.return_value = MOCK_SCORE_RESPONSE + result = asyncio.run(_score_one({"wallet_address": VALID_WALLET})) + mock_direct.assert_called_once() + assert result["score"] == 85 + + def test_score_one_with_wallet_key(self): + from revettr_mcp.server import _score_one + + with patch.dict("os.environ", {"REVETTR_WALLET_KEY": "0xfakekey"}): + with patch("revettr_mcp.server._call_with_x402_payment", new_callable=AsyncMock) as mock_x402: + mock_x402.return_value = MOCK_SCORE_RESPONSE + result = asyncio.run(_score_one({"wallet_address": VALID_WALLET})) + mock_x402.assert_called_once() + assert result["score"] == 85 diff --git a/tests/test_security.py b/tests/test_security.py index e28f04d..064ef3a 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -50,14 +50,6 @@ def test_http_rejected(self): with pytest.raises(ValueError, match="HTTPS"): Revettr(base_url="http://example.com") - @pytest.mark.xfail( - reason=( - "Known vulnerability: hostname.startswith('127.') matches " - "127.0.0.1.evil.com. Tracked for fix — the check should verify " - "the hostname is a real loopback IP, not just a prefix match." - ), - strict=True, - ) def test_127_subdomain_rejected(self): """http://127.0.0.1.evil.com must not be treated as localhost.""" with pytest.raises(ValueError, match="HTTPS"): @@ -91,10 +83,6 @@ def test_mcp_validate_url_localhost_http_accepted(self): _validate_url("http://127.0.0.1:8000") @pytest.mark.skipif(not _fastmcp_available(), reason="fastmcp not installed") - @pytest.mark.xfail( - reason="Same startswith('127.') vulnerability as client _validate_base_url", - strict=True, - ) def test_mcp_validate_url_127_subdomain_rejected(self): from revettr_mcp.server import _validate_url with pytest.raises(ValueError, match="HTTPS"): @@ -254,27 +242,35 @@ def test_domain_validation(self): assert "253" in result["error"] def test_defensive_instructions_in_tool_desc(self): - """Tool docstring contains defensive text about not fabricating data. + """All tools accepting user identifiers must contain defensive text. Reads the source AST so the test works even when fastmcp is not - installed. + installed. Tools that accept user-provided data (wallets, domains, + IPs, company names) should instruct the LLM not to fabricate inputs. + health_check is exempt — it takes no user identifiers. """ + tools_requiring_defense = [ + "score_counterparty", + "is_safe_to_transact", + "score_batch", + "explain_risk", + ] + source = (_PROJECT_ROOT / "revettr_mcp" / "server.py").read_text() tree = ast.parse(source) - # Find the score_counterparty function and extract its docstring - docstring = None + found = {} for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - if node.name == "score_counterparty": - docstring = ast.get_docstring(node) or "" - break - - assert docstring is not None, "score_counterparty function not found in server.py" - assert "do not fabricate" in docstring.lower(), ( - "score_counterparty docstring must contain defensive instruction " - "about not fabricating input values" - ) + if node.name in tools_requiring_defense: + found[node.name] = ast.get_docstring(node) or "" + + for tool_name in tools_requiring_defense: + assert tool_name in found, f"{tool_name} function not found in server.py" + assert "do not fabricate" in found[tool_name].lower(), ( + f"{tool_name} docstring must contain defensive instruction " + "about not fabricating input values" + ) # ---------------------------------------------------------------------------