mcp_server/server.py

"""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")