diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000000000000000000000000000000000000..3720de04cbb7cba29abc4efdcb8e0db83d8a7f98 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +__pycache__ +*.pyc +.git +tests +*.db +.env diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000000000000000000000000000000000000..d59c214fa9784e9885117503e14254e9b77bf059 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,18 @@ +root = true + +[*] +indent_style = space +indent_size = 4 +end_of_line = lf +charset = utf-8 +trim_trailing_whitespace = true +insert_final_newline = true + +[*.{yml,yaml}] +indent_size = 2 + +[*.md] +trim_trailing_whitespace = false + +[Dockerfile] +indent_size = 4 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000000000000000000000000000000000000..bb35edb6a99007e6d782e7bb370bccf9011dab7e --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,30 @@ +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.11.5 + hooks: + - id: ruff + args: [--fix] + - id: ruff-format + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.15.0 + hooks: + - id: mypy + args: [--ignore-missing-imports] + additional_dependencies: + - fastapi + - uvicorn + - httpx + - jinja2 + - plotly + - pydantic + - rich + files: ^mcpscope/ + + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v5.0.0 + hooks: + - id: trailing-whitespace + - id: end-of-file-fixer + - id: check-yaml + - id: check-added-large-files diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000000000000000000000000000000000000..9a4bb56a0a2b47bb2c5f3b62df1e283b56f4f76f --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,28 @@ +# MCPscop + +## [Unreleased] + +### Added +- Security events API and dashboard views +- JSON logging output +- Dashboard authentication +- MCPGuard event forwarder integration + +### Fixed +- API key validation in REST endpoints +- CORS headers for cross-origin dashboard access +- HTML sanitization in scan result display + +## [0.1.0] - 2025-08-01 + +### Added +- Unified dashboard for MCP/A2A scanner results +- FastAPI REST API for scan ingestion and querying +- SQLite storage backends +- Plotly-based interactive charts and visualizations +- Support for Cisco MCP/A2A scanner output +- Support for mcp-scan output +- Support for MCPwn output +- SARIF format ingestion +- Rich CLI output for terminal usage +- Scan comparison and trend analysis diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md new file mode 100644 index 0000000000000000000000000000000000000000..da28dfbe16e2f1024afcf45a42aa19d31a8d4225 --- /dev/null +++ b/CODE_OF_CONDUCT.md @@ -0,0 +1,43 @@ +# Contributor Covenant Code of Conduct + +## Our Pledge + +We as members, contributors, and leaders pledge to make participation in our +community a harassment-free experience for everyone, regardless of age, body +size, visible or invisible disability, ethnicity, sex characteristics, gender +identity and expression, level of experience, education, socio-economic status, +nationality, personal appearance, race, caste, color, religion, or sexual +identity and orientation. + +We pledge to act and interact in ways that contribute to an open, welcoming, +diverse, inclusive, and healthy community. + +## Our Standards + +Examples of behavior that contributes to a positive environment: + +- Demonstrating empathy and kindness toward other people +- Being respectful of differing opinions, viewpoints, and experiences +- Giving and gracefully accepting constructive feedback +- Accepting responsibility and apologizing to those affected by our mistakes +- Focusing on what is best for the overall community + +Examples of unacceptable behavior: + +- The use of sexualized language or imagery, and sexual attention or advances +- Trolling, insulting or derogatory comments, and personal or political attacks +- Public or private harassment +- Publishing others' private information without explicit permission + +## Enforcement + +Project maintainers are responsible for clarifying and enforcing our standards. +Instances of abusive, harassing, or otherwise unacceptable behavior may be +reported to the project team. All complaints will be reviewed and investigated +promptly and fairly. + +## Attribution + +This Code of Conduct is adapted from the [Contributor Covenant](https://www.contributor-covenant.org), +version 2.1, available at +https://www.contributor-covenant.org/version/2/1/code_of_conduct.html. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000000000000000000000000000000000000..64749f016ef49df1e1570c40fd60e7f03f1271ce --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,58 @@ +# Contributing + +Thanks for your interest in MCP-Scope! + +## Development Setup + +```bash +git clone https://github.com/Carlos-Projects/mcpscope +cd mcpscope +pip install -e ".[dev]" +``` + +## Running Tests + +```bash +pytest -v +``` + +All tests must pass before submitting a PR. + +## Code Style + +- Format: `ruff format .` +- Lint: `ruff check .` +- No commented-out code +- Type hints required for all public functions + +## Pull Request Process + +1. Create a feature branch from `main` +2. Write tests for your changes +3. Ensure all tests pass +4. Update `CHANGELOG.md` if applicable +5. Open a PR with a clear description + +## Commit Messages + +Follow [Conventional Commits](https://www.conventionalcommits.org/): + +``` +feat: add new scanner parser +fix: handle null severity in report +deps: bump fastapi to 0.136.0 +docs: update README badges +``` + +## Adding a New Scanner + +1. Create `mcpscope/ingest/new_scanner.py` extending `BaseParser` +2. Implement `parse()` and set `SCANNER_NAME` +3. Add to `PARSERS` dict in `cli.py` +4. Create test fixtures in `tests/fixtures/` +5. Add tests in `tests/test_parsers.py` +6. Update `README.md` supported scanners table + +## Reporting Issues + +Open a [GitHub Issue](https://github.com/Carlos-Projects/mcpscope/issues/new/choose) using the appropriate template. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000000000000000000000000000000000000..aeeb3594850636e253f3d9a2d94218c804ac5f06 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,14 @@ +FROM python:3.13-slim + +WORKDIR /app + +COPY pyproject.toml README.md ./ +COPY mcpscope/ mcpscope/ + +RUN pip install --no-cache-dir -e . + +EXPOSE 8080 + +VOLUME /root/.mcpscope + +CMD ["mcpscope", "serve", "--host", "0.0.0.0", "--port", "8080"] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000000000000000000000000000000000000..7866979e4588f29a34d35a7a2b526a69f6c5f300 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 MCP-Scope + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 8628cb35ab126d181d600be2cb263815e7913440..29053a1142a9e9232d77e058d67ba1162e8d8682 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,225 @@ +
+ +# MCP-Scope + +**Unified security dashboard for MCP/A2A scanner results** + +[![Python](https://img.shields.io/badge/python-3.10%2B-blue)](https://python.org) +[![Tests](https://img.shields.io/github/actions/workflow/status/Carlos-Projects/mcpscope/scan.yml?label=tests)](https://github.com/Carlos-Projects/mcpscope/actions) +[![License](https://img.shields.io/github/license/Carlos-Projects/mcpscope)](LICENSE) +[![GitHub release](https://img.shields.io/github/v/release/Carlos-Projects/mcpscope)](https://github.com/Carlos-Projects/mcpscope/releases) +[![Code style](https://img.shields.io/badge/code%20style-ruff-261230)](https://docs.astral.sh/ruff) + --- -title: MCP-Scope -emoji: 📊 -colorFrom: blue -colorTo: purple -sdk: docker -pinned: false -app_port: 8080 -fullWidth: true -short_description: Unified security dashboard for MCP/A2A scanner results ---- -# MCP-Scope 📊 +
+ +Consumes output from multiple MCP/A2A security scanners (Cisco MCP Scanner, Cisco A2A Scanner, mcp-scan, MCPwn, SARIF) and presents a consolidated view of your security posture via a web dashboard, REST API, and CLI. + +## Quick Start + +```bash +pip install -e . + +# Generate demo data and explore +mcpscope seed +mcpscope serve + +# Run a scan directly against a server +mcpscope scan mcp-scan https://mcp-server.example.com/mcp + +# Import results from different scanners +mcpscope import cisco-mcp results.json +mcpscope import cisco-a2a results.json +mcpscope import mcpwn results.json +mcpscope import sarif report.sarif + +# Export a compliance report +mcpscope report --format csv --output report.csv + +# Backup/Restore +mcpscope backup backup.db +mcpscope restore backup.db +``` + +## Commands + +| Command | Description | +|---------|-------------| +| `serve` | Start the FastAPI web dashboard | +| `scan` | Run a scanner directly against a target | +| `import` | Import scanner JSON/SARIF results into SQLite | +| `report` | Export JSON, CSV, or PDF compliance report | +| `seed` | Generate demo scan data | +| `prune` | Delete scans older than N days | +| `backup` | Backup the SQLite database | +| `restore` | Restore the SQLite database from backup | +| `config` | View or set configuration options | + +## API Endpoints + +| Endpoint | Description | +|----------|-------------| +| `GET /` | Dashboard UI with filters and tabs | +| `GET /findings/{id}` | Finding detail page | +| `GET /docs` | Swagger UI | +| `GET /api/health` | Health check | +| `GET /api/scans` | List scans (paginated) | +| `GET /api/scans/{id}` | Scan details with findings | +| `GET /api/scans/{a}/diff/{b}` | Compare two scans | +| `GET /api/findings` | Query findings (paginated, filterable) | +| `GET /api/findings/{id}` | Single finding | +| `GET /api/stats/summary` | Aggregated statistics | +| `GET /api/stats/top-tools` | Most vulnerable tools | +| `GET /api/stats/severity-trend` | Findings over time | +| `GET /api/stats/duplicates` | Deduplicated findings | +| `GET /api/report/json` | Full JSON report | +| `GET /api/report/csv` | CSV export | + +## API Usage Examples + +```bash +# Health check +curl http://localhost:8080/api/health + +# List scans (paginated) +curl "http://localhost:8080/api/scans?page=1&page_size=10" + +# Get scan with findings +curl http://localhost:8080/api/scans/scan-id-here + +# Query findings with filters +curl "http://localhost:8080/api/findings?severity=critical&page=1" + +# Search findings +curl "http://localhost:8080/api/findings?search=command" + +# Compare two scans +curl "http://localhost:8080/api/scans/scan-a/diff/scan-b" + +# Get stats summary +curl http://localhost:8080/api/stats/summary + +# Get duplicates +curl http://localhost:8080/api/stats/duplicates + +# Export as CSV +curl http://localhost:8080/api/report/csv -o report.csv + +# Full JSON report +curl http://localhost:8080/api/report/json + +# With API key authentication +curl -H "X-API-Key: your-key" http://localhost:8080/api/scans + +# Swagger docs +open http://localhost:8080/docs +``` + +## Dashboard Features + +- **Overview tab** — Severity pie chart, top tools bar chart, severity trend over time +- **Findings tab** — Filterable table with severity/scanner/tool/search, pagination, clickable rows for detail view +- **Duplicates tab** — Grouped findings by tool + title + severity across scans +- **Diff tab** — Side-by-side comparison between any two scans +- **Scans tab** — Historical scan table with severity counts +- **Auto-refresh** — Configurable auto-refresh interval +- **Finding detail page** — Full details including raw JSON data + +## Supported Scanners + +| Scanner | CLI Name | Format | +|---------|----------|--------| +| Cisco MCP Scanner | `cisco-mcp` | `scan_results` with analyzer-grouped findings | +| Cisco A2A Scanner | `cisco-a2a` | `findings` with AI Security Taxonomy metadata | +| mcp-scan (Invariant Labs) | `mcp-scan` / `mcpscan` | `issues` array with severity codes | +| MCPwn (ressl) | `mcpwn` | Standard findings with MCP-XXX IDs | +| MCPwn (Teycir legacy) | `mcpwn` | Legacy test-based format | +| SARIF | `sarif` | Standard SARIF 2.1 format | + +## Screenshots + +| Dashboard Overview | Findings Table | +|---|---| +| ![Overview](https://via.placeholder.com/600x300/1e293b/3b82f6?text=Severity+Pie+%2B+Trend+%2B+Top+Tools) | ![Findings](https://via.placeholder.com/600x300/1e293b/3b82f6?text=Filterable+Findings+Table) | +| **Scan Diff** | **Finding Detail** | +| ![Diff](https://via.placeholder.com/600x300/1e293b/3b82f6?text=Scan+Comparison) | ![Detail](https://via.placeholder.com/600x300/1e293b/3b82f6?text=Finding+Detail+View) | + +_Run `mcpscope seed && mcpscope serve` and open http://localhost:8080 to see the live dashboard._ + +## CI/CD Integration + +Secure the API with an API key: +```bash +mcpscope config set api_key "your-secret-key" +mcpscope serve +# All /api/* endpoints now require: X-API-Key: your-secret-key +``` + +A [GitHub Actions workflow](.github/workflows/scan.yml) is included for automated scanning. + +Slack alerts for critical/high findings: +```bash +mcpscope config set slack_webhook_url "https://hooks.slack.com/services/..." +``` + +Webhook URLs for custom integrations: +```bash +mcpscope config set webhook_urls '["https://your-server.com/webhook"]' +``` + +## Configuration + +Config file at `~/.mcpscope/config.json`: + +```bash +mcpscope config show +mcpscope config set port 9090 +mcpscope config set auto_refresh_seconds 60 +mcpscope config set max_upload_mb 100 +``` + +## Docker + +```bash +docker build -t mcpscope . +docker run -p 8080:8080 -v mcpscope-data:/root/.mcpscope mcpscope +``` + +## Development + +```bash +pip install -e ".[dev]" +pytest +``` + +## Security Events (MCPGuard Integration) + +MCP-Scope can receive real-time security events from MCPGuard: + +```bash +# Configure MCPGuard's config.yaml: +mcpscop_url: http://localhost:8000 + +# Events appear in the "Live Events" dashboard tab +``` + +| Endpoint | Method | Description | +|----------|--------|-------------| +| `/api/events` | POST | Ingest a security event | +| `/api/events` | GET | List events (filters: severity, event_type) | +| `/api/events/stats` | GET | Event statistics | +| `/api/events` | DELETE | Clear all events | + +## Related Projects + +MCPscop is part of the **Carlos-Projects** security ecosystem for AI agents: -**Unified security dashboard** for MCP/A2A scanner results. +- [**MCPGuard**](https://github.com/Carlos-Projects/mcpguard) — Runtime security proxy for MCP/A2A protocols with HTMX dashboard +- [**MCPwn**](https://github.com/Carlos-Projects/mcpwn) — Offensive security testing framework for MCP servers +- [**Palisade Scanner**](https://github.com/Carlos-Projects/palisade-scanner) — Scan web content for prompt injection and adversarial content +- [**AgentGate**](https://github.com/Carlos-Projects/agentgate) — Policy-based firewall and honeypot middleware for AI agents accessing websites -Consumes output from MCPGuard, MCPwn, Cisco MCP Scanner, mcp-scan, SARIF. +## License -Visit the [GitHub repo](https://github.com/Carlos-Projects/mcpscope) for usage. +[MIT](LICENSE) diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000000000000000000000000000000000000..d53dafeaca8ec91950c4f83069c2a1ef125942ae --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,34 @@ +# Security Policy + +## Reporting a Vulnerability + +If you discover a security vulnerability in MCP-Scope, please report it privately. + +**Do not create a public GitHub issue.** + +Send details to: **carlos@aiagentobservatory.org** + +Please include: + +- Description of the vulnerability +- Steps to reproduce +- Potential impact +- Any suggested mitigation (optional) + +We will acknowledge receipt within 48 hours and provide a timeline for a fix. + +## Scope + +- The web dashboard (FastAPI) +- The REST API +- CLI commands that process external input +- Dependencies with known CVEs + +## Out of Scope + +- The SQLite database file (local by default) +- Scanner tools that MCP-Scope wraps (report issues to their respective projects) + +## Preferred Languages + +English or Spanish. diff --git a/mcpscope/__init__.py b/mcpscope/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/mcpscope/__pycache__/__init__.cpython-313.pyc b/mcpscope/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..600f9aa9bc0f18e777b6876a34fb90276e8b7f8c Binary files /dev/null and b/mcpscope/__pycache__/__init__.cpython-313.pyc differ diff --git a/mcpscope/__pycache__/cli.cpython-313.pyc b/mcpscope/__pycache__/cli.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..864505cabae2ba0714b8fe6e2477a48d8e7e4bcd Binary files /dev/null and b/mcpscope/__pycache__/cli.cpython-313.pyc differ diff --git a/mcpscope/__pycache__/config.cpython-313.pyc b/mcpscope/__pycache__/config.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ae28918cd77be67bbe14c83c1c363651fb24ac6c Binary files /dev/null and b/mcpscope/__pycache__/config.cpython-313.pyc differ diff --git a/mcpscope/__pycache__/scanner.cpython-313.pyc b/mcpscope/__pycache__/scanner.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..29c962dd03acb21479d752f44f04b4e83e239f5a Binary files /dev/null and b/mcpscope/__pycache__/scanner.cpython-313.pyc differ diff --git a/mcpscope/__pycache__/webhooks.cpython-313.pyc b/mcpscope/__pycache__/webhooks.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..dd8cfb441f6d7e904c318a00f9c047e1daf79875 Binary files /dev/null and b/mcpscope/__pycache__/webhooks.cpython-313.pyc differ diff --git a/mcpscope/api/__init__.py b/mcpscope/api/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..a440a3785177b9e05cb4e19d29386b23f8c71481 --- /dev/null +++ b/mcpscope/api/__init__.py @@ -0,0 +1,3 @@ +from .server import create_app + +__all__ = ["create_app"] diff --git a/mcpscope/api/__pycache__/__init__.cpython-313.pyc b/mcpscope/api/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..13c5846ca01c6185f61a9baf2b8a96423d1d0d3d Binary files /dev/null and b/mcpscope/api/__pycache__/__init__.cpython-313.pyc differ diff --git a/mcpscope/api/__pycache__/routes.cpython-313.pyc b/mcpscope/api/__pycache__/routes.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..e8c368c66c05dd92df0a23c061608cdcbd8790b7 Binary files /dev/null and b/mcpscope/api/__pycache__/routes.cpython-313.pyc differ diff --git a/mcpscope/api/__pycache__/server.cpython-313.pyc b/mcpscope/api/__pycache__/server.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..5eff92f07602697b86b41885772a658dbd48f3be Binary files /dev/null and b/mcpscope/api/__pycache__/server.cpython-313.pyc differ diff --git a/mcpscope/api/routes.py b/mcpscope/api/routes.py new file mode 100644 index 0000000000000000000000000000000000000000..3eae5eb2b4a2ba1b5be1571dbfa385d78f120e68 --- /dev/null +++ b/mcpscope/api/routes.py @@ -0,0 +1,476 @@ +from __future__ import annotations +import csv +import hmac +import io +import re +from datetime import datetime, timezone + +from fastapi import APIRouter, Request, HTTPException, Query, Response +from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse + +from mcpscope.models.finding import Severity +from mcpscope.models.security_event import SecurityEvent +from mcp_taxonomy import ( + AttackCategory, Severity as TaxSeverity, DetectionMethod, + palisade_finding_to_taxonomy, + mcpguard_event_to_taxonomy, + mcpwn_finding_to_taxonomy, + agentgate_signal_to_taxonomy, +) +from mcp_taxonomy.core import CATEGORY_SEVERITY + +router = APIRouter() +PAGE_SIZE = 50 + +_JS_UNSAFE_RE = re.compile(r"") + + +def _sanitize_js(val: str, max_len: int = 200) -> str: + s = str(val) + s = _JS_UNSAFE_RE.sub("", s) + s = s.replace("\0", "") + return s[:max_len] + + +def _sanitize_tools(tools: list[dict]) -> list[dict]: + return [ + {k: _sanitize_js(v) if isinstance(v, str) else v for k, v in t.items()} + for t in tools + ] + + +def _sanitize_trend(trend: list[dict]) -> list[dict]: + return [ + {k: _sanitize_js(v) if isinstance(v, str) else v for k, v in t.items()} + for t in trend + ] + + +def get_store(request: Request): + return request.app.state.store + + +def get_templates(request: Request): + return request.app.state.templates + + +@router.get("/api/health") +def health(): + return {"status": "ok"} + + +@router.get("/api/taxonomy") +def taxonomy_info(): + """Return the canonical taxonomy definition.""" + return { + "attack_categories": [c.value for c in AttackCategory], + "severities": [s.value for s in TaxSeverity], + "detection_methods": [m.value for m in DetectionMethod], + "category_severity": {c.value: s.value for c, s in CATEGORY_SEVERITY.items()}, + } + + +@router.post("/api/taxonomy/normalize") +def normalize_taxonomy(body: dict): + """Normalize a finding/event from any project into the canonical taxonomy.""" + source = body.get("source", "") + raw = body.get("raw", body) + if source == "palisade-scanner": + event = palisade_finding_to_taxonomy(raw) + elif source == "mcpguard": + event = mcpguard_event_to_taxonomy(raw) + elif source == "mcpwn": + event = mcpwn_finding_to_taxonomy(raw) + elif source == "agentgate": + event = agentgate_signal_to_taxonomy( + signal_type=raw.get("signal_type", ""), + weight=raw.get("weight", 0), + action=raw.get("action", ""), + path=raw.get("path", ""), + user_agent=raw.get("userAgent", raw.get("user_agent", "")), + score=raw.get("score", 0), + ) + else: + return JSONResponse({"error": f"Unknown source: {source}"}, status_code=400) + return JSONResponse({ + "source": event.source, + "attack_category": event.attack_category.value, + "severity": event.severity.value, + "confidence": event.confidence.value, + "detection_method": event.detection_method.value if isinstance(event.detection_method, DetectionMethod) else event.detection_method, + "title": event.title, + "description": event.description, + "recommendation": event.recommendation, + "target": event.target, + "snippet": event.snippet[:200] if event.snippet else "", + "blocked": event.blocked, + "risk_score": event.risk_score, + }) + + +def _session_value(password: str, client_ip: str) -> str: + h = hmac.new(password.encode(), client_ip.encode(), "sha256") + return h.hexdigest() + + +@router.get("/login", response_class=HTMLResponse) +def login_page(request: Request): + templates = get_templates(request) + if not templates: + return HTMLResponse("

Not found

", status_code=500) + return templates.TemplateResponse(request, "login.html", {}) + + +@router.post("/api/login") +def login(request: Request, response: Response, body: dict | None = None): + cfg = request.app.state + if not cfg.dashboard_password: + return JSONResponse({"error": "Dashboard auth not configured"}, status_code=403) + password = (body or {}).get("password", "") + if not hmac.compare_digest(password, cfg.dashboard_password): + return JSONResponse({"error": "Invalid password"}, status_code=401) + ip = request.client.host if request.client else "" + session = _session_value(cfg.dashboard_password, ip) + response.set_cookie( + key="mcpscope_session", + value=session, + max_age=86400, + httponly=True, + samesite="lax", + secure=False, + ) + return {"status": "ok"} + + +@router.post("/api/logout") +def logout(response: Response): + response.delete_cookie("mcpscope_session") + return {"status": "ok"} + + +@router.get("/api/scans") +def list_scans( + request: Request, + page: int = Query(1, ge=1), + page_size: int = Query(PAGE_SIZE, ge=1, le=200), +): + store = get_store(request) + scans, total = store.get_scans_paginated(page, page_size) + return { + "scans": [s.model_dump() for s in scans], + "total": total, + "page": page, + "page_size": page_size, + "pages": (total + page_size - 1) // page_size, + } + + +@router.get("/api/scans/{scan_id}") +def get_scan( + request: Request, + scan_id: str, + page: int = Query(1, ge=1), + page_size: int = Query(PAGE_SIZE, ge=1, le=200), +): + store = get_store(request) + scan = store.get_scan(scan_id) + if not scan: + raise HTTPException(404, "Scan not found") + findings, total = store.get_findings( + scan_id=scan_id, page=page, page_size=page_size + ) + return { + "scan": scan.model_dump(), + "findings": [f.model_dump() for f in findings], + "total": total, + "page": page, + "page_size": page_size, + } + + +@router.get("/api/scans/{scan_id}/diff/{other_id}") +def diff_scans(request: Request, scan_id: str, other_id: str): + store = get_store(request) + try: + result = store.diff_scans(scan_id, other_id) + except ValueError as e: + raise HTTPException(404, str(e)) + return result + + +@router.get("/api/findings") +def list_findings( + request: Request, + scan_id: str | None = Query(None), + severity: str | None = Query(None), + tool_name: str | None = Query(None), + scanner: str | None = Query(None), + search: str | None = Query(None), + page: int = Query(1, ge=1), + page_size: int = Query(PAGE_SIZE, ge=1, le=200), +): + store = get_store(request) + findings, total = store.get_findings( + scan_id=scan_id, + severity=severity, + tool_name=tool_name, + scanner=scanner, + search=search, + page=page, + page_size=page_size, + ) + return { + "findings": [f.model_dump() for f in findings], + "total": total, + "page": page, + "page_size": page_size, + "pages": (total + page_size - 1) // page_size, + } + + +@router.get("/api/findings/{finding_id}") +def get_finding(request: Request, finding_id: str): + store = get_store(request) + finding = store.get_finding(finding_id) + if not finding: + raise HTTPException(404, "Finding not found") + scan = store.get_scan(finding.scan_id) + return { + "finding": finding.model_dump(), + "scan": scan.model_dump() if scan else None, + } + + +@router.get("/api/stats/top-tools") +def top_tools(request: Request): + store = get_store(request) + return {"tools": store.get_top_tools()} + + +@router.get("/api/stats/severity-trend") +def severity_trend(request: Request): + store = get_store(request) + return {"trend": store.get_severity_trend()} + + +@router.get("/api/stats/scanners") +def list_scanners(request: Request): + store = get_store(request) + return {"scanners": store.get_scanners()} + + +@router.get("/api/stats/tool-names") +def list_tool_names(request: Request): + store = get_store(request) + return {"tools": store.get_tool_names()} + + +@router.get("/api/stats/duplicates") +def list_duplicates(request: Request): + store = get_store(request) + return {"duplicates": store.get_duplicates()} + + +@router.get("/api/stats/summary") +def summary(request: Request): + store = get_store(request) + history = store.get_scan_history() + top = store.get_top_tools() + dups = store.get_duplicates() + return { + "total_scans": len(history.scans), + "total_findings": history.total_findings, + "critical": history.total_critical, + "high": history.total_high, + "medium": history.total_medium, + "low": history.total_low, + "info": history.total_info, + "top_tools": top, + "duplicates": dups, + "scans": [s.model_dump() for s in history.scans], + } + + +@router.get("/api/report/csv") +def report_csv(request: Request): + store = get_store(request) + findings, _ = store.get_findings(page_size=100000) + output = io.StringIO() + writer = csv.writer(output) + writer.writerow( + [ + "id", + "scan_id", + "scanner", + "tool_name", + "severity", + "title", + "description", + "recommendation", + "cvss_score", + "cve_id", + "created_at", + ] + ) + for f in findings: + sev = f.severity.value if isinstance(f.severity, Severity) else str(f.severity) + writer.writerow( + [ + f.id, + f.scan_id, + f.scanner, + f.tool_name, + sev, + f.title, + f.description or "", + f.recommendation or "", + f.cvss_score or "", + f.cve_id or "", + f.created_at, + ] + ) + return PlainTextResponse( + output.getvalue(), + media_type="text/csv", + headers={"Content-Disposition": "attachment; filename=mcpscope-report.csv"}, + ) + + +@router.get("/api/report/json") +def report_json(request: Request): + store = get_store(request) + history = store.get_scan_history() + top = store.get_top_tools() + trend = store.get_severity_trend() + dups = store.get_duplicates() + return { + "generated_at": datetime.now(timezone.utc).isoformat(), + "summary": { + "total_scans": len(history.scans), + "total_findings": history.total_findings, + "critical": history.total_critical, + "high": history.total_high, + "medium": history.total_medium, + "low": history.total_low, + "info": history.total_info, + }, + "top_tools": top, + "duplicates": dups, + "severity_trend": trend, + "scans": [s.model_dump() for s in history.scans], + } + + +@router.post("/api/events") +async def ingest_event(request: Request): + store = get_store(request) + body = await request.json() + event = SecurityEvent(**body) + saved = store.save_event(event) + return {"status": "ok", "id": saved.id} + + +@router.get("/api/events") +def list_events( + request: Request, + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + severity: str | None = Query(None), + event_type: str | None = Query(None), +): + store = get_store(request) + events, total = store.get_events( + limit=limit, + offset=offset, + severity=severity, + event_type=event_type, + ) + return { + "events": [e.model_dump() for e in events], + "total": total, + "limit": limit, + "offset": offset, + } + + +@router.get("/api/events/stats") +def event_stats(request: Request): + store = get_store(request) + return store.get_event_stats() + + +@router.delete("/api/events") +def clear_events(request: Request): + store = get_store(request) + store.clear_events() + return {"status": "ok"} + + +@router.get("/", response_class=HTMLResponse) +def dashboard(request: Request): + store = get_store(request) + history = store.get_scan_history() + top_tools_data = store.get_top_tools() + trend_data = store.get_severity_trend() + scanners = store.get_scanners() + tool_names = store.get_tool_names() + dups = store.get_duplicates() + + templates = get_templates(request) + if not templates: + return HTMLResponse("

Dashboard templates not found

", status_code=500) + + return templates.TemplateResponse( + request, + "dashboard.html", + { + "total_scans": len(history.scans), + "total_findings": history.total_findings, + "critical": history.total_critical, + "high": history.total_high, + "medium": history.total_medium, + "low": history.total_low, + "info": history.total_info, + "top_tools": _sanitize_tools(top_tools_data), + "trend_data": _sanitize_trend(trend_data), + "scans": history.scans, + "scanners": scanners, + "tool_names": tool_names, + "duplicates": dups, + "auto_refresh": getattr(request.app.state, "auto_refresh", 30), + }, + ) + + +@router.get("/findings/{finding_id}", response_class=HTMLResponse) +def finding_detail(request: Request, finding_id: str): + store = get_store(request) + finding = store.get_finding(finding_id) + if not finding: + return HTMLResponse("

Finding not found

", status_code=404) + scan = store.get_scan(finding.scan_id) + templates = get_templates(request) + if not templates: + return HTMLResponse("

Not found

", status_code=500) + return templates.TemplateResponse( + request, + "dashboard.html", + { + "total_scans": 0, + "total_findings": 0, + "critical": 0, + "high": 0, + "medium": 0, + "low": 0, + "info": 0, + "top_tools": [], + "trend_data": [], + "scans": [], + "scanners": [], + "tool_names": [], + "duplicates": [], + "detail_finding": finding, + "detail_scan": scan, + "auto_refresh": 0, + }, + ) diff --git a/mcpscope/api/server.py b/mcpscope/api/server.py new file mode 100644 index 0000000000000000000000000000000000000000..770aecab00144b8f55056df70f2bff95d1a2652c --- /dev/null +++ b/mcpscope/api/server.py @@ -0,0 +1,123 @@ +from __future__ import annotations +import hmac +import json +import logging +from pathlib import Path + +from fastapi import FastAPI, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse, RedirectResponse + +from mcpscope.storage.store import Store +from mcpscope.config import Settings + + +API_KEY_HEADER = "X-API-Key" +DASHBOARD_COOKIE = "mcpscope_session" + + +class JSONLogFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + return json.dumps( + { + "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + "module": record.module, + "function": record.funcName, + } + ) + + +def setup_logging(settings: Settings) -> None: + root = logging.getLogger("mcpscope") + root.setLevel(getattr(logging, settings.log_level.upper(), logging.INFO)) + handler = logging.StreamHandler() + if settings.log_json: + handler.setFormatter(JSONLogFormatter()) + else: + handler.setFormatter( + logging.Formatter( + "%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + ) + root.handlers.clear() + root.addHandler(handler) + + +def _verify_session(request: Request, password: str) -> bool: + cookie = request.cookies.get(DASHBOARD_COOKIE, "") + if not cookie: + return False + expected = _session_value(password, request.client.host if request.client else "") + return hmac.compare_digest(cookie, expected) + + +def _session_value(password: str, client_ip: str) -> str: + h = hmac.new(password.encode(), client_ip.encode(), "sha256") + return h.hexdigest() + + +def create_app(store: Store | None = None, settings: Settings | None = None) -> FastAPI: + cfg = settings or Settings.load() + setup_logging(cfg) + logger = logging.getLogger("mcpscope.api") + + app = FastAPI( + title="MCP-Scope API", + description="Unified security dashboard for MCP/A2A scanner results", + version="0.1.0", + docs_url="/docs", + redoc_url="/redoc", + openapi_url="/api/openapi.json", + ) + + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=False, + allow_methods=["*"], + allow_headers=["*", API_KEY_HEADER], + ) + + app.state.store = store or Store(db_path=cfg.db_path) + app.state.auto_refresh = cfg.auto_refresh_seconds + app.state.api_key = cfg.api_key + app.state.dashboard_password = cfg.dashboard_password + + @app.middleware("http") + async def auth_middleware(request: Request, call_next): + path = request.url.path + + if path in ("/health", "/api/health", "/api/login", "/api/logout"): + return await call_next(request) + + if cfg.api_key and path.startswith("/api/") and request.method != "OPTIONS": + req_key = request.headers.get(API_KEY_HEADER) + if not hmac.compare_digest(req_key or "", cfg.api_key): + logger.warning("API key rejected for %s", path) + return JSONResponse( + status_code=401, content={"error": "Invalid or missing API key"} + ) + + if cfg.dashboard_password and not path.startswith("/api/"): + if not _verify_session(request, cfg.dashboard_password): + if path != "/login": + return RedirectResponse(url="/login") + + response = await call_next(request) + return response + + templates_dir = Path(__file__).resolve().parent.parent / "templates" + if templates_dir.exists(): + from fastapi.templating import Jinja2Templates + + app.state.templates = Jinja2Templates(directory=str(templates_dir)) + + from mcpscope.api.routes import router + + app.include_router(router) + + return app diff --git a/mcpscope/cli.py b/mcpscope/cli.py new file mode 100644 index 0000000000000000000000000000000000000000..d4eb91704b1e539885f82fb677ee9832bb5b4dc9 --- /dev/null +++ b/mcpscope/cli.py @@ -0,0 +1,438 @@ +from __future__ import annotations +import sys +import json +from pathlib import Path + +from rich.console import Console +from rich.table import Table +from rich.progress import Progress + +from mcpscope.api.server import create_app +from mcpscope.storage.store import Store +from mcpscope.ingest.cisco_mcp import CiscoMCPParser +from mcpscope.ingest.cisco_a2a import CiscoA2AParser +from mcpscope.ingest.mcpscan import MCPScanParser +from mcpscope.ingest.mcpwn import MCPwnParser +from mcpscope.ingest.sarif import SarifParser +from mcpscope.ingest.base import ParseError +from mcpscope.models.scan import ScanRun +from mcpscope.config import Settings +from mcpscope.scanner import ScannerRunner + +console = Console() +PARSERS = { + "cisco-mcp": CiscoMCPParser(), + "cisco-a2a": CiscoA2AParser(), + "mcp-scan": MCPScanParser(), + "mcpscan": MCPScanParser(), + "mcpwn": MCPwnParser(), + "sarif": SarifParser(), +} + + +def cmd_import(args: list[str]): + if len(args) < 2: + console.print("[red]Usage: mcpscope import [target][/red]") + sys.exit(1) + + scanner_name = args[0].lower() + file_path = Path(args[1]) + target = args[2] if len(args) > 2 else None + + parser = PARSERS.get(scanner_name) + if not parser: + console.print( + f"[red]Unknown scanner: {scanner_name}. Available: {', '.join(PARSERS)}[/red]" + ) + sys.exit(1) + + if not file_path.exists(): + console.print(f"[red]File not found: {file_path}[/red]") + sys.exit(1) + + store = Store() + + try: + with Progress() as progress: + task = progress.add_task(f"Parsing {file_path.name}...", total=None) + findings = parser.parse_file(file_path) + progress.update(task, completed=True) + except ParseError as e: + console.print(f"[red]Parse error: {e}[/red]") + if e.details: + console.print(f"[dim]{e.details}[/dim]") + sys.exit(1) + + scanner_label = findings[0].scanner + scan = ScanRun( + id=file_path.stem, + scanner=scanner_label, + target=target or scanner_label, + raw_file=str(file_path.resolve()), + ) + saved = store.save_scan(scan, findings) + + table = Table(title=f"Imported: {file_path.name}") + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green") + table.add_row("Scanner", scanner_label) + table.add_row("Findings", str(len(findings))) + table.add_row("Critical", str(saved.critical_count)) + table.add_row("High", str(saved.high_count)) + table.add_row("Medium", str(saved.medium_count)) + table.add_row("Low", str(saved.low_count)) + table.add_row("Info", str(saved.info_count)) + table.add_row("Scan ID", saved.id) + console.print(table) + + +def cmd_serve(args: list[str]): + settings = Settings.load() + for i, a in enumerate(args): + if a == "--port" and i + 1 < len(args): + settings.port = int(args[i + 1]) + elif a == "--host" and i + 1 < len(args): + settings.host = args[i + 1] + elif a == "--config" and i + 1 < len(args): + settings = Settings.load(args[i + 1]) + + import uvicorn + + store = Store(db_path=settings.db_path) + app = create_app(store) + app.state.auto_refresh = settings.auto_refresh_seconds + console.print( + f"[green]MCP-Scope dashboard running at http://{settings.host}:{settings.port}[/green]" + ) + console.print( + f"[dim]DB: {settings.db_path} | Auto-refresh: {settings.auto_refresh_seconds}s[/dim]" + ) + uvicorn.run( + app, host=settings.host, port=settings.port, log_level=settings.log_level + ) + + +def cmd_report(args: list[str]): + fmt = "json" + output = "mcpscope-report.json" + for i, a in enumerate(args): + if a == "--format" and i + 1 < len(args): + fmt = args[i + 1].lower() + elif a == "--output" and i + 1 < len(args): + output = args[i + 1] + + store = Store() + history = store.get_scan_history() + top = store.get_top_tools() + + report_data = { + "generated_at": __import__("datetime").datetime.utcnow().isoformat(), + "summary": { + "total_scans": len(history.scans), + "total_findings": history.total_findings, + "critical": history.total_critical, + "high": history.total_high, + "medium": history.total_medium, + "low": history.total_low, + "info": history.total_info, + }, + "scans": [s.model_dump() for s in history.scans], + "top_tools": top, + } + + out_path = Path(output) + if fmt == "json": + with open(out_path, "w") as f: + json.dump(report_data, f, indent=2) + console.print(f"[green]Report written to {out_path.resolve()}[/green]") + + elif fmt == "csv": + import csv + + store = Store() + findings, _ = store.get_findings(page_size=100000) + with open(out_path, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow( + [ + "id", + "scan_id", + "scanner", + "tool_name", + "severity", + "title", + "description", + "recommendation", + "created_at", + ] + ) + for finding in findings: + sev = ( + finding.severity.value + if hasattr(finding.severity, "value") + else finding.severity + ) + writer.writerow( + [ + finding.id, + finding.scan_id, + finding.scanner, + finding.tool_name, + sev, + finding.title, + (finding.description or "")[:200], + (finding.recommendation or "")[:200], + finding.created_at, + ] + ) + console.print(f"[green]CSV report written to {out_path.resolve()}[/green]") + + elif fmt == "pdf": + try: + from weasyprint import HTML + except ImportError: + console.print( + "[red]PDF generation requires: pip install mcpscope[pdf][/red]" + ) + sys.exit(1) + + html_content = _render_report_html(report_data) + HTML(string=html_content).write_pdf(str(out_path)) + console.print(f"[green]PDF report written to {out_path.resolve()}[/green]") + + else: + console.print(f"[red]Unsupported format: {fmt}. Use json, csv, or pdf.[/red]") + sys.exit(1) + + +def cmd_seed(args: list[str]): + store = Store() + store.seed_demo_data() + history = store.get_scan_history() + console.print( + f"[green]Seeded {len(history.scans)} demo scans with {history.total_findings} findings[/green]" + ) + + +def cmd_backup(args: list[str]): + output = ( + args[0] + if args + else f"mcpscope-backup-{__import__('datetime').datetime.now().strftime('%Y%m%d_%H%M%S')}.db" + ) + store = Store() + store.backup(output) + console.print(f"[green]Backup saved to {Path(output).resolve()}[/green]") + + +def cmd_restore(args: list[str]): + if not args: + console.print("[red]Usage: mcpscope restore [/red]") + sys.exit(1) + path = Path(args[0]) + if not path.exists(): + console.print(f"[red]Backup not found: {path}[/red]") + sys.exit(1) + store = Store() + store.restore(path) + history = store.get_scan_history() + console.print( + f"[green]Restored {len(history.scans)} scans, {history.total_findings} findings[/green]" + ) + + +def cmd_config(args: list[str]): + settings = Settings.load() + if not args: + console.print("[bold]Current configuration:[/bold]") + for k, v in settings.as_dict().items(): + console.print(f" {k}: {v}") + console.print(f"\nConfig file: {Path.home() / '.mcpscope' / 'config.json'}") + return + + action = args[0] + if action == "show": + for k, v in settings.as_dict().items(): + console.print(f" {k}: {v}") + elif action == "set" and len(args) >= 3: + key, value = args[1], args[2] + if hasattr(settings, key): + current = getattr(settings, key) + if value.lower() in ("none", "null", ""): + setattr(settings, key, None) + elif isinstance(current, bool): + setattr(settings, key, value.lower() in ("true", "1", "yes")) + elif isinstance(current, int): + setattr(settings, key, int(value)) + elif isinstance(current, list): + import json + + try: + setattr(settings, key, json.loads(value)) + except (json.JSONDecodeError, TypeError): + console.print(f"[red]Expected JSON array for {key}[/red]") + sys.exit(1) + else: + setattr(settings, key, value) + settings.save() + console.print(f"[green]{key} set to {value}[/green]") + else: + console.print(f"[red]Unknown config key: {key}[/red]") + sys.exit(1) + else: + console.print("Usage: mcpscope config [show|set ]") + + +def cmd_scan(args: list[str]): + if len(args) < 2: + console.print("[red]Usage: mcpscope scan [/red]") + console.print(f"Scanners: {', '.join(ScannerRunner.PARSERS)}") + sys.exit(1) + + scanner_name = args[0].lower() + target = args[1] + runner = ScannerRunner() + store = Store() + + try: + with Progress() as progress: + task = progress.add_task( + f"Running {scanner_name} against {target}...", total=None + ) + scan = runner.scan(scanner_name, target, store=store) + progress.update(task, completed=True) + except (ValueError, RuntimeError) as e: + console.print(f"[red]{e}[/red]") + sys.exit(1) + + table = Table(title=f"Scan: {scan.id[:12]}") + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green") + table.add_row("Scanner", scan.scanner) + table.add_row("Target", target) + table.add_row("Findings", str(scan.findings_count)) + table.add_row("Critical", str(scan.critical_count)) + table.add_row("High", str(scan.high_count)) + table.add_row("Medium", str(scan.medium_count)) + table.add_row("Low", str(scan.low_count)) + table.add_row("Info", str(scan.info_count)) + table.add_row("Scan ID", scan.id) + console.print(table) + + +def cmd_prune(args: list[str]): + keep_days = 30 + for i, a in enumerate(args): + if a == "--keep" and i + 1 < len(args): + keep_days = int(args[i + 1]) + + store = Store() + count = store.prune(keep_days) + if count: + console.print( + f"[green]Pruned {count} scans older than {keep_days} days[/green]" + ) + else: + console.print(f"[yellow]No scans older than {keep_days} days to prune[/yellow]") + + +def _render_report_html(data: dict) -> str: + summary = data["summary"] + scans_rows = "" + for s in data["scans"]: + scans_rows += f""" + {s["id"]} + {s["scanner"]} + {s["findings_count"]} + {s["critical_count"]} + {s["high_count"]} + {s["medium_count"]} + {s["low_count"]} + {s["info_count"]} + {s["created_at"]} + """ + + tools_rows = "" + for t in data.get("top_tools", []): + tools_rows += f""" + {t["tool_name"]} + {t["total"]} + {t["critical_high"]} + """ + + return f""" +MCP-Scope Report + +

MCP-Scope Security Report

+

Generated: {data["generated_at"]}

+

Summary

+
+

Total Scans

{summary["total_scans"]}
+

Total Findings

{summary["total_findings"]}
+

Critical

{summary["critical"]}
+

High

{summary["high"]}
+

Medium

{summary["medium"]}
+

Low

{summary["low"]}
+
+

Scan History

+ +{scans_rows}
IDScannerTotalCriticalHighMediumLowInfoDate
+

Top Vulnerable Tools

+ +{tools_rows}
ToolTotal FindingsCritical/High
+""" + + +def cli(): + if len(sys.argv) < 2: + console.print("[bold]MCP-Scope[/bold] - Unified Security Dashboard") + console.print("Usage:") + console.print(" mcpscope serve [--port PORT] [--host HOST] [--config FILE]") + console.print(" mcpscope scan ") + console.print(" mcpscope import [target]") + console.print(" mcpscope report [--format json|csv|pdf] [--output FILE]") + console.print(" mcpscope seed") + console.print(" mcpscope prune [--keep DAYS]") + console.print(" mcpscope backup [file]") + console.print(" mcpscope restore ") + console.print(" mcpscope config [show|set ]") + console.print("\nScanners (import): " + ", ".join(PARSERS)) + console.print("Scanners (scan): " + ", ".join(ScannerRunner.PARSERS)) + return + + command = sys.argv[1] + args = sys.argv[2:] + + commands = { + "serve": cmd_serve, + "scan": cmd_scan, + "import": cmd_import, + "report": cmd_report, + "seed": cmd_seed, + "prune": cmd_prune, + "backup": cmd_backup, + "restore": cmd_restore, + "config": cmd_config, + } + + fn = commands.get(command) + if fn: + fn(args) + else: + console.print(f"[red]Unknown command: {command}[/red]") + sys.exit(1) + + +if __name__ == "__main__": + cli() diff --git a/mcpscope/config.py b/mcpscope/config.py new file mode 100644 index 0000000000000000000000000000000000000000..89053ad67215e8736ec934d4aa3082b6e79219e3 --- /dev/null +++ b/mcpscope/config.py @@ -0,0 +1,44 @@ +from __future__ import annotations +import json +from pathlib import Path +from dataclasses import dataclass, field + +DEFAULT_CONFIG_PATH = Path.home() / ".mcpscope" / "config.json" + + +@dataclass +class Settings: + db_path: str = str(Path.home() / ".mcpscope" / "mcpscope.db") + host: str = "127.0.0.1" + port: int = 8080 + log_level: str = "info" + auto_refresh_seconds: int = 30 + api_key: str | None = None + webhook_urls: list[str] = field(default_factory=list) + slack_webhook_url: str | None = None + max_upload_mb: int = 50 + dashboard_password: str | None = None + log_json: bool = False + + @classmethod + def load(cls, path: str | Path | None = None) -> Settings: + path = Path(path) if path else DEFAULT_CONFIG_PATH + if path.exists(): + try: + with open(path) as f: + data = json.load(f) + return cls( + **{k: v for k, v in data.items() if k in cls.__dataclass_fields__} + ) + except (json.JSONDecodeError, TypeError): + pass + return cls() + + def save(self, path: str | Path | None = None): + path = Path(path) if path else DEFAULT_CONFIG_PATH + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as f: + json.dump(self.as_dict(), f, indent=2) + + def as_dict(self) -> dict: + return {k: getattr(self, k) for k in self.__dataclass_fields__} diff --git a/mcpscope/ingest/__init__.py b/mcpscope/ingest/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..c18b7adfddceedb07cdb5dcb0e7583da9a77a44c --- /dev/null +++ b/mcpscope/ingest/__init__.py @@ -0,0 +1,11 @@ +from .base import BaseParser, ParseError +from .cisco_mcp import CiscoMCPParser +from .cisco_a2a import CiscoA2AParser +from .mcpscan import MCPScanParser +from .mcpwn import MCPwnParser +from .sarif import SarifParser + +__all__ = [ + "BaseParser", "ParseError", + "CiscoMCPParser", "CiscoA2AParser", "MCPScanParser", "MCPwnParser", "SarifParser", +] diff --git a/mcpscope/ingest/__pycache__/__init__.cpython-313.pyc b/mcpscope/ingest/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..776ffa6db7baf4bcb9ac978151ab66d7b24181c2 Binary files /dev/null and b/mcpscope/ingest/__pycache__/__init__.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/base.cpython-313.pyc b/mcpscope/ingest/__pycache__/base.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..b6fac1daf6f8be91af6fbf1098e9701b16dc5306 Binary files /dev/null and b/mcpscope/ingest/__pycache__/base.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/cisco_a2a.cpython-313.pyc b/mcpscope/ingest/__pycache__/cisco_a2a.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ec316c71cb49849dc1d976645489c5f4891a4c59 Binary files /dev/null and b/mcpscope/ingest/__pycache__/cisco_a2a.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/cisco_mcp.cpython-313.pyc b/mcpscope/ingest/__pycache__/cisco_mcp.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..b748debae4d79a12aa2c28c6df10456500df6d4d Binary files /dev/null and b/mcpscope/ingest/__pycache__/cisco_mcp.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/mcpscan.cpython-313.pyc b/mcpscope/ingest/__pycache__/mcpscan.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..13c1834b289b70853c35eea5308a9e3bd69a191a Binary files /dev/null and b/mcpscope/ingest/__pycache__/mcpscan.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/mcpwn.cpython-313.pyc b/mcpscope/ingest/__pycache__/mcpwn.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..9231551e7b4d7749a084cf58b36199e007117abb Binary files /dev/null and b/mcpscope/ingest/__pycache__/mcpwn.cpython-313.pyc differ diff --git a/mcpscope/ingest/__pycache__/sarif.cpython-313.pyc b/mcpscope/ingest/__pycache__/sarif.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..1e4be19e9c48a9e17464a9880a8d4fac2b170b6f Binary files /dev/null and b/mcpscope/ingest/__pycache__/sarif.cpython-313.pyc differ diff --git a/mcpscope/ingest/base.py b/mcpscope/ingest/base.py new file mode 100644 index 0000000000000000000000000000000000000000..04d9e14197790a37c8937909f8e86b70e8f08fd9 --- /dev/null +++ b/mcpscope/ingest/base.py @@ -0,0 +1,69 @@ +from __future__ import annotations +import json +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Any + +from mcpscope.models.finding import Finding +from mcpscope.config import Settings + +MAX_UPLOAD_MB = Settings.load().max_upload_mb + + +class ParseError(Exception): + def __init__( + self, message: str, path: str | None = None, details: str | None = None + ): + self.path = path + self.details = details + super().__init__(message) + + +class BaseParser(ABC): + SCANNER_NAME: str = "base" + + @abstractmethod + def parse(self, data: dict) -> list[Finding]: ... + + def validate(self, data: Any, path: str | None = None): + if not isinstance(data, dict): + raise ParseError( + f"Expected a JSON object at root, got {type(data).__name__}", + path=path, + ) + + def load_json(self, path: str | Path) -> dict: + path = Path(path) + if not path.exists(): + raise ParseError(f"File not found: {path}", path=str(path)) + if path.suffix not in (".json", ".sarif"): + raise ParseError( + f"Unsupported file type: {path.suffix} (expected .json or .sarif)", + path=str(path), + ) + + size_mb = path.stat().st_size / (1024 * 1024) + if size_mb > MAX_UPLOAD_MB: + raise ParseError( + f"File too large: {size_mb:.1f}MB exceeds limit of {MAX_UPLOAD_MB}MB", + path=str(path), + ) + + try: + with open(path) as f: + return json.load(f) + except json.JSONDecodeError as e: + raise ParseError( + f"Invalid JSON: {e.msg} at line {e.lineno} col {e.colno}", + path=str(path), + details=str(e), + ) + + def parse_file(self, path: str | Path) -> list[Finding]: + path = Path(path) + raw = self.load_json(path) + self.validate(raw, path=str(path)) + results = self.parse(raw) + if not results: + raise ParseError("No findings could be extracted", path=str(path)) + return results diff --git a/mcpscope/ingest/cisco_a2a.py b/mcpscope/ingest/cisco_a2a.py new file mode 100644 index 0000000000000000000000000000000000000000..4b69bb9668afbda19aee0a0a22ddde14f6ec58f6 --- /dev/null +++ b/mcpscope/ingest/cisco_a2a.py @@ -0,0 +1,72 @@ +from __future__ import annotations +from mcpscope.ingest.base import BaseParser, ParseError +from mcpscope.models.finding import Finding, Severity + + +class CiscoA2AParser(BaseParser): + SCANNER_NAME = "cisco-a2a" + + def validate(self, data: dict, path: str | None = None): + super().validate(data, path) + if "findings" not in data and "results" not in data: + raise ParseError( + "Missing 'findings' or 'results' key — not a Cisco A2A Scanner output", + path=path, + ) + + def parse(self, data: dict) -> list[Finding]: + findings = [] + scanner = self.SCANNER_NAME + target = data.get("target", data.get("card", data.get("endpoint", "unknown"))) + + raw_findings = data.get("findings", data.get("results", data.get("assessments", []))) + if not isinstance(raw_findings, list): + raise ParseError(f"Expected 'findings' to be a list, got {type(raw_findings).__name__}") + + for item in raw_findings: + if not isinstance(item, dict): + continue + threat_name = item.get("threat_name", "") + sev_raw = (item.get("severity") or item.get("risk") or "info").upper() + sev = Severity.CRITICAL if sev_raw == "CRITICAL" else ( + Severity.HIGH if sev_raw == "HIGH" else ( + Severity.MEDIUM if sev_raw == "MEDIUM" else ( + Severity.LOW if sev_raw == "LOW" else Severity.INFO))) + analyzer = item.get("analyzer", "unknown") + summary = item.get("summary", "") + description = item.get("description", summary) + aitech = item.get("aitech", "") + aitech_name = item.get("aitech_name", "") + aisubtech = item.get("aisubtech", "") + aisubtech_name = item.get("aisubtech_name", "") + + location = "" + if isinstance(item.get("details"), dict): + location = item["details"].get("field", "") + + title = threat_name or f"[{analyzer}] {summary[:60]}" if summary else f"{analyzer} finding" + if aitech: + title = f"[{aitech}] {aitech_name}" + if aisubtech: + title = f"[{aisubtech}] {aisubtech_name or title}" + + findings.append(Finding( + scan_id=target, + scanner=scanner, + tool_name=f"a2a-{analyzer.lower()}", + severity=sev, + title=title, + description=description or summary, + raw_data={ + "threat_name": threat_name, + "analyzer": analyzer, + "aitech": aitech, + "aitech_name": aitech_name, + "aisubtech": aisubtech, + "aisubtech_name": aisubtech_name, + "location": location, + "details": item.get("details"), + }, + )) + + return findings diff --git a/mcpscope/ingest/cisco_mcp.py b/mcpscope/ingest/cisco_mcp.py new file mode 100644 index 0000000000000000000000000000000000000000..29a43b3bac707e9cc2331bcbbd18462c97dad77d --- /dev/null +++ b/mcpscope/ingest/cisco_mcp.py @@ -0,0 +1,128 @@ +from __future__ import annotations +from mcpscope.ingest.base import BaseParser, ParseError +from mcpscope.models.finding import Finding, Severity + +SEVERITY_MAP = { + "CRITICAL": Severity.CRITICAL, + "HIGH": Severity.HIGH, + "MEDIUM": Severity.MEDIUM, + "LOW": Severity.LOW, + "INFO": Severity.INFO, + "SAFE": Severity.INFO, + "UNKNOWN": Severity.INFO, +} + +ANALYZER_DISPLAY_NAMES = { + "api_analyzer": "API", + "yara_analyzer": "YARA", + "llm_analyzer": "LLM", + "behavioral_analyzer": "Behavioral", + "readiness_analyzer": "Readiness", +} + + +class CiscoMCPParser(BaseParser): + SCANNER_NAME = "cisco-mcp" + + def validate(self, data: dict, path: str | None = None): + super().validate(data, path) + if "scan_results" not in data and "results" not in data: + raise ParseError( + "Missing 'scan_results' or 'results' key — not a Cisco MCP Scanner output", + path=path, + ) + + def parse(self, data: dict) -> list[Finding]: + findings = [] + scanner = self.SCANNER_NAME + server_url = data.get("server_url", "unknown") + scan_results = data.get("scan_results", data.get("results", [data])) + + for result in scan_results: + tool_name = result.get("tool_name", result.get("prompt_name", result.get("resource_uri", "unknown"))) + item_type = result.get("item_type", "tool") + status = result.get("status", "unknown") + is_safe = result.get("is_safe", True) + + result_findings = result.get("findings", {}) + if not isinstance(result_findings, dict): + continue + + for analyzer_key, analyzer_data in result_findings.items(): + analyzer_label = ANALYZER_DISPLAY_NAMES.get(analyzer_key, analyzer_key) + sev_raw = analyzer_data.get("severity", "INFO").upper() + sev = SEVERITY_MAP.get(sev_raw, Severity.INFO) + + total = analyzer_data.get("total_findings", 0) + + threat_names = analyzer_data.get("threat_names", []) + threat_summary = analyzer_data.get("threat_summary", "") + threats = analyzer_data.get("threats", {}) + + threat_items = threats.get("items", []) if isinstance(threats, dict) else [] + + if threat_items: + for technique in threat_items: + technique_id = technique.get("technique_id", "") + technique_name = technique.get("technique_name", "") + sub_items = technique.get("items", []) + for sub in sub_items: + sub_id = sub.get("sub_technique_id", "") + sub_name = sub.get("sub_technique_name", "") + max_sev = sub.get("max_severity", sev_raw) + desc = sub.get("description", threat_summary) + + title = f"[{technique_id}] {technique_name}" + if sub_id: + title = f"[{sub_id}] {sub_name}" + + findings.append(Finding( + scan_id=server_url, + scanner=scanner, + tool_name=tool_name, + severity=SEVERITY_MAP.get(max_sev.upper(), sev), + title=title, + description=desc, + raw_data={ + "analyzer": analyzer_label, + "item_type": item_type, + "status": status, + "is_safe": is_safe, + "technique": technique, + "sub_technique": sub, + }, + )) + elif threat_names: + for threat in threat_names: + findings.append(Finding( + scan_id=server_url, + scanner=scanner, + tool_name=tool_name, + severity=sev, + title=f"[{analyzer_label}] {threat}", + description=threat_summary, + raw_data={ + "analyzer": analyzer_label, + "item_type": item_type, + "status": status, + "is_safe": is_safe, + "threat": threat, + }, + )) + elif total > 0 and threat_summary: + findings.append(Finding( + scan_id=server_url, + scanner=scanner, + tool_name=tool_name, + severity=sev, + title=f"[{analyzer_label}] {threat_summary[:80]}", + description=threat_summary, + raw_data={ + "analyzer": analyzer_label, + "item_type": item_type, + "status": status, + "is_safe": is_safe, + }, + )) + + return findings diff --git a/mcpscope/ingest/mcpscan.py b/mcpscope/ingest/mcpscan.py new file mode 100644 index 0000000000000000000000000000000000000000..ad415ff2cfd153a82a2139e31613dd2ca300ad70 --- /dev/null +++ b/mcpscope/ingest/mcpscan.py @@ -0,0 +1,76 @@ +from __future__ import annotations +from mcpscope.ingest.base import BaseParser, ParseError +from mcpscope.models.finding import Finding, Severity + +SEVERITY_MAP = { + "critical": Severity.CRITICAL, + "high": Severity.HIGH, + "medium": Severity.MEDIUM, + "low": Severity.LOW, + "info": Severity.INFO, + "safe": Severity.INFO, +} + + +class MCPScanParser(BaseParser): + SCANNER_NAME = "mcp-scan" + + def validate(self, data: dict, path: str | None = None): + super().validate(data, path) + if "issues" not in data and "results" not in data: + raise ParseError( + "Missing 'issues' or 'results' key — not an mcp-scan JSON output", + path=path, + ) + + def parse(self, data: dict) -> list[Finding]: + findings = [] + scanner = self.SCANNER_NAME + target = data.get("target", data.get("config", "unknown")) + + issues = data.get("issues", data.get("results", data.get("findings", []))) + if not isinstance(issues, list): + raise ParseError(f"Expected 'issues' to be a list, got {type(issues).__name__}") + + servers = data.get("servers", []) + server_map = {} + for srv in servers: + if isinstance(srv, dict): + srv_name = srv.get("name", srv.get("server_name", "")) + server_map[srv.get("id", "")] = srv_name + + for issue in issues: + if not isinstance(issue, dict): + continue + code = issue.get("code", issue.get("id", "")) + sev_raw = issue.get("severity", "info").lower() + sev = SEVERITY_MAP.get(sev_raw, Severity.INFO) + message = issue.get("message", issue.get("summary", issue.get("title", ""))) + description = issue.get("description", issue.get("detail", "")) + recommendation = issue.get("recommendation", issue.get("remediation", "")) + tool_name = issue.get("tool_name", issue.get("tool", "")) + server_id = issue.get("server_id", issue.get("server", "")) + server_name = issue.get("server_name", server_map.get(server_id, "")) + if not tool_name and server_name: + tool_name = f"server:{server_name}" + + title = f"[{code}] {message}" if code else message + if not title: + title = f"mcp-scan {sev_raw} finding" + + findings.append(Finding( + scan_id=target, + scanner=scanner, + tool_name=tool_name or "mcp-scan", + severity=sev, + title=title, + description=description or message, + recommendation=recommendation, + raw_data={ + "code": code, + "server_name": server_name, + **issue, + }, + )) + + return findings diff --git a/mcpscope/ingest/mcpwn.py b/mcpscope/ingest/mcpwn.py new file mode 100644 index 0000000000000000000000000000000000000000..9832bc6b80fc8215f6cfc8ee14d5c8ebba366548 --- /dev/null +++ b/mcpscope/ingest/mcpwn.py @@ -0,0 +1,70 @@ +from __future__ import annotations +from mcpscope.ingest.base import BaseParser, ParseError +from mcpscope.models.finding import Finding, Severity + +SEVERITY_MAP = { + "critical": Severity.CRITICAL, + "high": Severity.HIGH, + "medium": Severity.MEDIUM, + "low": Severity.LOW, + "info": Severity.INFO, +} + + +class MCPwnParser(BaseParser): + SCANNER_NAME = "mcpwn" + + def validate(self, data: dict, path: str | None = None): + super().validate(data, path) + if "findings" not in data and "exploits" not in data and "modules" not in data: + raise ParseError( + "Missing 'findings', 'exploits', or 'modules' key — not an MCPwn output", + path=path, + ) + + def parse(self, data: dict) -> list[Finding]: + findings = [] + scanner = self.SCANNER_NAME + target = data.get("target", data.get("tool", data.get("card", "unknown"))) + + raw_findings = data.get("findings", data.get("exploits", data.get("modules", []))) + if not isinstance(raw_findings, list): + raise ParseError(f"Expected findings to be a list, got {type(raw_findings).__name__}") + + for item in raw_findings: + if not isinstance(item, dict): + continue + sev_raw = item.get("severity", "info").lower() + sev = SEVERITY_MAP.get(sev_raw, Severity.INFO) + vuln_id = item.get("id", item.get("type", "")) + title = item.get("title", item.get("name", "")) + test = item.get("test", item.get("module", "")) + tool_name = item.get("tool", item.get("tool_name", "")) + if not tool_name: + tool_name = test or "mcpwn" + description = item.get("description", item.get("output", item.get("detection", ""))) + recommendation = item.get("recommendation", item.get("fix", "")) + evidence = item.get("detection", item.get("evidence", "")) + + if not title and vuln_id: + title = f"[{vuln_id}] {test}" if test else vuln_id + if not title and description: + title = description[:80] + + findings.append(Finding( + scan_id=target, + scanner=scanner, + tool_name=tool_name, + severity=sev, + title=title or "MCPwn finding", + description=description, + recommendation=recommendation, + raw_data={ + "vuln_id": vuln_id, + "test": test, + "evidence": evidence, + **item, + }, + )) + + return findings diff --git a/mcpscope/ingest/sarif.py b/mcpscope/ingest/sarif.py new file mode 100644 index 0000000000000000000000000000000000000000..3d9085176eadb341fb47c1e01f77ee8a69cab1ac --- /dev/null +++ b/mcpscope/ingest/sarif.py @@ -0,0 +1,103 @@ +from __future__ import annotations +from mcpscope.ingest.base import BaseParser, ParseError +from mcpscope.models.finding import Finding, Severity + +SEVERITY_MAP = { + "error": Severity.CRITICAL, + "warning": Severity.MEDIUM, + "note": Severity.LOW, + "none": Severity.INFO, +} + + +class SarifParser(BaseParser): + SCANNER_NAME = "sarif" + + def validate(self, data: dict, path: str | None = None): + super().validate(data, path) + if "$schema" not in str(data.get("$schema", "")): + if "version" not in data or "runs" not in data: + raise ParseError( + "Missing '$schema'/'version' and 'runs' — not a SARIF file", + path=path, + ) + + def parse(self, data: dict) -> list[Finding]: + findings = [] + scanner = self.SCANNER_NAME + target = data.get("target", "sarif-report") + + runs = data.get("runs", []) + if not isinstance(runs, list): + raise ParseError(f"Expected 'runs' to be a list, got {type(runs).__name__}") + + for run in runs: + tool_name = "unknown" + if isinstance(run.get("tool"), dict): + driver = run["tool"].get("driver", {}) + tool_name = driver.get("name", driver.get("fullName", "unknown")) + tool_version = driver.get("version", "") + + run_results = run.get("results", []) + if not isinstance(run_results, list): + continue + + rules_map = {} + if isinstance(run.get("tool"), dict): + driver = run["tool"].get("driver", {}) + rules = driver.get("rules", []) + if isinstance(rules, list): + for rule in rules: + rule_id = rule.get("id", "") + rules_map[rule_id] = { + "name": rule.get("name", ""), + "shortDescription": "", + "fullDescription": "", + } + if isinstance(rule.get("shortDescription"), dict): + rules_map[rule_id]["shortDescription"] = rule["shortDescription"].get("text", "") + if isinstance(rule.get("fullDescription"), dict): + rules_map[rule_id]["fullDescription"] = rule["fullDescription"].get("text", "") + + for result in run_results: + if not isinstance(result, dict): + continue + rule_id = result.get("ruleId", "") + rule_info = rules_map.get(rule_id, {}) + + level = result.get("level", "warning") + sev = SEVERITY_MAP.get(level, Severity.INFO) + + message = "" + if isinstance(result.get("message"), dict): + message = result["message"].get("text", "") + + title = rule_info.get("name", rule_id) or rule_id + description = rule_info.get("fullDescription", rule_info.get("shortDescription", message)) + if not description: + description = message + + locations = result.get("locations", []) + loc_str = "" + if isinstance(locations, list) and locations: + phys = locations[0].get("physicalLocation", {}) + artifact = phys.get("artifactLocation", {}) + loc_str = artifact.get("uri", "") + + findings.append(Finding( + scan_id=target, + scanner=scanner, + tool_name=tool_name, + tool_version=tool_version, + severity=sev, + title=title, + description=description, + raw_data={ + "rule_id": rule_id, + "level": level, + "location": loc_str, + **result, + }, + )) + + return findings diff --git a/mcpscope/main.py b/mcpscope/main.py new file mode 100644 index 0000000000000000000000000000000000000000..0b7cb3bf4a1ab5e0fc13aab7aa04a7974860dccb --- /dev/null +++ b/mcpscope/main.py @@ -0,0 +1,4 @@ +from mcpscope.cli import cli + +if __name__ == "__main__": + cli() diff --git a/mcpscope/models/__init__.py b/mcpscope/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..0e9570162642fef010f092f1b3e69d268a83cc0a --- /dev/null +++ b/mcpscope/models/__init__.py @@ -0,0 +1,4 @@ +from .finding import Finding, Severity +from .scan import ScanRun, ScanHistory + +__all__ = ["Finding", "Severity", "ScanRun", "ScanHistory"] diff --git a/mcpscope/models/__pycache__/__init__.cpython-313.pyc b/mcpscope/models/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..2b129a281c2d968b46cb85f35484b202f21b1bf0 Binary files /dev/null and b/mcpscope/models/__pycache__/__init__.cpython-313.pyc differ diff --git a/mcpscope/models/__pycache__/finding.cpython-313.pyc b/mcpscope/models/__pycache__/finding.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..9faec620dbe62502b6bb62261de65f37c96cc6be Binary files /dev/null and b/mcpscope/models/__pycache__/finding.cpython-313.pyc differ diff --git a/mcpscope/models/__pycache__/scan.cpython-313.pyc b/mcpscope/models/__pycache__/scan.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..7a3b28826fa1c897b0344def78a62e31a9fde441 Binary files /dev/null and b/mcpscope/models/__pycache__/scan.cpython-313.pyc differ diff --git a/mcpscope/models/__pycache__/security_event.cpython-313.pyc b/mcpscope/models/__pycache__/security_event.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..aacf3932b5ec121ca3a4b5526e77656a173b87a6 Binary files /dev/null and b/mcpscope/models/__pycache__/security_event.cpython-313.pyc differ diff --git a/mcpscope/models/finding.py b/mcpscope/models/finding.py new file mode 100644 index 0000000000000000000000000000000000000000..4a80e8e97161b7bc028f20d3d419199f918bbc34 --- /dev/null +++ b/mcpscope/models/finding.py @@ -0,0 +1,41 @@ +from __future__ import annotations +from enum import Enum +from datetime import datetime, timezone +from pydantic import BaseModel, Field + + +class Severity(str, Enum): + CRITICAL = "critical" + HIGH = "high" + MEDIUM = "medium" + LOW = "low" + INFO = "info" + + +SEVERITY_ORDER = {s: i for i, s in enumerate(Severity)} +SEVERITY_COLORS = { + Severity.CRITICAL: "#dc2626", + Severity.HIGH: "#ea580c", + Severity.MEDIUM: "#ca8a04", + Severity.LOW: "#2563eb", + Severity.INFO: "#6b7280", +} + + +class Finding(BaseModel): + id: str | None = None + scan_id: str + scanner: str + tool_name: str + tool_version: str | None = None + severity: Severity + title: str + description: str | None = None + recommendation: str | None = None + cvss_score: float | None = None + cve_id: str | None = None + raw_data: dict | None = None + created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def severity_order(self) -> int: + return SEVERITY_ORDER.get(self.severity, 99) diff --git a/mcpscope/models/scan.py b/mcpscope/models/scan.py new file mode 100644 index 0000000000000000000000000000000000000000..28e6353dd4cc524f740485e31d3d5e7fd28dbff5 --- /dev/null +++ b/mcpscope/models/scan.py @@ -0,0 +1,27 @@ +from __future__ import annotations +from datetime import datetime, timezone +from pydantic import BaseModel, Field + + +class ScanRun(BaseModel): + id: str + scanner: str + target: str | None = None + findings_count: int = 0 + critical_count: int = 0 + high_count: int = 0 + medium_count: int = 0 + low_count: int = 0 + info_count: int = 0 + raw_file: str | None = None + created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + +class ScanHistory(BaseModel): + scans: list[ScanRun] = [] + total_findings: int = 0 + total_critical: int = 0 + total_high: int = 0 + total_medium: int = 0 + total_low: int = 0 + total_info: int = 0 diff --git a/mcpscope/models/security_event.py b/mcpscope/models/security_event.py new file mode 100644 index 0000000000000000000000000000000000000000..9489044a01cce44ac8ffdd944eaa4dce49fb5fe2 --- /dev/null +++ b/mcpscope/models/security_event.py @@ -0,0 +1,26 @@ +from __future__ import annotations +from enum import Enum +from datetime import datetime, timezone +from pydantic import BaseModel, Field + + +class EventSeverity(str, Enum): + CRITICAL = "critical" + HIGH = "high" + MEDIUM = "medium" + LOW = "low" + INFO = "info" + + +class SecurityEvent(BaseModel): + id: str | None = None + event_type: str + severity: str + message: str + source: str = "mcpguard" + tool: str | None = None + details: dict | None = None + blocked: bool = True + created_at: str = Field( + default_factory=lambda: datetime.now(timezone.utc).isoformat() + ) diff --git a/mcpscope/scanner.py b/mcpscope/scanner.py new file mode 100644 index 0000000000000000000000000000000000000000..8fb6ab9d447a9a554596c4d008d85f0ce083e619 --- /dev/null +++ b/mcpscope/scanner.py @@ -0,0 +1,80 @@ +from __future__ import annotations +import json +import subprocess +import tempfile +import uuid +from pathlib import Path +from typing import Any + +from mcpscope.ingest.base import ParseError +from mcpscope.ingest.cisco_mcp import CiscoMCPParser +from mcpscope.ingest.mcpscan import MCPScanParser +from mcpscope.ingest.mcpwn import MCPwnParser +from mcpscope.models.scan import ScanRun +from mcpscope.storage.store import Store + + +class ScannerRunner: + PARSERS: dict[str, dict[str, Any]] = { + "mcp-scan": { + "parser": MCPScanParser(), + "install": "pip install mcp-scan", + "cmd": ["mcp-scan", "--json"], + }, + "cisco-mcp": { + "parser": CiscoMCPParser(), + "install": "pip install cisco-ai-mcp-scanner", + "cmd": ["mcp-scanner", "scan", "--output-format", "raw"], + }, + "mcpwn": { + "parser": MCPwnParser(), + "install": "pip install mcpwn", + "cmd": ["mcpwn", "scan", "--format", "json"], + }, + } + + def scan( + self, scanner_name: str, target: str, store: Store | None = None + ) -> ScanRun: + info = self.PARSERS.get(scanner_name) + if not info: + raise ValueError( + f"Unknown scanner: {scanner_name}. Supported: {', '.join(self.PARSERS)}" + ) + + parser = info["parser"] + cmd = info["cmd"] + [target] + + with tempfile.NamedTemporaryFile(suffix=".json", delete=False, mode="w") as tmp: + tmp_path = tmp.name + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + if result.returncode != 0: + raise RuntimeError( + f"Scanner failed (exit {result.returncode}): {result.stderr[:500]}" + ) + raw = json.loads(result.stdout) + except FileNotFoundError: + raise RuntimeError( + f"Scanner '{scanner_name}' not found. Install it: {info['install']}" + ) + except json.JSONDecodeError as e: + raise RuntimeError(f"Scanner output is not valid JSON: {e}") + except subprocess.TimeoutExpired: + raise RuntimeError("Scanner timed out (120s)") + finally: + Path(tmp_path).unlink(missing_ok=True) + + findings = parser.parse(raw) + if not findings: + raise RuntimeError("Scanner completed but no findings were detected") + + scan = ScanRun( + id=str(uuid.uuid4()), + scanner=findings[0].scanner, + target=target, + ) + if store: + store.save_scan(scan, findings) + return scan diff --git a/mcpscope/storage/__init__.py b/mcpscope/storage/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..895df7cc78f9f62f266b062d11edb9716f560f5e --- /dev/null +++ b/mcpscope/storage/__init__.py @@ -0,0 +1,3 @@ +from .store import Store + +__all__ = ["Store"] diff --git a/mcpscope/storage/__pycache__/__init__.cpython-313.pyc b/mcpscope/storage/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c77150afcb2eaf7f2295e7f98a8b804a6c4c8603 Binary files /dev/null and b/mcpscope/storage/__pycache__/__init__.cpython-313.pyc differ diff --git a/mcpscope/storage/__pycache__/store.cpython-313.pyc b/mcpscope/storage/__pycache__/store.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..6e84f86317d7f5704dd2b3ed0159f250b3eee5ff Binary files /dev/null and b/mcpscope/storage/__pycache__/store.cpython-313.pyc differ diff --git a/mcpscope/storage/store.py b/mcpscope/storage/store.py new file mode 100644 index 0000000000000000000000000000000000000000..12d5ec934208741a42dd8e91ad558210c97a8051 --- /dev/null +++ b/mcpscope/storage/store.py @@ -0,0 +1,536 @@ +from __future__ import annotations +import sqlite3 +import uuid +import json +import shutil +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from mcpscope.models.finding import Finding, Severity +from mcpscope.models.scan import ScanRun, ScanHistory +from mcpscope.models.security_event import SecurityEvent +from mcpscope.config import Settings + +DEFAULT_DB = Path.home() / ".mcpscope" / "mcpscope.db" +PAGE_SIZE = 50 + + +class Store: + def __init__(self, db_path: str | Path = DEFAULT_DB): + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._init_db() + self._webhooks = Settings.load().webhook_urls + + def _init_db(self): + self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS scan_runs ( + id TEXT PRIMARY KEY, + scanner TEXT NOT NULL, + target TEXT, + findings_count INTEGER DEFAULT 0, + critical_count INTEGER DEFAULT 0, + high_count INTEGER DEFAULT 0, + medium_count INTEGER DEFAULT 0, + low_count INTEGER DEFAULT 0, + info_count INTEGER DEFAULT 0, + raw_file TEXT, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS findings ( + id TEXT PRIMARY KEY, + scan_id TEXT NOT NULL, + scanner TEXT NOT NULL, + tool_name TEXT NOT NULL, + tool_version TEXT, + severity TEXT NOT NULL, + title TEXT NOT NULL, + description TEXT, + recommendation TEXT, + cvss_score REAL, + cve_id TEXT, + raw_data TEXT, + created_at TEXT NOT NULL, + FOREIGN KEY (scan_id) REFERENCES scan_runs(id) + ); + CREATE INDEX IF NOT EXISTS idx_findings_scan_id ON findings(scan_id); + CREATE INDEX IF NOT EXISTS idx_findings_severity ON findings(severity); + CREATE INDEX IF NOT EXISTS idx_findings_tool ON findings(tool_name); + CREATE INDEX IF NOT EXISTS idx_findings_created ON findings(created_at); + CREATE INDEX IF NOT EXISTS idx_findings_title ON findings(title); + CREATE TABLE IF NOT EXISTS security_events ( + id TEXT PRIMARY KEY, + event_type TEXT NOT NULL, + severity TEXT NOT NULL, + message TEXT NOT NULL, + source TEXT NOT NULL DEFAULT 'mcpguard', + tool TEXT, + details TEXT, + blocked INTEGER DEFAULT 1, + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_events_created ON security_events(created_at); + CREATE INDEX IF NOT EXISTS idx_events_severity ON security_events(severity); + CREATE INDEX IF NOT EXISTS idx_events_type ON security_events(event_type); + """) + + def save_scan(self, scan: ScanRun, findings: list[Finding]) -> ScanRun: + scan.id = scan.id or str(uuid.uuid4()) + scan.created_at = scan.created_at or datetime.now(timezone.utc).isoformat() + + counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0} + for f in findings: + f.id = f.id or str(uuid.uuid4()) + f.scan_id = scan.id + sev = f.severity.value if isinstance(f.severity, Severity) else f.severity + if sev in counts: + counts[sev] += 1 + + scan.findings_count = len(findings) + scan.critical_count = counts["critical"] + scan.high_count = counts["high"] + scan.medium_count = counts["medium"] + scan.low_count = counts["low"] + scan.info_count = counts["info"] + + with self._conn: + self._conn.execute( + """INSERT OR REPLACE INTO scan_runs + (id, scanner, target, findings_count, critical_count, high_count, + medium_count, low_count, info_count, raw_file, created_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + ( + scan.id, + scan.scanner, + scan.target, + scan.findings_count, + scan.critical_count, + scan.high_count, + scan.medium_count, + scan.low_count, + scan.info_count, + scan.raw_file, + scan.created_at, + ), + ) + for f in findings: + self._conn.execute( + """INSERT OR REPLACE INTO findings + (id, scan_id, scanner, tool_name, tool_version, severity, + title, description, recommendation, cvss_score, cve_id, raw_data, created_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + f.id, + f.scan_id, + f.scanner, + f.tool_name, + f.tool_version, + f.severity.value + if isinstance(f.severity, Severity) + else f.severity, + f.title, + f.description, + f.recommendation, + f.cvss_score, + f.cve_id, + json.dumps(f.raw_data) if f.raw_data else None, + f.created_at, + ), + ) + + cfg = Settings.load() + if self._webhooks or cfg.slack_webhook_url: + from mcpscope.webhooks import notify_scan_imported + + notify_scan_imported( + self._webhooks, + findings=findings, + findings_count=scan.findings_count, + critical_count=scan.critical_count, + high_count=scan.high_count, + scan_id=scan.id, + scanner=scan.scanner, + slack_url=cfg.slack_webhook_url or "", + ) + + return scan + + def get_all_scans(self) -> list[ScanRun]: + rows = self._conn.execute( + "SELECT * FROM scan_runs ORDER BY created_at DESC" + ).fetchall() + return [ScanRun(**dict(r)) for r in rows] + + def get_scans_paginated( + self, page: int = 1, page_size: int = PAGE_SIZE + ) -> tuple[list[ScanRun], int]: + offset = (page - 1) * page_size + count_row = self._conn.execute("SELECT COUNT(*) as c FROM scan_runs").fetchone() + total = count_row["c"] if count_row else 0 + rows = self._conn.execute( + "SELECT * FROM scan_runs ORDER BY created_at DESC LIMIT ? OFFSET ?", + (page_size, offset), + ).fetchall() + return [ScanRun(**dict(r)) for r in rows], total + + def get_scan(self, scan_id: str) -> ScanRun | None: + row = self._conn.execute( + "SELECT * FROM scan_runs WHERE id = ?", (scan_id,) + ).fetchone() + return ScanRun(**dict(row)) if row else None + + def get_findings( + self, + scan_id: str | None = None, + severity: str | None = None, + tool_name: str | None = None, + scanner: str | None = None, + search: str | None = None, + page: int = 1, + page_size: int = PAGE_SIZE, + ) -> tuple[list[Finding], int]: + query = "SELECT * FROM findings WHERE 1=1" + count_query = "SELECT COUNT(*) as c FROM findings WHERE 1=1" + params = [] + if scan_id: + query += " AND scan_id = ?" + count_query += " AND scan_id = ?" + params.append(scan_id) + if severity: + query += " AND severity = ?" + count_query += " AND severity = ?" + params.append(severity) + if tool_name: + query += " AND tool_name = ?" + count_query += " AND tool_name = ?" + params.append(tool_name) + if scanner: + query += " AND scanner = ?" + count_query += " AND scanner = ?" + params.append(scanner) + if search: + query += " AND (title LIKE ? OR description LIKE ? OR tool_name LIKE ?)" + count_query += ( + " AND (title LIKE ? OR description LIKE ? OR tool_name LIKE ?)" + ) + like = f"%{search}%" + params.extend([like, like, like]) + + count_row = self._conn.execute(count_query, params).fetchone() + total = count_row["c"] if count_row else 0 + + offset = (page - 1) * page_size + query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" + rows = self._conn.execute(query, params + [page_size, offset]).fetchall() + + result = [] + for r in rows: + d = dict(r) + if d.get("raw_data"): + try: + d["raw_data"] = json.loads(d["raw_data"]) + except (json.JSONDecodeError, TypeError): + pass + result.append(Finding(**d)) + return result, total + + def get_finding(self, finding_id: str) -> Finding | None: + row = self._conn.execute( + "SELECT * FROM findings WHERE id = ?", (finding_id,) + ).fetchone() + if not row: + return None + d = dict(row) + if d.get("raw_data"): + try: + d["raw_data"] = json.loads(d["raw_data"]) + except (json.JSONDecodeError, TypeError): + pass + return Finding(**d) + + def get_scan_history(self) -> ScanHistory: + scans = self.get_all_scans() + history = ScanHistory(scans=scans) + for s in scans: + history.total_findings += s.findings_count + history.total_critical += s.critical_count + history.total_high += s.high_count + history.total_medium += s.medium_count + history.total_low += s.low_count + history.total_info += s.info_count + return history + + def get_top_tools(self, limit: int = 10) -> list[dict]: + rows = self._conn.execute( + """SELECT tool_name, COUNT(*) as total, + SUM(CASE WHEN severity IN ('critical','high') THEN 1 ELSE 0 END) as critical_high + FROM findings GROUP BY tool_name ORDER BY critical_high DESC LIMIT ?""", + (limit,), + ).fetchall() + return [dict(r) for r in rows] + + def get_severity_trend(self) -> list[dict]: + rows = self._conn.execute( + """SELECT DATE(created_at) as date, severity, COUNT(*) as count + FROM findings + GROUP BY DATE(created_at), severity + ORDER BY date ASC""", + ).fetchall() + return [dict(r) for r in rows] + + def get_scanners(self) -> list[str]: + rows = self._conn.execute( + "SELECT DISTINCT scanner FROM scan_runs ORDER BY scanner" + ).fetchall() + return [r["scanner"] for r in rows] + + def get_tool_names(self) -> list[str]: + rows = self._conn.execute( + "SELECT DISTINCT tool_name FROM findings ORDER BY tool_name" + ).fetchall() + return [r["tool_name"] for r in rows] + + def get_duplicates(self, threshold: float = 0.8) -> list[dict]: + rows = self._conn.execute( + """SELECT tool_name, title, severity, scanner, + COUNT(*) as count, + GROUP_CONCAT(id) as finding_ids, + GROUP_CONCAT(scan_id) as scan_ids, + GROUP_CONCAT(created_at) as created_ats + FROM findings + GROUP BY tool_name, title, severity + HAVING COUNT(*) > 1 + ORDER BY COUNT(*) DESC""" + ).fetchall() + return [dict(r) for r in rows] + + def diff_scans(self, scan_a: str, scan_b: str) -> dict: + sa = self.get_scan(scan_a) + sb = self.get_scan(scan_b) + if not sa or not sb: + raise ValueError("One or both scans not found") + + fa_list = self.get_findings(scan_id=scan_a, page_size=10000)[0] + fb_list = self.get_findings(scan_id=scan_b, page_size=10000)[0] + + key_a = { + ( + f.tool_name, + f.title, + f.severity.value if isinstance(f.severity, Severity) else f.severity, + ) + for f in fa_list + } + key_b = { + ( + f.tool_name, + f.title, + f.severity.value if isinstance(f.severity, Severity) else f.severity, + ) + for f in fb_list + } + + new_in_b = [ + f.model_dump() + for f in fb_list + if ( + f.tool_name, + f.title, + f.severity.value if isinstance(f.severity, Severity) else f.severity, + ) + not in key_a + ] + fixed_in_b = [ + f.model_dump() + for f in fa_list + if ( + f.tool_name, + f.title, + f.severity.value if isinstance(f.severity, Severity) else f.severity, + ) + not in key_b + ] + + return { + "scan_a": { + "id": scan_a, + "scanner": sa.scanner, + "target": sa.target, + "findings_count": sa.findings_count, + }, + "scan_b": { + "id": scan_b, + "scanner": sb.scanner, + "target": sb.target, + "findings_count": sb.findings_count, + }, + "new_findings": new_in_b, + "fixed_findings": fixed_in_b, + "new_count": len(new_in_b), + "fixed_count": len(fixed_in_b), + "unchanged_count": len(fb_list) - len(new_in_b), + } + + def save_event(self, event: SecurityEvent) -> SecurityEvent: + import uuid + + event.id = event.id or str(uuid.uuid4()) + with self._conn: + self._conn.execute( + """INSERT OR REPLACE INTO security_events + (id, event_type, severity, message, source, tool, details, blocked, created_at) + VALUES (?,?,?,?,?,?,?,?,?)""", + ( + event.id, + event.event_type, + event.severity, + event.message, + event.source, + event.tool, + json.dumps(event.details) if event.details else None, + 1 if event.blocked else 0, + event.created_at, + ), + ) + return event + + def get_events( + self, + limit: int = 100, + offset: int = 0, + severity: str | None = None, + event_type: str | None = None, + ) -> tuple[list[SecurityEvent], int]: + query = "SELECT * FROM security_events WHERE 1=1" + count_query = "SELECT COUNT(*) as c FROM security_events WHERE 1=1" + params: list = [] + if severity: + query += " AND severity = ?" + count_query += " AND severity = ?" + params.append(severity) + if event_type: + query += " AND event_type = ?" + count_query += " AND event_type = ?" + params.append(event_type) + count_row = self._conn.execute(count_query, params).fetchone() + total = count_row["c"] if count_row else 0 + query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" + rows = self._conn.execute(query, params + [limit, offset]).fetchall() + result = [] + for r in rows: + d = dict(r) + if d.get("details"): + try: + d["details"] = json.loads(d["details"]) + except (json.JSONDecodeError, TypeError): + pass + d["blocked"] = bool(d["blocked"]) + result.append(SecurityEvent(**d)) + return result, total + + def get_event_stats(self) -> dict: + total = self._conn.execute( + "SELECT COUNT(*) as c FROM security_events" + ).fetchone()["c"] + blocked = self._conn.execute( + "SELECT COUNT(*) as c FROM security_events WHERE blocked = 1" + ).fetchone()["c"] + by_type = self._conn.execute( + "SELECT event_type, COUNT(*) as count FROM security_events GROUP BY event_type ORDER BY count DESC" + ).fetchall() + by_severity = self._conn.execute( + "SELECT severity, COUNT(*) as count FROM security_events GROUP BY severity ORDER BY count DESC" + ).fetchall() + return { + "total": total, + "blocked": blocked, + "by_type": [dict(r) for r in by_type], + "by_severity": [dict(r) for r in by_severity], + } + + def clear_events(self): + with self._conn: + self._conn.execute("DELETE FROM security_events") + + def clear(self): + with self._conn: + self._conn.execute("DELETE FROM findings") + self._conn.execute("DELETE FROM scan_runs") + + def backup(self, path: str | Path): + path = Path(path) + self._conn.commit() + shutil.copy2(self.db_path, path) + + def restore(self, path: str | Path): + path = Path(path) + if not path.exists(): + raise FileNotFoundError(f"Backup not found: {path}") + self._conn.close() + shutil.copy2(path, self.db_path) + self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + + def prune(self, keep_days: int = 30) -> int: + from datetime import datetime, timedelta, timezone + + cutoff = (datetime.now(timezone.utc) - timedelta(days=keep_days)).isoformat() + with self._conn: + self._conn.execute( + "DELETE FROM findings WHERE scan_id IN (SELECT id FROM scan_runs WHERE created_at < ?)", + (cutoff,), + ) + cur = self._conn.execute( + "DELETE FROM scan_runs WHERE created_at < ?", (cutoff,) + ) + deleted_scans = cur.rowcount + return deleted_scans + + def seed_demo_data(self): + import random + + scanners_demo = ["cisco-mcp", "cisco-a2a", "mcp-scan", "mcpwn"] + tools_pool = [ + ("execute_command", "critical"), + ("read_file", "high"), + ("fetch_url", "critical"), + ("search_db", "medium"), + ("write_file", "high"), + ("list_dir", "low"), + ("send_email", "medium"), + ("run_shell", "critical"), + ("parse_xml", "low"), + ("make_request", "high"), + ] + now = datetime.now(timezone.utc) + for i in range(6): + scanner = random.choice(scanners_demo) + scan_id = f"demo-scan-{i + 1}" + target = f"demo-server-{i % 3 + 1}.local" + created = (now - timedelta(hours=i * 12)).isoformat() + count = random.randint(2, 6) + selected = random.sample(tools_pool, min(count, len(tools_pool))) + findings = [] + for tool_name, sev_str in selected: + sev = Severity(sev_str) + findings.append( + Finding( + scan_id=scan_id, + scanner=scanner, + tool_name=tool_name, + severity=sev, + title=f"{sev.name.title()}: {tool_name.replace('_', ' ').title()}", + description=f"Security issue detected in {tool_name}", + recommendation=f"Review and fix {tool_name} configuration", + created_at=created, + ) + ) + scan = ScanRun( + id=scan_id, + scanner=scanner, + target=target, + created_at=created, + ) + self.save_scan(scan, findings) diff --git a/mcpscope/templates/dashboard.html b/mcpscope/templates/dashboard.html new file mode 100644 index 0000000000000000000000000000000000000000..a33708645d0dda1330232b0b0cc814ea161415f3 --- /dev/null +++ b/mcpscope/templates/dashboard.html @@ -0,0 +1,575 @@ + + + + + + MCP-Scope Dashboard + + {% if auto_refresh and auto_refresh > 0 %} + + {% endif %} + + + +
+
+

MCP-Scope

+

Unified MCP/A2A Security Scanner Dashboard

+
+
+ {% if auto_refresh and auto_refresh > 0 %}Auto-refresh: {{ auto_refresh }}s{% endif %} +
+
+
+ {% if detail_finding is defined and detail_finding %} +
+ ← Dashboard + Finding Detail +
+
+

{{ detail_finding.title }}

+
+
+
Severity
+
{{ detail_finding.severity.value|upper }}
+
+
+
Scanner
+
{{ detail_finding.scanner }}
+
+
+
Tool / Resource
+
{{ detail_finding.tool_name }}
+
+
+
Scan ID
+
{{ detail_finding.scan_id }}
+
+ {% if detail_scan %} +
+
Scan Target
+
{{ detail_scan.target }}
+
+
+
Scan Date
+
{{ detail_scan.created_at[:19] }}
+
+ {% endif %} +
+
Finding ID
+
{{ detail_finding.id }}
+
+
+
Created At
+
{{ detail_finding.created_at[:19] }}
+
+
+ {% if detail_finding.description %} +
+
Description
+
{{ detail_finding.description }}
+
+ {% endif %} + {% if detail_finding.recommendation %} +
+
Recommendation
+
{{ detail_finding.recommendation }}
+
+ {% endif %} + {% if detail_finding.cvss_score %} +
+
CVSS Score
+
{{ detail_finding.cvss_score }}
+
+ {% endif %} + {% if detail_finding.cve_id %} +
+
CVE
+
{{ detail_finding.cve_id }}
+
+ {% endif %} + {% if detail_finding.raw_data %} +
+
Raw Data
+ 
{{ detail_finding.raw_data | tojson(indent=2) }}
+
+ {% endif %} +
+ {% elif total_scans == 0 %} +
+

No scan data yet

+

Import scanner results to get started:

+

mcpscope import cisco-mcp results.json

+

mcpscope import mcpwn results.json

+

mcpscope import sarif report.sarif

+

Or generate demo data:

+

mcpscope seed

+

Then start the dashboard with mcpscope serve

+
+ {% else %} +
+
Total Scans
{{ total_scans }}
+
Total Findings
{{ total_findings }}
+
Critical
{{ critical }}
+
High
{{ high }}
+
Medium
{{ medium }}
+
Low
{{ low }}
+
Info
{{ info }}
+
+ + + +
+ Overview + Findings + Scans + Duplicates ({{ duplicates|length }}) + Diff + Live Events +
+ +
+
+
+

Findings by Severity

+
+
+

Top Vulnerable Tools

+
+
+

Severity Trend Over Time

+
+
+
+ + + + + + + + + + + {% endif %} +
+ + + + diff --git a/mcpscope/templates/login.html b/mcpscope/templates/login.html new file mode 100644 index 0000000000000000000000000000000000000000..5a10aa10f8d04efe81b020d440034bb4b193411f --- /dev/null +++ b/mcpscope/templates/login.html @@ -0,0 +1,46 @@ + + + + + + MCP-Scope Login + + + + + + + diff --git a/mcpscope/webhooks.py b/mcpscope/webhooks.py new file mode 100644 index 0000000000000000000000000000000000000000..e32f28e4bb51fea8a1014444f019f3e6826bf62f --- /dev/null +++ b/mcpscope/webhooks.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import httpx +from mcpscope.models.finding import Finding, Severity + + +async def fire_webhooks(webhooks: list[str], event: str, payload: dict): + if not webhooks: + return + async with httpx.AsyncClient(timeout=10) as client: + for url in webhooks: + try: + await client.post(url, json={"event": event, **payload}) + except Exception: + pass + + +def _build_slack_message( + scan_id: str, scanner: str, summary: dict, alerts: list[dict] +) -> dict: + color = "#dc2626" if summary.get("critical_count", 0) > 0 else "#ea580c" + fields = [ + {"title": "Scanner", "value": scanner, "short": True}, + { + "title": "Findings", + "value": str(summary.get("findings_count", 0)), + "short": True, + }, + { + "title": "Critical", + "value": str(summary.get("critical_count", 0)), + "short": True, + }, + {"title": "High", "value": str(summary.get("high_count", 0)), "short": True}, + ] + attachments = [] + for a in alerts[:5]: + attachments.append( + { + "color": "#dc2626" if a["severity"] == "critical" else "#ea580c", + "title": f"[{a['severity'].upper()}] {a['title']}", + "text": a.get("description", "") or "", + "fields": [{"title": "Tool", "value": a["tool"], "short": True}], + } + ) + return { + "username": "MCP-Scope", + "icon_emoji": ":shield:", + "text": f"*MCP-Scope Alert* — Scan `{scan_id[:12]}...` ({scanner})", + "attachments": [{"color": color, "fields": fields}] + attachments, + } + + +async def fire_slack( + slack_url: str, scan_id: str, scanner: str, summary: dict, alerts: list[dict] +): + if not slack_url: + return + msg = _build_slack_message(scan_id, scanner, summary, alerts) + try: + async with httpx.AsyncClient(timeout=10) as client: + await client.post(slack_url, json=msg) + except Exception: + pass + + +def notify_scan_imported( + webhooks: list[str], + findings: list[Finding], + findings_count: int, + critical_count: int, + high_count: int, + scan_id: str = "", + scanner: str = "", + slack_url: str = "", +): + if (critical_count == 0 and high_count == 0) or (not webhooks and not slack_url): + return + + critical_findings = [f for f in findings if f.severity == Severity.CRITICAL] + high_findings = [f for f in findings if f.severity == Severity.HIGH] + + summary = { + "scan_id": scan_id, + "scanner": scanner, + "findings_count": findings_count, + "critical_count": critical_count, + "high_count": high_count, + } + + alerts = [] + for f in critical_findings[:10]: + alerts.append( + { + "severity": "critical", + "tool": f.tool_name, + "title": f.title, + "description": f.description, + "finding_id": f.id, + } + ) + for f in high_findings[:10]: + alerts.append( + { + "severity": "high", + "tool": f.tool_name, + "title": f.title, + "description": f.description, + "finding_id": f.id, + } + ) + + import asyncio + + if webhooks: + payload = {"event": "scan_imported", "summary": summary, "alerts": alerts} + try: + asyncio.create_task(fire_webhooks(webhooks, "scan_imported", payload)) + except RuntimeError: + pass + if slack_url: + try: + asyncio.create_task( + fire_slack(slack_url, scan_id, scanner, summary, alerts) + ) + except RuntimeError: + pass diff --git a/opencode.jsonc b/opencode.jsonc new file mode 100644 index 0000000000000000000000000000000000000000..d2d2c40598aa9101307eee629f9c61ae1912e5d6 --- /dev/null +++ b/opencode.jsonc @@ -0,0 +1,30 @@ +{ + "$schema": "https://opencode.ai/config.json", + + "agent": { + "build": { + "tools": { + "skill": true + } + } + }, + + "command": { + "test": { + "template": "Run `python -m pytest tests/ -v` from the mcpscope directory and report results.", + "description": "Run MCPscop tests" + }, + "serve": { + "template": "Start the MCPscop dashboard with `python -m mcpscope serve`.", + "description": "Start MCPscop dashboard" + }, + "seed": { + "template": "Seed demo data with `python -m mcpscope seed`.", + "description": "Seed demo scan data" + }, + "import": { + "template": "Import scanner results into MCPscop.", + "description": "Import scan results" + } + } +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000000000000000000000000000000000000..d8fba8e114d4336d58be8e0517016d6d62e37d82 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,47 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "mcpscope" +version = "0.1.0" +description = "Unified security dashboard for MCP/A2A scanner results" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +keywords = ["mcp", "a2a", "security", "scanning"] +dependencies = [ + "fastapi>=0.110.0", + "uvicorn[standard]>=0.27.0", + "httpx>=0.27.0", + "jinja2>=3.1.0", + "plotly>=5.20.0", + "pydantic>=2.6.0", + "rich>=13.7.0", + "mcp-taxonomy", +] +[project.optional-dependencies] +pdf = ["weasyprint>=62.0"] +dev = ["pytest>=8.0", "pytest-asyncio>=0.24", "httpx>=0.27.0"] + +[project.scripts] +mcpscope = "mcpscope.cli:cli" + +[project.urls] +Homepage = "https://github.com/Carlos-Projects/mcpscope" +Repository = "https://github.com/Carlos-Projects/mcpscope" + +[tool.setuptools.packages.find] +include = ["mcpscope", "mcpscope.*"] + +[tool.mypy] +python_version = "3.10" +strict = false +ignore_missing_imports = true +warn_unused_ignores = true +exclude = ["tests/"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +asyncio_mode = "auto" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..aba6fe89572cbefc8881ade03e05f050e642da24 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +Jinja2==3.1.6 +fastapi==0.136.1 +httpx-sse==0.4.3 +httpx==0.28.1 +plotly==6.7.0 +pydantic-settings==2.12.0 +pydantic==2.12.5 +pydantic_core==2.41.5 +pytest-httpx==0.36.2 +rich==14.3.4 +uvicorn==0.34.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/tests/__pycache__/__init__.cpython-313.pyc b/tests/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..6e27e13d876427e0e1e89ede1c8b7a76dc00c138 Binary files /dev/null and b/tests/__pycache__/__init__.cpython-313.pyc differ diff --git a/tests/__pycache__/test_http.cpython-313-pytest-9.0.3.pyc b/tests/__pycache__/test_http.cpython-313-pytest-9.0.3.pyc new file mode 100644 index 0000000000000000000000000000000000000000..68351f281fc9defc1fdc49f4ddabb2659f24b9ee Binary files /dev/null and b/tests/__pycache__/test_http.cpython-313-pytest-9.0.3.pyc differ diff --git a/tests/__pycache__/test_parsers.cpython-313-pytest-9.0.3.pyc b/tests/__pycache__/test_parsers.cpython-313-pytest-9.0.3.pyc new file mode 100644 index 0000000000000000000000000000000000000000..4bb07c26541cd8f6c0a3254c9425f551d8d16a8e Binary files /dev/null and b/tests/__pycache__/test_parsers.cpython-313-pytest-9.0.3.pyc differ diff --git a/tests/fixtures/cisco-a2a.json b/tests/fixtures/cisco-a2a.json new file mode 100644 index 0000000000000000000000000000000000000000..6f89f15aba20c452e8a43ec02c8f50bf0c837f2a --- /dev/null +++ b/tests/fixtures/cisco-a2a.json @@ -0,0 +1,14 @@ +{ + "target": "https://agent.example.com", + "timestamp": "2026-05-23T10:00:00Z", + "findings": [ + { + "threat_name": "PROMPT INJECTION", + "severity": "HIGH", + "analyzer": "YARA", + "aitech": "AITech-1.1", + "aitech_name": "Direct Prompt Injection", + "description": "Prompt injection patterns found" + } + ] +} diff --git a/tests/fixtures/cisco-mcp.json b/tests/fixtures/cisco-mcp.json new file mode 100644 index 0000000000000000000000000000000000000000..54b60a25d061025c8490b4a9110570756dd4b8a1 --- /dev/null +++ b/tests/fixtures/cisco-mcp.json @@ -0,0 +1,29 @@ +{ + "server_url": "https://mcp-server.example.com/mcp", + "scan_results": [ + { + "tool_name": "add", + "status": "completed", + "findings": { + "api_analyzer": { + "severity": "HIGH", + "total_findings": 1, + "threats": { "items": [{ + "technique_id": "AITech-8.2", + "technique_name": "Data Exfiltration", + "items": [{ + "sub_technique_id": "AISubtech-8.2.3", + "sub_technique_name": "Data Exfiltration via Agent Tooling", + "max_severity": "HIGH", + "description": "Sensitive data exposure" + }] + }]}, + "threat_names": ["Data Exfiltration"], + "threat_summary": "Tool may expose data" + }, + "yara_analyzer": {"severity": "SAFE", "total_findings": 0} + }, + "is_safe": false + } + ] +} diff --git a/tests/fixtures/mcpscan.json b/tests/fixtures/mcpscan.json new file mode 100644 index 0000000000000000000000000000000000000000..6ef0ca6146a89f3ab85a7da4a4988cd8e44d6d86 --- /dev/null +++ b/tests/fixtures/mcpscan.json @@ -0,0 +1,13 @@ +{ + "target": "mcp.json", + "servers": [{"id": "s1", "name": "Demo"}], + "issues": [ + { + "code": "E001", + "severity": "high", + "message": "Prompt injection risk", + "tool_name": "add", + "description": "Tool description issue" + } + ] +} diff --git a/tests/fixtures/mcpwn.json b/tests/fixtures/mcpwn.json new file mode 100644 index 0000000000000000000000000000000000000000..4f028549491bd7644fdbc8d5e825ebf7c06d7294 --- /dev/null +++ b/tests/fixtures/mcpwn.json @@ -0,0 +1,14 @@ +{ + "tool": "Mcpwn", + "target": "server.py", + "findings": [ + { + "id": "MCP-001", + "severity": "CRITICAL", + "title": "Tool Poisoning", + "tool": "fetch_url", + "description": "Hidden instructions found", + "recommendation": "Audit descriptions" + } + ] +} diff --git a/tests/fixtures/sarif.json b/tests/fixtures/sarif.json new file mode 100644 index 0000000000000000000000000000000000000000..4e3610dfaa78aec17d6617dff8e29d124306842d --- /dev/null +++ b/tests/fixtures/sarif.json @@ -0,0 +1,23 @@ +{ + "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/main/sarif-2.1.json", + "version": "2.1.0", + "runs": [{ + "tool": { + "driver": { + "name": "TestScanner", + "version": "1.0", + "rules": [{ + "id": "RULE001", + "name": "Test Rule", + "shortDescription": {"text": "A test rule"}, + "fullDescription": {"text": "Full description"} + }] + } + }, + "results": [{ + "ruleId": "RULE001", + "level": "error", + "message": {"text": "Vulnerability found"} + }] + }] +} diff --git a/tests/test_http.py b/tests/test_http.py new file mode 100644 index 0000000000000000000000000000000000000000..8506980d166c0e1bc8ed2f8dfb25d0fc8e9ef524 --- /dev/null +++ b/tests/test_http.py @@ -0,0 +1,329 @@ +import pytest + +from mcpscope.api.server import create_app +from mcpscope.storage.store import Store +from mcpscope.models.finding import Finding, Severity +from mcpscope.models.scan import ScanRun + + +@pytest.fixture +def store(): + import tempfile + import os + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + s = Store(db_path=db_path) + s.save_scan( + ScanRun(id="http-test", scanner="test", target="t"), + [ + Finding( + scan_id="http-test", + scanner="test", + tool_name="tool_a", + severity=Severity.CRITICAL, + title="Critical finding", + ), + Finding( + scan_id="http-test", + scanner="test", + tool_name="tool_b", + severity=Severity.LOW, + title="Low priority", + ), + ], + ) + yield s + s._conn.close() + os.unlink(db_path) + + +@pytest.fixture +def client(store): + from httpx import AsyncClient, ASGITransport + + app = create_app(store) + transport = ASGITransport(app=app) + return AsyncClient(transport=transport, base_url="http://test") + + +@pytest.mark.asyncio +async def test_health(client): + r = await client.get("/api/health") + assert r.status_code == 200 + assert r.json() == {"status": "ok"} + + +@pytest.mark.asyncio +async def test_list_scans(client): + r = await client.get("/api/scans") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 1 + assert len(data["scans"]) >= 1 + assert "page" in data + assert "pages" in data + + +@pytest.mark.asyncio +async def test_list_scans_pagination(client): + r = await client.get("/api/scans?page=1&page_size=1") + assert r.status_code == 200 + data = r.json() + assert data["page"] == 1 + assert data["page_size"] == 1 + + +@pytest.mark.asyncio +async def test_get_scan(client): + r = await client.get("/api/scans/http-test") + assert r.status_code == 200 + data = r.json() + assert data["scan"]["id"] == "http-test" + assert len(data["findings"]) >= 1 + + +@pytest.mark.asyncio +async def test_get_scan_not_found(client): + r = await client.get("/api/scans/nonexistent") + assert r.status_code == 404 + + +@pytest.mark.asyncio +async def test_list_findings(client): + r = await client.get("/api/findings") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 2 + assert len(data["findings"]) >= 2 + + +@pytest.mark.asyncio +async def test_findings_filter_severity(client): + r = await client.get("/api/findings?severity=critical") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 1 + for f in data["findings"]: + assert f["severity"] == "critical" + + +@pytest.mark.asyncio +async def test_findings_filter_search(client): + r = await client.get("/api/findings?search=Critical") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 1 + + +@pytest.mark.asyncio +async def test_findings_filter_scanner(client): + r = await client.get("/api/findings?scanner=test") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 2 + + +@pytest.mark.asyncio +async def test_get_finding(client): + r = await client.get("/api/findings") + fid = r.json()["findings"][0]["id"] + r = await client.get(f"/api/findings/{fid}") + assert r.status_code == 200 + assert r.json()["finding"]["id"] == fid + assert r.json()["scan"] is not None + + +@pytest.mark.asyncio +async def test_get_finding_not_found(client): + r = await client.get("/api/findings/nonexistent") + assert r.status_code == 404 + + +@pytest.mark.asyncio +async def test_stats_summary(client): + r = await client.get("/api/stats/summary") + assert r.status_code == 200 + data = r.json() + assert "total_scans" in data + assert "critical" in data + assert data["total_scans"] >= 1 + + +@pytest.mark.asyncio +async def test_stats_top_tools(client): + r = await client.get("/api/stats/top-tools") + assert r.status_code == 200 + data = r.json() + assert "tools" in data + + +@pytest.mark.asyncio +async def test_stats_scanners(client): + r = await client.get("/api/stats/scanners") + assert r.status_code == 200 + data = r.json() + assert "test" in data["scanners"] + + +@pytest.mark.asyncio +async def test_stats_duplicates(client): + r = await client.get("/api/stats/duplicates") + assert r.status_code == 200 + assert "duplicates" in r.json() + + +@pytest.mark.asyncio +async def test_report_json(client): + r = await client.get("/api/report/json") + assert r.status_code == 200 + data = r.json() + assert "summary" in data + assert "generated_at" in data + + +@pytest.mark.asyncio +async def test_report_csv(client): + r = await client.get("/api/report/csv") + assert r.status_code == 200 + assert r.headers["content-type"] == "text/csv; charset=utf-8" + body = r.text + assert "severity" in body + assert "critical" in body + + +@pytest.mark.asyncio +async def test_dashboard_html(client): + r = await client.get("/") + assert r.status_code == 200 + assert "MCP-Scope" in r.text + + +@pytest.mark.asyncio +async def test_finding_detail_html(client): + r = await client.get("/api/findings") + fid = r.json()["findings"][0]["id"] + r = await client.get(f"/findings/{fid}") + assert r.status_code == 200 + assert "Finding Detail" in r.text or "MCP-Scope" in r.text + + +@pytest.mark.asyncio +async def test_api_key_protection(client): + from mcpscope.api.server import create_app + from mcpscope.config import Settings + import tempfile + import os + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + s = Store(db_path=db_path) + cfg = Settings(api_key="test-key-123") + app = create_app(s, settings=cfg) + from httpx import AsyncClient, ASGITransport + + transport = ASGITransport(app=app) + c = AsyncClient(transport=transport, base_url="http://test") + + r = await c.get("/api/scans") + assert r.status_code == 401 + + r = await c.get("/api/scans", headers={"X-API-Key": "test-key-123"}) + assert r.status_code == 200 + + r = await c.get("/") + assert r.status_code == 200 + + await c.aclose() + s._conn.close() + os.unlink(db_path) + + +@pytest.mark.asyncio +async def test_diff_endpoint(client): + r = await client.get("/api/scans") + scans = r.json()["scans"] + if len(scans) >= 2: + a, b = scans[0]["id"], scans[1]["id"] + r = await client.get(f"/api/scans/{a}/diff/{b}") + assert r.status_code == 200 + data = r.json() + assert "new_count" in data + assert "fixed_count" in data + + +@pytest.mark.asyncio +async def test_cors_headers(client): + r = await client.options("/api/health") + assert "access-control-allow-origin" in r.headers or r.status_code in ( + 200, + 204, + 405, + ) + + +@pytest.mark.asyncio +async def test_swagger_docs(client): + r = await client.get("/docs") + assert r.status_code in (200, 404) + + +@pytest.mark.asyncio +async def test_openapi_json(client): + r = await client.get("/api/openapi.json") + assert r.status_code in (200, 404) + if r.status_code == 200: + assert r.json()["info"]["title"] == "MCP-Scope API" + + +@pytest.mark.asyncio +async def test_post_event(client): + r = await client.post( + "/api/events", + json={ + "event_type": "prompt_injection", + "severity": "high", + "message": "Test injection blocked", + "source": "mcpguard-test", + "blocked": True, + }, + ) + assert r.status_code == 200 + data = r.json() + assert data["status"] == "ok" + assert data["id"] is not None + + +@pytest.mark.asyncio +async def test_list_events(client): + await client.post( + "/api/events", + json={ + "event_type": "jailbreak_pattern", + "severity": "critical", + "message": "GODMODE detected", + "tool": "test_tool", + }, + ) + r = await client.get("/api/events") + assert r.status_code == 200 + data = r.json() + assert data["total"] >= 1 + assert len(data["events"]) >= 1 + assert data["events"][0]["event_type"] == "jailbreak_pattern" + + +@pytest.mark.asyncio +async def test_event_stats(client): + r = await client.get("/api/events/stats") + assert r.status_code == 200 + data = r.json() + assert "total" in data + assert "blocked" in data + + +@pytest.mark.asyncio +async def test_clear_events(client): + r = await client.delete("/api/events") + assert r.status_code == 200 + assert r.json()["status"] == "ok" diff --git a/tests/test_parsers.py b/tests/test_parsers.py new file mode 100644 index 0000000000000000000000000000000000000000..62c7d4688ee82c7bc1c204e2fae292a0cadeddd5 --- /dev/null +++ b/tests/test_parsers.py @@ -0,0 +1,540 @@ +import pytest +import json +from pathlib import Path +from mcpscope.ingest.cisco_mcp import CiscoMCPParser +from mcpscope.ingest.cisco_a2a import CiscoA2AParser +from mcpscope.ingest.mcpscan import MCPScanParser +from mcpscope.ingest.mcpwn import MCPwnParser +from mcpscope.ingest.sarif import SarifParser +from mcpscope.ingest.base import ParseError +from mcpscope.models.finding import Severity + + +def load_fixture(name: str) -> dict: + path = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / name + with open(path) as f: + return json.load(f) + + +class TestCiscoMCPParser: + def setup_method(self): + self.parser = CiscoMCPParser() + + def test_parse_valid(self): + data = { + "server_url": "https://example.com/mcp", + "scan_results": [ + { + "tool_name": "add", + "status": "completed", + "findings": { + "api_analyzer": { + "severity": "HIGH", + "total_findings": 1, + "threats": { + "items": [ + { + "technique_id": "AITech-8.2", + "technique_name": "Data Exfiltration", + "items": [ + { + "sub_technique_id": "AISubtech-8.2.3", + "sub_technique_name": "Data Exfiltration via Agent Tooling", + "max_severity": "HIGH", + "description": "Sensitive data may be exposed", + } + ], + } + ] + }, + "threat_names": ["Data Exfiltration"], + "threat_summary": "Tool may expose data", + } + }, + "is_safe": False, + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].tool_name == "add" + assert findings[0].severity == Severity.HIGH + assert "AISubtech-8.2.3" in findings[0].title + + def test_parse_safe_tool(self): + data = { + "server_url": "https://example.com/mcp", + "scan_results": [ + { + "tool_name": "safe_tool", + "status": "completed", + "findings": { + "api_analyzer": {"severity": "SAFE", "total_findings": 0}, + "yara_analyzer": {"severity": "SAFE", "total_findings": 0}, + }, + "is_safe": True, + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 0 + + def test_validate_rejects_bad_data(self): + with pytest.raises(ParseError, match="Missing 'scan_results'"): + self.parser.validate({"foo": "bar"}) + + +class TestCiscoA2AParser: + def setup_method(self): + self.parser = CiscoA2AParser() + + def test_parse_valid(self): + data = { + "target": "https://agent.example.com", + "findings": [ + { + "threat_name": "PROMPT INJECTION", + "severity": "HIGH", + "analyzer": "YARA", + "aitech": "AITech-1.1", + "aitech_name": "Direct Prompt Injection", + "description": "Prompt injection detected", + "summary": "Found injection patterns", + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].severity == Severity.HIGH + assert "AITech-1.1" in findings[0].title + + def test_validate_rejects_bad_data(self): + with pytest.raises(ParseError, match="Missing 'findings'"): + self.parser.validate({"nope": True}) + + +class TestMCPScanParser: + def setup_method(self): + self.parser = MCPScanParser() + + def test_parse_valid(self): + data = { + "target": "mcp.json", + "servers": [{"id": "s1", "name": "Demo Server"}], + "issues": [ + { + "code": "E001", + "severity": "high", + "message": "Prompt injection risk", + "tool_name": "add", + "description": "Tool description contains injection vectors", + "recommendation": "Sanitize description", + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].severity == Severity.HIGH + assert "E001" in findings[0].title + + def test_validate_rejects_bad_data(self): + with pytest.raises(ParseError, match="Missing 'issues'"): + self.parser.validate({"x": []}) + + +class TestMCPwnParser: + def setup_method(self): + self.parser = MCPwnParser() + + def test_parse_new_format(self): + data = { + "target": "server.py", + "findings": [ + { + "id": "MCP-001", + "severity": "CRITICAL", + "title": "Tool Poisoning", + "tool": "fetch_url", + "description": "Hidden instructions found", + "recommendation": "Audit tool descriptions", + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].severity == Severity.CRITICAL + assert findings[0].tool_name == "fetch_url" + + def test_parse_legacy_format(self): + data = { + "tool": "Mcpwn", + "findings": [ + { + "test": "tool_injection", + "type": "RCE", + "severity": "CRITICAL", + "tool": "exec_cmd", + "detection": "uid=1000", + "description": "RCE via command injection", + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].severity == Severity.CRITICAL + + def test_validate_rejects_bad_data(self): + with pytest.raises(ParseError, match="Missing 'findings'"): + self.parser.validate({"version": "1.0"}) + + +class TestSarifParser: + def setup_method(self): + self.parser = SarifParser() + + def test_parse_valid(self): + data = { + "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/main/sarif-2.1.json", + "version": "2.1.0", + "runs": [ + { + "tool": { + "driver": { + "name": "TestScanner", + "version": "1.0", + "rules": [ + { + "id": "RULE001", + "name": "Test Rule", + "shortDescription": {"text": "A test rule"}, + "fullDescription": { + "text": "Full description of test rule" + }, + } + ], + } + }, + "results": [ + { + "ruleId": "RULE001", + "level": "error", + "message": {"text": "Vulnerability found"}, + } + ], + } + ], + } + findings = self.parser.parse(data) + assert len(findings) == 1 + assert findings[0].severity == Severity.CRITICAL + assert findings[0].tool_name == "TestScanner" + + def test_validate_rejects_bad_data(self): + with pytest.raises(ParseError, match="Missing"): + self.parser.validate({"foo": "bar"}) + + +class TestStore: + def test_save_and_retrieve(self): + from mcpscope.storage.store import Store + import tempfile + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + store = Store(db_path=db_path) + from mcpscope.models.finding import Finding + from mcpscope.models.scan import ScanRun + + scan = ScanRun(id="test-scan", scanner="test", target="test-target") + findings = [ + Finding( + scan_id="test-scan", + scanner="test", + tool_name="tool1", + severity=Severity.CRITICAL, + title="Critical finding", + ), + Finding( + scan_id="test-scan", + scanner="test", + tool_name="tool2", + severity=Severity.LOW, + title="Low finding", + ), + ] + saved = store.save_scan(scan, findings) + assert saved.id == "test-scan" + assert saved.critical_count == 1 + assert saved.low_count == 1 + retrieved, total = store.get_findings(scan_id="test-scan") + assert total == 2 + titles = {f.title for f in retrieved} + assert "Critical finding" in titles + assert "Low finding" in titles + import os + + os.unlink(db_path) + + +class TestCLI: + def test_import_unknown_scanner(self): + from mcpscope.cli import PARSERS + + assert "cisco-mcp" in PARSERS + assert "cisco-a2a" in PARSERS + assert "mcp-scan" in PARSERS + assert "mcpscan" in PARSERS + assert "mcpwn" in PARSERS + assert "sarif" in PARSERS + + +class TestConfig: + def test_defaults(self): + from mcpscope.config import Settings + + s = Settings() + assert s.port == 8080 + assert s.host == "127.0.0.1" + assert s.auto_refresh_seconds == 30 + assert s.api_key is None + assert s.webhook_urls == [] + assert s.max_upload_mb == 50 + + def test_save_load_roundtrip(self): + from mcpscope.config import Settings + import tempfile + import os + import json + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + path = f.name + json.dump({"port": 9090, "host": "0.0.0.0", "auto_refresh_seconds": 60}, f) + s = Settings.load(path) + assert s.port == 9090 + assert s.host == "0.0.0.0" + assert s.auto_refresh_seconds == 60 + os.unlink(path) + + +class TestDedup: + def test_duplicates(self): + import tempfile + import os + from mcpscope.storage.store import Store + from mcpscope.models.finding import Finding + from mcpscope.models.scan import ScanRun + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + store = Store(db_path=db_path) + scan1 = ScanRun(id="s1", scanner="test", target="t1") + store.save_scan( + scan1, + [ + Finding( + scan_id="s1", + scanner="test", + tool_name="foo", + severity=Severity.CRITICAL, + title="Same issue", + ), + Finding( + scan_id="s1", + scanner="test", + tool_name="bar", + severity=Severity.LOW, + title="Unique", + ), + ], + ) + scan2 = ScanRun(id="s2", scanner="test", target="t2") + store.save_scan( + scan2, + [ + Finding( + scan_id="s2", + scanner="test", + tool_name="foo", + severity=Severity.CRITICAL, + title="Same issue", + ), + Finding( + scan_id="s2", + scanner="test", + tool_name="baz", + severity=Severity.HIGH, + title="Another", + ), + ], + ) + dups = store.get_duplicates() + assert len(dups) >= 1 + dup = [d for d in dups if d["tool_name"] == "foo"][0] + assert dup["count"] == 2 + os.unlink(db_path) + + +class TestScanDiff: + def test_diff(self): + import tempfile + import os + from mcpscope.storage.store import Store + from mcpscope.models.finding import Finding + from mcpscope.models.scan import ScanRun + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + store = Store(db_path=db_path) + store.save_scan( + ScanRun(id="a", scanner="t", target="t1"), + [ + Finding( + scan_id="a", + scanner="t", + tool_name="x", + severity=Severity.CRITICAL, + title="Only in A", + ), + Finding( + scan_id="a", + scanner="t", + tool_name="y", + severity=Severity.HIGH, + title="Both", + ), + ], + ) + store.save_scan( + ScanRun(id="b", scanner="t", target="t2"), + [ + Finding( + scan_id="b", + scanner="t", + tool_name="y", + severity=Severity.HIGH, + title="Both", + ), + Finding( + scan_id="b", + scanner="t", + tool_name="z", + severity=Severity.MEDIUM, + title="Only in B", + ), + ], + ) + diff = store.diff_scans("a", "b") + assert diff["new_count"] == 1 + assert diff["fixed_count"] == 1 + assert diff["unchanged_count"] == 1 + os.unlink(db_path) + + +class TestWebhookPayload: + def test_notify_skips_safe(self): + from mcpscope.webhooks import notify_scan_imported + from mcpscope.models.finding import Finding, Severity + + findings = [ + Finding( + scan_id="x", + scanner="t", + tool_name="t", + severity=Severity.INFO, + title="safe", + ) + ] + notify_scan_imported( + [], findings=findings, findings_count=1, critical_count=0, high_count=0 + ) + + def test_notify_builds_alerts(self): + from mcpscope.webhooks import notify_scan_imported + from mcpscope.models.finding import Finding, Severity + + findings = [ + Finding( + scan_id="x", + scanner="t", + tool_name="t1", + severity=Severity.CRITICAL, + title="bad", + ), + Finding( + scan_id="x", + scanner="t", + tool_name="t2", + severity=Severity.HIGH, + title="warn", + ), + ] + notify_scan_imported( + ["http://localhost:9999/noexist"], + findings=findings, + findings_count=2, + critical_count=1, + high_count=1, + ) + + +class TestFileSizeValidation: + def test_large_file_rejected(self): + import mcpscope.ingest.base as base_mod + from mcpscope.ingest.mcpscan import MCPScanParser + from mcpscope.ingest.base import ParseError + import tempfile + import os + + orig_max = base_mod.MAX_UPLOAD_MB + base_mod.MAX_UPLOAD_MB = 1 + try: + parser = MCPScanParser() + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + f.write('{"issues": []}') + path = f.name + os.truncate(path, 2 * 1024 * 1024 + 1) + with pytest.raises(ParseError, match="too large"): + parser.parse_file(path) + finally: + base_mod.MAX_UPLOAD_MB = orig_max + os.unlink(path) + + +class TestCSVReport: + def test_csv_format(self): + from mcpscope.storage.store import Store + from mcpscope.models.finding import Finding, Severity + from mcpscope.models.scan import ScanRun + import tempfile + import os + import csv + import io + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + store = Store(db_path=db_path) + store.save_scan( + ScanRun(id="csv-test", scanner="t", target="t"), + [ + Finding( + scan_id="csv-test", + scanner="t", + tool_name="x", + severity=Severity.CRITICAL, + title="CSV finding", + ), + ], + ) + findings, _ = store.get_findings(page_size=1000) + output = io.StringIO() + w = csv.writer(output) + w.writerow(["id", "severity", "title"]) + for f in findings: + sev = f.severity.value if hasattr(f.severity, "value") else f.severity + w.writerow([f.id, sev, f.title]) + result = output.getvalue() + assert "CSV finding" in result + assert "critical" in result + os.unlink(db_path)