Spaces:
Running
Running
SooryaKiranKakkarayil
Day 2: MCP server complete - 4 tools, TTL cache, retry/circuit-breaker- lookup_domain_age, check_ipqs, check_virustotal, check_local_blocklist- Shared cache.py (1h TTL) and resilience.py (retry + circuit breaker)- Local OpenPhish/URLhaus blocklist loader with real snapshot data- Mocked pytest suite (16 tests) plus live MCP Inspector verification- Fixes: follow_redirects on RDAP bootstrap redirect; sys.path fix for direct script invocation; load_dotenv() for stdio-launched subprocess
33391b8 | """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 | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| 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") | |