"""scamsentry-reputation-mcp Read-only reputation lookup server for ScamSentry. Exposes exactly four tools, none of which can write, delete, or send anything — this is a structural property of the tool list, not a policy statement. See Section 5 of the design doc. Transport: stdio (co-located subprocess, no network-level auth needed). """ import os import re import sys from datetime import datetime, timezone from pathlib import Path from urllib.parse import quote # Makes this file runnable both as `python mcp_server/server.py` (what MCP # Inspector and most stdio launchers use) and as `python -m mcp_server.server`. # Running a .py file directly only adds its OWN folder to sys.path, not the # project root — so the `from mcp_server.xxx import ...` lines below would # otherwise raise ModuleNotFoundError. Inserting the parent directory here # fixes that for every invocation style. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import httpx from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP from mcp_server.blocklist import Blocklist from mcp_server.cache import cache_get, cache_set from mcp_server.resilience import RateLimited, with_resilience load_dotenv() mcp = FastMCP("scamsentry-reputation") _blocklist = Blocklist() HTTP_TIMEOUT_SECONDS = 5 # --------------------------------------------------------------------------- # Tool 1: lookup_domain_age # --------------------------------------------------------------------------- def _parse_rdap_registration_date(rdap_json: dict) -> str | None: """RDAP's `events` list isn't a fixed schema across registries — find the 'registration' event defensively rather than assuming index [0].""" for event in rdap_json.get("events", []): if event.get("eventAction") == "registration": return event.get("eventDate") return None @mcp.tool() def lookup_domain_age(domain: str) -> dict: """Return RDAP-based domain registration age for a domain. Returns: {created_date, age_days, registrar, status: "ok"|"degraded"} Newly-registered domains (a handful of days old) are one of the strongest cheap signals of a scam/phishing site. """ key = f"age:{domain}" if (cached := cache_get(key)) is not None: return cached degraded = {"status": "degraded", "domain": domain, "created_date": None, "age_days": None, "registrar": None} def call() -> dict: resp = httpx.get( f"https://rdap.org/domain/{domain}", timeout=HTTP_TIMEOUT_SECONDS, follow_redirects=True, # rdap.org 302s to the authoritative registry's # RDAP server for the TLD — that redirect IS the # normal response path, not an error condition. ) if resp.status_code == 429: raise RateLimited() resp.raise_for_status() data = resp.json() created_date = _parse_rdap_registration_date(data) age_days = None if created_date: try: created_dt = datetime.fromisoformat(created_date.replace("Z", "+00:00")) age_days = (datetime.now(timezone.utc) - created_dt).days except ValueError: pass registrar = None for entity in data.get("entities", []): if "registrar" in entity.get("roles", []): vcard = entity.get("vcardArray", [None, []])[1] for field in vcard: if field[0] == "fn": registrar = field[3] break return { "status": "ok", "domain": domain, "created_date": created_date, "age_days": age_days, "registrar": registrar, } result = with_resilience("rdap", call, degraded) return cache_set(key, result) # --------------------------------------------------------------------------- # Tool 2: check_ipqs # --------------------------------------------------------------------------- @mcp.tool() def check_ipqs(url: str) -> dict: """Check a URL against IPQualityScore's Malicious URL Scanner. IMPORTANT (ToS + PII hygiene): only bare URLs go to this endpoint — never full message text, email bodies, or anything containing PII. Returns: {risk_score, is_suspicious, is_phishing, is_malware, status} """ key = f"ipqs:{url}" if (cached := cache_get(key)) is not None: return cached degraded = {"status": "degraded", "url": url, "risk_score": None, "is_suspicious": None, "is_phishing": None, "is_malware": None} rate_limited = {"status": "rate_limited", "url": url, "risk_score": None, "is_suspicious": None, "is_phishing": None, "is_malware": None} def call() -> dict: api_key = os.environ["IPQS_API_KEY"] encoded_url = quote(url, safe="") resp = httpx.get( f"https://www.ipqualityscore.com/api/json/url/{api_key}/{encoded_url}", timeout=HTTP_TIMEOUT_SECONDS, ) if resp.status_code == 429: raise RateLimited() resp.raise_for_status() data = resp.json() if data.get("success") is False: # IPQS returns 200 with success:false for invalid key / bad request raise RuntimeError(data.get("message", "IPQS error")) return { "status": "ok", "url": url, "risk_score": data.get("risk_score", 0), "is_suspicious": data.get("suspicious", False), "is_phishing": data.get("phishing", False), "is_malware": data.get("malware", False), } result = with_resilience("ipqs", call, degraded, rate_limited) return cache_set(key, result) # --------------------------------------------------------------------------- # Tool 3: check_virustotal # --------------------------------------------------------------------------- @mcp.tool() def check_virustotal(url: str) -> dict: """Check a URL against VirusTotal's public API (URL report). 4 req/min, 500/day free tier — the TTL cache + per-session circuit breaker are load-bearing here, not just nice-to-have. Returns: {malicious_count, total_engines, status: "ok"|"degraded"|"rate_limited"} """ key = f"vt:{url}" if (cached := cache_get(key)) is not None: return cached degraded = {"status": "degraded", "url": url, "malicious_count": None, "total_engines": None} rate_limited = {"status": "rate_limited", "url": url, "malicious_count": None, "total_engines": None} def call() -> dict: import base64 api_key = os.environ["VIRUSTOTAL_API_KEY"] url_id = base64.urlsafe_b64encode(url.encode()).decode().strip("=") resp = httpx.get( f"https://www.virustotal.com/api/v3/urls/{url_id}", headers={"x-apikey": api_key}, timeout=HTTP_TIMEOUT_SECONDS, ) if resp.status_code == 429: raise RateLimited() if resp.status_code == 404: # URL never submitted/analyzed before — not an error, just no data yet. return {"status": "ok", "url": url, "malicious_count": 0, "total_engines": 0} resp.raise_for_status() data = resp.json() stats = data.get("data", {}).get("attributes", {}).get("last_analysis_stats", {}) malicious = stats.get("malicious", 0) total = sum(stats.values()) if stats else 0 return {"status": "ok", "url": url, "malicious_count": malicious, "total_engines": total} result = with_resilience("virustotal", call, degraded, rate_limited) return cache_set(key, result) # --------------------------------------------------------------------------- # Tool 4: check_local_blocklist # --------------------------------------------------------------------------- @mcp.tool() def check_local_blocklist(url_or_domain: str) -> dict: """Check a URL/domain against the local OpenPhish + URLhaus snapshot. Pure in-memory set lookup against a pre-downloaded snapshot — no network call, no rate limit, works even if both feeds are down. Returns: {found: bool, source: "openphish"|"urlhaus"|None, status: "ok"} """ found, source = _blocklist.contains(url_or_domain) return {"found": found, "source": source, "status": "ok"} if __name__ == "__main__": mcp.run(transport="stdio")