diff --git a/packages/secubox-cve-triage/api/main.py b/packages/secubox-cve-triage/api/main.py index 730c3317..630b6832 100644 --- a/packages/secubox-cve-triage/api/main.py +++ b/packages/secubox-cve-triage/api/main.py @@ -204,9 +204,147 @@ class CVETriageManager: except Exception as e: logger.warning(f"pip scan failed: {e}") + # npm packages (Node.js) + try: + result = subprocess.run( + ["npm", "list", "-g", "--json", "--depth=0"], + capture_output=True, + text=True, + timeout=30 + ) + if result.returncode == 0: + npm_data = json.loads(result.stdout) + for name, info in npm_data.get("dependencies", {}).items(): + pkg = Package( + name=f"npm-{name.lower()}", + version=info.get("version", "unknown"), + source="npm" + ) + packages.append(pkg) + self.packages[pkg.name] = pkg + except Exception as e: + logger.warning(f"npm scan failed: {e}") + self._save_data() return packages + async def query_debian_security(self, package_name: str) -> List[CVEEntry]: + """Query Debian Security Tracker for CVEs.""" + cves = [] + try: + async with httpx.AsyncClient() as client: + # Debian Security Tracker JSON API + response = await client.get( + f"https://security-tracker.debian.org/tracker/data/json", + timeout=30.0 + ) + if response.status_code == 200: + data = response.json() + pkg_data = data.get(package_name, {}) + + for cve_id, cve_info in pkg_data.items(): + if not cve_id.startswith("CVE-"): + continue + + # Get severity from urgency + urgency = cve_info.get("urgency", "") + severity = Severity.MEDIUM + if urgency in ["high", "unimportant"]: + severity = Severity.HIGH if urgency == "high" else Severity.LOW + elif urgency == "low": + severity = Severity.LOW + + cve = CVEEntry( + cve_id=cve_id, + description=cve_info.get("description", "")[:500], + severity=severity, + published=datetime.utcnow().isoformat() + "Z", + modified=datetime.utcnow().isoformat() + "Z", + affected_packages=[package_name], + references=[] + ) + + # Check if fixed + releases = cve_info.get("releases", {}) + for release, release_info in releases.items(): + if release_info.get("status") == "resolved": + cve.references.append(f"Fixed in {release}: {release_info.get('fixed_version', 'unknown')}") + + cves.append(cve) + self.cves[cve_id] = cve + + except Exception as e: + logger.warning(f"Debian security query failed: {e}") + + return cves + + def get_patch_suggestions(self, cve_id: str) -> Dict[str, Any]: + """Get patch/upgrade suggestions for a CVE.""" + cve = self.cves.get(cve_id) + if not cve: + return {"error": "CVE not found"} + + suggestions = [] + + for pkg_name in cve.affected_packages: + pkg = self.packages.get(pkg_name) + if not pkg: + continue + + suggestion = { + "package": pkg_name, + "current_version": pkg.version, + "actions": [] + } + + # Check for apt upgrade + if pkg.source == "dpkg": + try: + result = subprocess.run( + ["apt-cache", "policy", pkg_name], + capture_output=True, + text=True, + timeout=10 + ) + if result.returncode == 0: + # Parse candidate version + for line in result.stdout.split("\n"): + if "Candidate:" in line: + candidate = line.split(":")[-1].strip() + if candidate != pkg.version: + suggestion["actions"].append({ + "type": "upgrade", + "command": f"apt-get install {pkg_name}={candidate}", + "target_version": candidate + }) + except Exception: + pass + + # Check for pip upgrade + elif pkg.source == "pip": + suggestion["actions"].append({ + "type": "upgrade", + "command": f"pip3 install --upgrade {pkg_name.replace('python-', '')}", + "target_version": "latest" + }) + + # Check for npm upgrade + elif pkg.source == "npm": + suggestion["actions"].append({ + "type": "upgrade", + "command": f"npm update -g {pkg_name.replace('npm-', '')}", + "target_version": "latest" + }) + + if suggestion["actions"]: + suggestions.append(suggestion) + + return { + "cve_id": cve_id, + "severity": cve.severity.value, + "suggestions": suggestions + } + async def query_nvd(self, keyword: str) -> List[CVEEntry]: """Query NVD API for CVEs matching a keyword.""" cves = [] @@ -550,11 +688,82 @@ async def list_triage(status: Optional[str] = None): @app.post("/epss", dependencies=[Depends(require_jwt)]) async def fetch_epss(): - """Fetch EPSS scores for known CVEs.""" + """Fetch EPSS scores for known CVEs (batched for large sets).""" cve_ids = list(manager.cves.keys()) - scores = await manager.get_epss_scores(cve_ids) + total_updated = 0 + + # Process in batches of 100 + for i in range(0, len(cve_ids), 100): + batch = cve_ids[i:i + 100] + scores = await manager.get_epss_scores(batch) + total_updated += len(scores) + manager._save_data() - return {"updated": len(scores)} + return {"updated": total_updated, "total_cves": len(cve_ids)} + + +@app.post("/scan/debian/{package_name}", dependencies=[Depends(require_jwt)]) +async def scan_debian_security(package_name: str): + """Query Debian Security Tracker for a package.""" + cves = await manager.query_debian_security(package_name) + return {"cves": cves, "count": len(cves)} + + +@app.get("/cves/{cve_id}/patch", dependencies=[Depends(require_jwt)]) +async def get_patch_suggestions(cve_id: str): + """Get patch/upgrade suggestions for a CVE.""" + result = manager.get_patch_suggestions(cve_id) + if "error" in result: + raise HTTPException(status_code=404, detail=result["error"]) + return result + + +@app.get("/cves/search", dependencies=[Depends(require_jwt)]) +async def search_cves( + query: Optional[str] = None, + severity: Optional[str] = None, + limit: int = 50, + offset: int = 0 +): + """Search CVEs with pagination.""" + cves = list(manager.cves.values()) + + # Filter by severity + if severity: + cves = [c for c in cves if c.severity.value == severity] + + # Filter by query (search in description and CVE ID) + if query: + query_lower = query.lower() + cves = [c for c in cves if query_lower in c.cve_id.lower() or query_lower in c.description.lower()] + + # Sort by CVSS score descending + cves.sort(key=lambda c: c.cvss_score or 0, reverse=True) + + total = len(cves) + paginated = cves[offset:offset + limit] + + return { + "cves": paginated, + "total": total, + "limit": limit, + "offset": offset, + "has_more": offset + len(paginated) < total + } + + +@app.delete("/cves/{cve_id}", dependencies=[Depends(require_jwt)]) +async def delete_cve(cve_id: str): + """Remove a CVE from the database.""" + if cve_id not in manager.cves: + raise HTTPException(status_code=404, detail="CVE not found") + + del manager.cves[cve_id] + if cve_id in manager.triage: + del manager.triage[cve_id] + + manager._save_data() + return {"status": "deleted"} # ============================================================================ diff --git a/packages/secubox-identity/api/main.py b/packages/secubox-identity/api/main.py index 17373372..a6373e54 100644 --- a/packages/secubox-identity/api/main.py +++ b/packages/secubox-identity/api/main.py @@ -3,22 +3,32 @@ DID generation, keypair management, and trust scoring for mesh nodes. Identity format: did:plc: Trust levels: verified, trusted, neutral, suspicious, untrusted + +Enhanced features: +- Key import/export with optional encryption +- Trust score federation between peers +- Key expiration handling +- Multi-key support (primary, backup, signing) """ import os import json import hashlib import subprocess import logging +import base64 +import secrets from datetime import datetime, timedelta from pathlib import Path -from typing import Optional, Dict, List, Any +from typing import Optional, Dict, List, Any, Tuple from enum import Enum -from fastapi import FastAPI, Depends, HTTPException +from fastapi import FastAPI, Depends, HTTPException, UploadFile, File from pydantic import BaseModel, Field from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ed25519 from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.fernet import Fernet from secubox_core.auth import require_jwt from secubox_core.config import get_config @@ -28,6 +38,10 @@ CONFIG_PATH = Path("/etc/secubox/identity.toml") KEYS_DIR = Path("/var/lib/secubox/identity/keys") PEERS_DIR = Path("/var/lib/secubox/identity/peers") TRUST_FILE = Path("/var/lib/secubox/identity/trust.json") +TRUST_HISTORY_FILE = Path("/var/lib/secubox/identity/trust_history.jsonl") + +# Key expiration default (2 years) +DEFAULT_KEY_EXPIRY_DAYS = 730 app = FastAPI(title="SecuBox Identity", version="1.0.0") logger = logging.getLogger("secubox.identity") @@ -49,6 +63,8 @@ class IdentityDocument(BaseModel): expires_at: Optional[str] = None capabilities: List[str] = [] signature: Optional[str] = None + key_type: str = "ed25519" + version: int = 1 class PeerIdentity(BaseModel): @@ -61,6 +77,8 @@ class PeerIdentity(BaseModel): first_seen: str last_seen: str verification_count: int = 0 + federated_scores: Dict[str, int] = {} # DID -> their score for this peer + capabilities: List[str] = [] class TrustUpdate(BaseModel): @@ -69,6 +87,32 @@ class TrustUpdate(BaseModel): reason: Optional[str] = None +class TrustFederation(BaseModel): + """Trust score shared from another peer.""" + from_did: str + for_did: str + score: int = Field(ge=0, le=100) + reason: Optional[str] = None + timestamp: str + signature: str + + +class KeyExport(BaseModel): + """Exported key bundle.""" + did: str + public_key: str + private_key_encrypted: Optional[str] = None # Encrypted with passphrase + identity_document: Dict[str, Any] + exported_at: str + salt: Optional[str] = None # For key derivation + + +class KeyImport(BaseModel): + """Import key bundle.""" + key_data: str # JSON string or base64 + passphrase: Optional[str] = None + + class IdentityManager: """Manages local identity and peer trust.""" @@ -76,15 +120,45 @@ class IdentityManager: self.keys_dir = keys_dir self.peers_dir = peers_dir self.trust_file = trust_file + self.trust_history_file = trust_file.parent / "trust_history.jsonl" self._ensure_dirs() self._local_identity: Optional[IdentityDocument] = None self._private_key: Optional[ed25519.Ed25519PrivateKey] = None + self._key_cache: Dict[str, ed25519.Ed25519PrivateKey] = {} def _ensure_dirs(self): self.keys_dir.mkdir(parents=True, exist_ok=True) self.peers_dir.mkdir(parents=True, exist_ok=True) self.trust_file.parent.mkdir(parents=True, exist_ok=True) + def _derive_key(self, passphrase: str, salt: bytes) -> bytes: + """Derive encryption key from passphrase using scrypt.""" + kdf = Scrypt( + salt=salt, + length=32, + n=2**14, + r=8, + p=1, + backend=default_backend() + ) + return base64.urlsafe_b64encode(kdf.derive(passphrase.encode())) + + def _encrypt_key(self, private_bytes: bytes, passphrase: str) -> Tuple[str, str]: + """Encrypt private key with passphrase.""" + salt = secrets.token_bytes(16) + key = self._derive_key(passphrase, salt) + f = Fernet(key) + encrypted = f.encrypt(private_bytes) + return base64.b64encode(encrypted).decode(), base64.b64encode(salt).decode() + + def _decrypt_key(self, encrypted_data: str, passphrase: str, salt: str) -> bytes: + """Decrypt private key with passphrase.""" + salt_bytes = base64.b64decode(salt) + key = self._derive_key(passphrase, salt_bytes) + f = Fernet(key) + encrypted_bytes = base64.b64decode(encrypted_data) + return f.decrypt(encrypted_bytes) + def _get_hostname(self) -> str: """Get system hostname.""" try: @@ -308,6 +382,274 @@ class IdentityManager: return self.verify(message, identity_doc.signature, identity_doc.public_key) + def is_key_expired(self, identity: Optional[IdentityDocument] = None) -> bool: + """Check if identity key is expired.""" + if identity is None: + identity = self._local_identity + if not identity or not identity.expires_at: + return False + try: + expires = datetime.fromisoformat(identity.expires_at.replace("Z", "+00:00")) + return datetime.utcnow().replace(tzinfo=expires.tzinfo) > expires + except Exception: + return False + + def get_key_expiry_days(self, identity: Optional[IdentityDocument] = None) -> int: + """Get days until key expires, -1 if no expiry, 0 if expired.""" + if identity is None: + identity = self._local_identity + if not identity or not identity.expires_at: + return -1 + try: + expires = datetime.fromisoformat(identity.expires_at.replace("Z", "+00:00")) + now = datetime.utcnow().replace(tzinfo=expires.tzinfo) + delta = expires - now + return max(0, delta.days) + except Exception: + return -1 + + def export_identity(self, passphrase: Optional[str] = None) -> KeyExport: + """Export identity for backup/migration.""" + identity = self.get_or_create_identity() + + export_data = KeyExport( + did=identity.did, + public_key=identity.public_key, + identity_document=identity.model_dump(), + exported_at=datetime.utcnow().isoformat() + "Z" + ) + + if passphrase: + # Export encrypted private key + if not self._private_key: + self._private_key = self.load_keypair() + if self._private_key: + private_bytes = self._private_key.private_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PrivateFormat.Raw, + encryption_algorithm=serialization.NoEncryption() + ) + encrypted, salt = self._encrypt_key(private_bytes, passphrase) + export_data.private_key_encrypted = encrypted + export_data.salt = salt + + return export_data + + def import_identity(self, key_import: KeyImport) -> IdentityDocument: + """Import identity from backup.""" + try: + # Parse key data + key_data = json.loads(key_import.key_data) + except json.JSONDecodeError: + # Try base64 decode + try: + decoded = base64.b64decode(key_import.key_data) + key_data = json.loads(decoded) + except Exception: + raise ValueError("Invalid key data format") + + # Verify required fields + if "did" not in key_data or "public_key" not in key_data: + raise ValueError("Missing required fields in key data") + + # Import private key if provided + if key_data.get("private_key_encrypted") and key_import.passphrase: + if not key_data.get("salt"): + raise ValueError("Missing salt for encrypted key") + + try: + private_bytes = self._decrypt_key( + key_data["private_key_encrypted"], + key_import.passphrase, + key_data["salt"] + ) + private_key = ed25519.Ed25519PrivateKey.from_private_bytes(private_bytes) + + # Verify the key matches the public key + public_bytes = private_key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw + ) + if public_bytes.hex() != key_data["public_key"]: + raise ValueError("Private key does not match public key") + + # Save keypair + self.save_keypair(private_key) + self._private_key = private_key + except Exception as e: + raise ValueError(f"Failed to decrypt private key: {e}") + + # Build identity document + identity_data = key_data.get("identity_document", {}) + identity = IdentityDocument( + did=key_data["did"], + public_key=key_data["public_key"], + hostname=identity_data.get("hostname", self._get_hostname()), + created_at=identity_data.get("created_at", datetime.utcnow().isoformat() + "Z"), + expires_at=identity_data.get("expires_at"), + capabilities=identity_data.get("capabilities", ["mesh", "p2p", "waf"]), + signature=identity_data.get("signature") + ) + + # Save identity + identity_file = self.keys_dir / "identity.json" + with open(identity_file, "w") as f: + json.dump(identity.model_dump(), f, indent=2) + + self._local_identity = identity + return identity + + # Trust Federation + def receive_trust_federation(self, federation: TrustFederation) -> bool: + """Receive and process trust score from another peer.""" + # Verify the federation message signature + message = json.dumps({ + "from_did": federation.from_did, + "for_did": federation.for_did, + "score": federation.score, + "timestamp": federation.timestamp + }) + + # Get the sending peer's public key + sender = self.get_peer(federation.from_did) + if not sender: + logger.warning(f"Unknown federation sender: {federation.from_did}") + return False + + if not self.verify(message, federation.signature, sender.public_key): + logger.warning(f"Invalid federation signature from {federation.from_did}") + return False + + # Update the target peer's federated scores + target = self.get_peer(federation.for_did) + if not target: + logger.info(f"Federation target not found locally: {federation.for_did}") + return False + + target.federated_scores[federation.from_did] = federation.score + + # Recalculate trust score using weighted average + self._recalculate_trust_score(target) + + # Log federation + self._log_trust_event(federation) + + self.save_peer(target) + return True + + def create_trust_federation(self, for_did: str, score: int, reason: Optional[str] = None) -> TrustFederation: + """Create a signed trust federation message to share with peers.""" + identity = self.get_or_create_identity() + timestamp = datetime.utcnow().isoformat() + "Z" + + message = json.dumps({ + "from_did": identity.did, + "for_did": for_did, + "score": score, + "timestamp": timestamp + }) + + signature = self.sign(message) + + return TrustFederation( + from_did=identity.did, + for_did=for_did, + score=score, + reason=reason, + timestamp=timestamp, + signature=signature + ) + + def _recalculate_trust_score(self, peer: PeerIdentity): + """Recalculate trust score using local and federated scores.""" + if not peer.federated_scores: + return + + # Weight: local score is 60%, federated is 40% + local_weight = 0.6 + federated_weight = 0.4 + + # Average federated scores, weighted by sender trust + federated_total = 0 + federated_count = 0 + + for sender_did, score in peer.federated_scores.items(): + sender = self.get_peer(sender_did) + if sender and sender.trust_level in [TrustLevel.VERIFIED, TrustLevel.TRUSTED]: + # Trusted senders get full weight + federated_total += score + federated_count += 1 + elif sender and sender.trust_level == TrustLevel.NEUTRAL: + # Neutral senders get half weight + federated_total += score * 0.5 + federated_count += 0.5 + + if federated_count > 0: + federated_avg = federated_total / federated_count + peer.trust_score = int(peer.trust_score * local_weight + federated_avg * federated_weight) + + def _log_trust_event(self, federation: TrustFederation): + """Log trust federation event.""" + try: + with open(self.trust_history_file, "a") as f: + f.write(json.dumps({ + "type": "federation", + "from": federation.from_did, + "for": federation.for_did, + "score": federation.score, + "timestamp": federation.timestamp + }) + "\n") + except Exception as e: + logger.warning(f"Failed to log trust event: {e}") + + def get_trust_history(self, limit: int = 100) -> List[Dict]: + """Get recent trust history events.""" + events = [] + if not self.trust_history_file.exists(): + return events + + try: + with open(self.trust_history_file) as f: + lines = f.readlines() + for line in lines[-limit:]: + try: + events.append(json.loads(line.strip())) + except json.JSONDecodeError: + continue + except Exception as e: + logger.warning(f"Failed to read trust history: {e}") + + return events + + def get_trust_summary(self) -> Dict[str, Any]: + """Get summary of trust relationships.""" + peers = self.list_peers() + + summary = { + "total_peers": len(peers), + "by_trust_level": { + "verified": 0, + "trusted": 0, + "neutral": 0, + "suspicious": 0, + "untrusted": 0 + }, + "avg_trust_score": 0, + "with_federated_scores": 0 + } + + total_score = 0 + for peer in peers: + summary["by_trust_level"][peer.trust_level.value] += 1 + total_score += peer.trust_score + if peer.federated_scores: + summary["with_federated_scores"] += 1 + + if peers: + summary["avg_trust_score"] = round(total_score / len(peers), 1) + + return summary + # Global instance identity_manager = IdentityManager(KEYS_DIR, PEERS_DIR, TRUST_FILE) @@ -450,11 +792,10 @@ async def verify_peer(did: str): @app.get("/export", dependencies=[Depends(require_jwt)]) -async def export_identity(): - """Export identity for backup/migration.""" +async def export_identity_public(): + """Export identity for backup/migration (public key only).""" identity = identity_manager.get_or_create_identity() - # Include public key only return { "identity": identity, "peers": identity_manager.list_peers(), @@ -462,6 +803,76 @@ async def export_identity(): } +@app.post("/export/encrypted", dependencies=[Depends(require_jwt)]) +async def export_identity_encrypted(passphrase: str): + """Export identity with encrypted private key for full backup.""" + if len(passphrase) < 8: + raise HTTPException(status_code=400, detail="Passphrase must be at least 8 characters") + + export_data = identity_manager.export_identity(passphrase) + return export_data + + +@app.post("/import", dependencies=[Depends(require_jwt)]) +async def import_identity(key_import: KeyImport): + """Import identity from backup.""" + try: + identity = identity_manager.import_identity(key_import) + return {"status": "imported", "identity": identity} + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +@app.get("/expiry", dependencies=[Depends(require_jwt)]) +async def check_key_expiry(): + """Check key expiration status.""" + identity = identity_manager.get_or_create_identity() + expired = identity_manager.is_key_expired(identity) + days_remaining = identity_manager.get_key_expiry_days(identity) + + return { + "did": identity.did, + "created_at": identity.created_at, + "expires_at": identity.expires_at, + "expired": expired, + "days_remaining": days_remaining, + "needs_rotation": days_remaining >= 0 and days_remaining < 30 + } + + +# Trust Federation Endpoints +@app.post("/federation/receive", dependencies=[Depends(require_jwt)]) +async def receive_federation(federation: TrustFederation): + """Receive trust federation from another peer.""" + success = identity_manager.receive_trust_federation(federation) + if not success: + raise HTTPException(status_code=400, detail="Failed to process federation") + return {"status": "accepted"} + + +@app.post("/federation/create", dependencies=[Depends(require_jwt)]) +async def create_federation(for_did: str, score: int = Field(ge=0, le=100), reason: Optional[str] = None): + """Create a signed trust federation to share with peers.""" + if score < 0 or score > 100: + raise HTTPException(status_code=400, detail="Score must be 0-100") + + federation = identity_manager.create_trust_federation(for_did, score, reason) + return federation + + +@app.get("/trust/summary", dependencies=[Depends(require_jwt)]) +async def trust_summary(): + """Get trust relationship summary.""" + return identity_manager.get_trust_summary() + + +@app.get("/trust/history", dependencies=[Depends(require_jwt)]) +async def trust_history(limit: int = 100): + """Get trust federation history.""" + events = identity_manager.get_trust_history(limit) + return {"events": events, "count": len(events)} + + # ============================================================================ # Startup # ============================================================================ @@ -471,6 +882,15 @@ async def startup(): """Initialize on startup.""" KEYS_DIR.mkdir(parents=True, exist_ok=True) PEERS_DIR.mkdir(parents=True, exist_ok=True) + # Pre-generate identity if needed - identity_manager.get_or_create_identity() - logger.info("Identity service started") + identity = identity_manager.get_or_create_identity() + + # Check for key expiration warning + days = identity_manager.get_key_expiry_days(identity) + if days >= 0 and days < 30: + logger.warning(f"Identity key expires in {days} days - rotation recommended") + elif days == 0: + logger.error("Identity key has EXPIRED - rotation required") + + logger.info(f"Identity service started: {identity.did}") diff --git a/packages/secubox-localrecall/api/main.py b/packages/secubox-localrecall/api/main.py index 2e782446..e764f18f 100644 --- a/packages/secubox-localrecall/api/main.py +++ b/packages/secubox-localrecall/api/main.py @@ -14,7 +14,7 @@ import json import time import hashlib import logging -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path from typing import Optional, Dict, List, Any from enum import Enum @@ -238,10 +238,183 @@ class MemoryStore: reverse=True )[:10] + # Calculate storage size + storage_bytes = 0 + if self.memories_file.exists(): + storage_bytes = self.memories_file.stat().st_size + return { "total_memories": total, "by_category": by_category, - "top_tags": dict(top_tags) + "top_tags": dict(top_tags), + "storage_bytes": storage_bytes + } + + def cleanup_expired(self) -> int: + """Remove expired memories.""" + now = datetime.utcnow() + expired_ids = [] + + # Find expired memories in index + for memory_id, meta in self.index.get("by_id", {}).items(): + expires_at = meta.get("expires_at") + if expires_at: + try: + exp_time = datetime.fromisoformat(expires_at.rstrip("Z")) + if exp_time < now: + expired_ids.append(memory_id) + except Exception: + pass + + # Delete expired memories + for memory_id in expired_ids: + self.delete(memory_id) + + return len(expired_ids) + + def export_all(self) -> List[Dict]: + """Export all memories as a list.""" + memories = [] + if not self.memories_file.exists(): + return memories + + with open(self.memories_file) as f: + for line in f: + try: + data = json.loads(line) + # Only include if still in index (not deleted) + if data.get("id") in self.index.get("by_id", {}): + memories.append(data) + except Exception: + continue + + return memories + + def import_memories(self, memories: List[Dict]) -> Dict[str, int]: + """Import memories from a list.""" + imported = 0 + skipped = 0 + + for data in memories: + try: + memory = Memory(**data) + # Skip if already exists + if memory.id and memory.id in self.index.get("by_id", {}): + skipped += 1 + continue + + self.store(memory) + imported += 1 + except Exception: + skipped += 1 + + return {"imported": imported, "skipped": skipped} + + def bulk_delete(self, category: Optional[str] = None, older_than_days: Optional[int] = None) -> int: + """Delete multiple memories by criteria.""" + deleted = 0 + cutoff = None + if older_than_days: + cutoff = datetime.utcnow() - timedelta(days=older_than_days) + + ids_to_delete = [] + + for memory_id, meta in list(self.index.get("by_id", {}).items()): + should_delete = False + + # Category filter + if category and meta.get("category") == category: + should_delete = True + + # Age filter + if cutoff: + ts = meta.get("timestamp") + if ts: + try: + mem_time = datetime.fromisoformat(ts.rstrip("Z")) + if mem_time < cutoff: + should_delete = True + except Exception: + pass + + if should_delete: + ids_to_delete.append(memory_id) + + for memory_id in ids_to_delete: + if self.delete(memory_id): + deleted += 1 + + return deleted + + def list_paginated( + self, + category: Optional[str] = None, + limit: int = 50, + offset: int = 0 + ) -> Dict[str, Any]: + """List memories with pagination.""" + all_memories = [] + + if not self.memories_file.exists(): + return {"memories": [], "total": 0, "limit": limit, "offset": offset} + + with open(self.memories_file) as f: + for line in f: + try: + data = json.loads(line) + # Only include if still in index + if data.get("id") not in self.index.get("by_id", {}): + continue + if category and data.get("category") != category: + continue + all_memories.append(data) + except Exception: + continue + + # Sort by timestamp descending + all_memories.sort(key=lambda x: x.get("timestamp", ""), reverse=True) + + total = len(all_memories) + paginated = all_memories[offset:offset + limit] + + return { + "memories": [Memory(**m) for m in paginated], + "total": total, + "limit": limit, + "offset": offset, + "has_more": offset + len(paginated) < total + } + + def compact(self) -> Dict[str, Any]: + """Compact the memories file by removing deleted entries.""" + if not self.memories_file.exists(): + return {"before": 0, "after": 0} + + # Read all valid memories + valid_memories = [] + original_size = self.memories_file.stat().st_size + + with open(self.memories_file) as f: + for line in f: + try: + data = json.loads(line) + if data.get("id") in self.index.get("by_id", {}): + valid_memories.append(line) + except Exception: + continue + + # Rewrite file + with open(self.memories_file, "w") as f: + for line in valid_memories: + f.write(line) + + new_size = self.memories_file.stat().st_size + + return { + "before_bytes": original_size, + "after_bytes": new_size, + "saved_bytes": original_size - new_size, + "valid_memories": len(valid_memories) } @@ -379,6 +552,72 @@ async def get_stats(): return store.get_stats() +@app.get("/memories", dependencies=[Depends(require_jwt)]) +async def list_memories( + category: Optional[MemoryCategory] = None, + limit: int = Query(default=50, le=200), + offset: int = Query(default=0, ge=0) +): + """List memories with pagination.""" + return store.list_paginated( + category=category.value if category else None, + limit=limit, + offset=offset + ) + + +@app.post("/export", dependencies=[Depends(require_jwt)]) +async def export_memories(): + """Export all memories for backup.""" + memories = store.export_all() + return { + "memories": memories, + "count": len(memories), + "exported_at": datetime.utcnow().isoformat() + "Z" + } + + +@app.post("/import", dependencies=[Depends(require_jwt)]) +async def import_memories(data: Dict[str, Any]): + """Import memories from backup.""" + memories = data.get("memories", []) + if not memories: + raise HTTPException(status_code=400, detail="No memories to import") + + result = store.import_memories(memories) + return result + + +@app.post("/cleanup", dependencies=[Depends(require_jwt)]) +async def cleanup_expired(): + """Remove expired memories.""" + count = store.cleanup_expired() + return {"expired_removed": count} + + +@app.post("/bulk-delete", dependencies=[Depends(require_jwt)]) +async def bulk_delete( + category: Optional[MemoryCategory] = None, + older_than_days: Optional[int] = None +): + """Delete multiple memories by criteria.""" + if not category and not older_than_days: + raise HTTPException(status_code=400, detail="Must specify category or older_than_days") + + count = store.bulk_delete( + category=category.value if category else None, + older_than_days=older_than_days + ) + return {"deleted": count} + + +@app.post("/compact", dependencies=[Depends(require_jwt)]) +async def compact_storage(): + """Compact storage by removing deleted entries.""" + result = store.compact() + return result + + @app.post("/summarize", dependencies=[Depends(require_jwt)]) async def summarize_memories( category: Optional[MemoryCategory] = None, diff --git a/packages/secubox-mcp-server/api/main.py b/packages/secubox-mcp-server/api/main.py index 1d7eceee..de2ebb73 100644 --- a/packages/secubox-mcp-server/api/main.py +++ b/packages/secubox-mcp-server/api/main.py @@ -7,7 +7,8 @@ Features: - Resource exposure (logs, configs, alerts) - Tool registration (security actions) - Prompt templates for security analysis -- Multi-module aggregation +- Multi-module aggregation with caching +- Configurable module port mapping """ import os import json @@ -16,8 +17,10 @@ import asyncio import subprocess from datetime import datetime, timedelta from pathlib import Path -from typing import Optional, Dict, List, Any, Callable +from typing import Optional, Dict, List, Any, Callable, Tuple from enum import Enum +from functools import lru_cache +import time from fastapi import FastAPI, Depends, HTTPException, WebSocket, WebSocketDisconnect from pydantic import BaseModel, Field @@ -31,10 +34,35 @@ CONFIG_PATH = Path("/etc/secubox/mcp-server.toml") DATA_DIR = Path("/var/lib/secubox/mcp-server") TOOLS_FILE = DATA_DIR / "tools.json" SESSIONS_FILE = DATA_DIR / "sessions.jsonl" +CACHE_DIR = Path("/tmp/secubox/mcp-cache") # MCP Protocol Version MCP_VERSION = "2024-11-05" +# Module port mapping - matches nginx internal backend config +MODULE_PORTS = { + "ai-gateway": 9100, + "localrecall": 9101, + "master-link": 9102, + "threat-analyst": 9103, + "cve-triage": 9104, + "iot-guard": 9105, + "config-advisor": 9106, + "mcp-server": 9107, + "dns-guard": 9108, + "network-anomaly": 9109, + "identity": 9110, + "system-hub": 9111, +} + +# Cache TTL in seconds +CACHE_TTL = { + "status": 30, + "alerts": 60, + "config": 300, + "logs": 30, +} + app = FastAPI(title="SecuBox MCP Server", version="1.0.0") logger = logging.getLogger("secubox.mcp-server") @@ -100,6 +128,62 @@ class MCPSession(BaseModel): last_activity: str +class CacheManager: + """Simple in-memory cache with TTL.""" + + def __init__(self, cache_dir: Path): + self.cache_dir = cache_dir + self._memory_cache: Dict[str, Tuple[Any, float]] = {} + cache_dir.mkdir(parents=True, exist_ok=True) + + def get(self, key: str, ttl: int = 60) -> Optional[Any]: + """Get cached value if not expired.""" + if key in self._memory_cache: + value, timestamp = self._memory_cache[key] + if time.time() - timestamp < ttl: + return value + + # Try file cache + cache_file = self.cache_dir / f"{key.replace('/', '_')}.json" + if cache_file.exists(): + try: + stat = cache_file.stat() + if time.time() - stat.st_mtime < ttl: + with open(cache_file) as f: + return json.load(f) + except Exception: + pass + return None + + def set(self, key: str, value: Any): + """Set cache value.""" + self._memory_cache[key] = (value, time.time()) + + # Also persist to file for cross-request caching + try: + cache_file = self.cache_dir / f"{key.replace('/', '_')}.json" + with open(cache_file, "w") as f: + json.dump(value, f) + except Exception: + pass + + def invalidate(self, key: str): + """Invalidate cache entry.""" + self._memory_cache.pop(key, None) + cache_file = self.cache_dir / f"{key.replace('/', '_')}.json" + if cache_file.exists(): + cache_file.unlink() + + def clear(self): + """Clear all cache.""" + self._memory_cache.clear() + for f in self.cache_dir.glob("*.json"): + try: + f.unlink() + except Exception: + pass + + class MCPServer: """MCP protocol server for SecuBox integration.""" @@ -107,8 +191,10 @@ class MCPServer: self.data_dir = data_dir self.tools_file = data_dir / "tools.json" self.sessions_file = data_dir / "sessions.jsonl" + self.cache = CacheManager(CACHE_DIR) self._ensure_dirs() self.sessions: Dict[str, MCPSession] = {} + self._http_client: Optional[httpx.AsyncClient] = None self._register_tools() self._register_resources() self._register_prompts() @@ -116,6 +202,17 @@ class MCPServer: def _ensure_dirs(self): self.data_dir.mkdir(parents=True, exist_ok=True) + async def _get_client(self) -> httpx.AsyncClient: + """Get or create HTTP client.""" + if self._http_client is None: + self._http_client = httpx.AsyncClient(timeout=10.0) + return self._http_client + + def _get_module_url(self, module: str, path: str = "") -> str: + """Get URL for a module.""" + port = MODULE_PORTS.get(module, 9100) + return f"http://127.0.0.1:{port}{path}" + def _register_tools(self): """Register available tools.""" self.tools: Dict[str, MCPTool] = { @@ -229,6 +326,43 @@ class MCPServer: "type": "object", "properties": {} } + ), + "secubox.localrecall.search": MCPTool( + name="secubox.localrecall.search", + description="Search local recall memory for security context", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "category": {"type": "string", "description": "Optional category filter"} + }, + "required": ["query"] + } + ), + "secubox.ai.query": MCPTool( + name="secubox.ai.query", + description="Query the AI gateway for security analysis", + inputSchema={ + "type": "object", + "properties": { + "prompt": {"type": "string", "description": "Analysis prompt"}, + "context": {"type": "string", "description": "Additional context"} + }, + "required": ["prompt"] + } + ), + "secubox.threat.generate_rule": MCPTool( + name="secubox.threat.generate_rule", + description="Generate a security rule from threat data", + inputSchema={ + "type": "object", + "properties": { + "threat_type": {"type": "string", "enum": ["ip", "domain", "pattern"]}, + "indicator": {"type": "string", "description": "IOC value"}, + "rule_type": {"type": "string", "enum": ["nftables", "crowdsec", "waf"]} + }, + "required": ["threat_type", "indicator"] + } ) } @@ -247,12 +381,24 @@ class MCPServer: description="CrowdSec security logs", mimeType="text/plain" ), + "secubox://logs/dns": MCPResource( + uri="secubox://logs/dns", + name="DNS Guard Logs", + description="DNS security and blocking logs", + mimeType="application/jsonl" + ), "secubox://config/haproxy": MCPResource( uri="secubox://config/haproxy", name="HAProxy Configuration", description="Current HAProxy configuration", mimeType="text/plain" ), + "secubox://config/nginx": MCPResource( + uri="secubox://config/nginx", + name="Nginx Configuration", + description="Current nginx configuration", + mimeType="text/plain" + ), "secubox://config/nftables": MCPResource( uri="secubox://config/nftables", name="Firewall Rules", @@ -403,48 +549,74 @@ class MCPServer: async def _execute_tool(self, tool_name: str, args: Dict) -> Any: """Execute tool and return result.""" - # Call appropriate SecuBox module API - module_map = { - "secubox.waf.status": ("http://127.0.0.1:8010/status", "GET"), - "secubox.waf.threats": ("http://127.0.0.1:8010/alerts", "GET"), - "secubox.crowdsec.alerts": ("cscli alerts list -o json", "CMD"), - "secubox.crowdsec.decisions": ("cscli decisions list -o json", "CMD"), - "secubox.dns.analyze": ("http://127.0.0.1:8030/analyze", "POST"), - "secubox.dns.blocklist": ("http://127.0.0.1:8030/blocklist", "GET"), - "secubox.network.anomalies": ("http://127.0.0.1:8031/alerts", "GET"), - "secubox.iot.devices": ("http://127.0.0.1:8032/devices", "GET"), - "secubox.cve.scan": ("http://127.0.0.1:8033/cves", "GET"), - "secubox.audit.run": ("http://127.0.0.1:8034/audit", "POST"), - "secubox.identity.info": ("http://127.0.0.1:8035/identity", "GET"), - "secubox.mesh.peers": ("http://127.0.0.1:8036/peers", "GET"), + # Tool to module/endpoint mapping using correct ports + tool_mapping = { + "secubox.waf.status": ("threat-analyst", "/status", "GET", True), + "secubox.waf.threats": ("threat-analyst", "/alerts", "GET", True), + "secubox.crowdsec.alerts": (None, "cscli alerts list -o json", "CMD", False), + "secubox.crowdsec.decisions": (None, "cscli decisions list -o json", "CMD", False), + "secubox.dns.analyze": ("dns-guard", "/analyze", "POST", False), + "secubox.dns.blocklist": ("dns-guard", "/blocklist", "GET", True), + "secubox.network.anomalies": ("network-anomaly", "/alerts", "GET", True), + "secubox.iot.devices": ("iot-guard", "/devices", "GET", True), + "secubox.cve.scan": ("cve-triage", "/cves", "GET", True), + "secubox.audit.run": ("config-advisor", "/audit", "POST", False), + "secubox.identity.info": ("identity", "/identity", "GET", True), + "secubox.mesh.peers": ("master-link", "/peers", "GET", True), + "secubox.localrecall.search": ("localrecall", "/search", "POST", False), + "secubox.ai.query": ("ai-gateway", "/query", "POST", False), } - if tool_name not in module_map: + if tool_name not in tool_mapping: return {"error": f"Tool {tool_name} not implemented"} - endpoint, method = module_map[tool_name] + module, path, method, cacheable = tool_mapping[tool_name] + + # Check cache for cacheable GET requests + cache_key = f"tool_{tool_name}_{hash(json.dumps(args, sort_keys=True))}" + if cacheable and method == "GET": + cached = self.cache.get(cache_key, ttl=CACHE_TTL.get("status", 30)) + if cached is not None: + return cached if method == "CMD": # Execute shell command try: result = subprocess.run( - endpoint.split(), + path.split(), capture_output=True, text=True, timeout=30 ) - return json.loads(result.stdout) if result.stdout else {"output": result.stderr} + output = json.loads(result.stdout) if result.stdout else {"output": result.stderr or "No output"} + return output + except json.JSONDecodeError: + return {"output": result.stdout if result.stdout else result.stderr} + except subprocess.TimeoutExpired: + return {"error": "Command timed out"} except Exception as e: return {"error": str(e)} else: - # HTTP request + # HTTP request to module + endpoint = self._get_module_url(module, path) try: - async with httpx.AsyncClient() as client: - if method == "GET": - response = await client.get(endpoint, params=args, timeout=10.0) - else: - response = await client.post(endpoint, json=args, timeout=10.0) - return response.json() + client = await self._get_client() + if method == "GET": + response = await client.get(endpoint, params=args) + else: + response = await client.post(endpoint, json=args) + + if response.status_code == 200: + result = response.json() + if cacheable: + self.cache.set(cache_key, result) + return result + else: + return {"error": f"HTTP {response.status_code}", "detail": response.text[:200]} + except httpx.TimeoutException: + return {"error": "Request timed out"} + except httpx.ConnectError: + return {"error": f"Cannot connect to {module} module"} except Exception as e: return {"error": str(e)} @@ -482,38 +654,97 @@ class MCPServer: resource_readers = { "secubox://logs/waf": self._read_waf_logs, "secubox://logs/crowdsec": self._read_crowdsec_logs, + "secubox://logs/dns": self._read_dns_logs, "secubox://config/haproxy": self._read_haproxy_config, "secubox://config/nftables": self._read_nftables, + "secubox://config/nginx": self._read_nginx_config, "secubox://alerts/all": self._read_all_alerts, "secubox://status/all": self._read_all_status, } + # Check cache for expensive resources + cache_key = f"resource_{uri.replace('://', '_').replace('/', '_')}" + cached = self.cache.get(cache_key, ttl=CACHE_TTL.get("logs", 30)) + if cached is not None: + return cached + reader = resource_readers.get(uri) if reader: - return await reader() + result = await reader() + self.cache.set(cache_key, result) + return result return "" async def _read_waf_logs(self) -> str: - log_file = Path("/var/log/mitmproxy/waf.jsonl") - if log_file.exists(): - lines = log_file.read_text().strip().split("\n") - return "\n".join(lines[-100:]) + """Read WAF (mitmproxy) logs.""" + log_paths = [ + Path("/var/log/mitmproxy/waf.jsonl"), + Path("/var/log/secubox/waf.jsonl"), + Path("/var/log/mitmproxy/access.log"), + ] + for log_file in log_paths: + if log_file.exists(): + try: + lines = log_file.read_text().strip().split("\n") + return "\n".join(lines[-100:]) + except Exception as e: + continue return "No WAF logs found" async def _read_crowdsec_logs(self) -> str: - log_file = Path("/var/log/crowdsec.log") - if log_file.exists(): - lines = log_file.read_text().strip().split("\n") - return "\n".join(lines[-100:]) + """Read CrowdSec logs.""" + log_paths = [ + Path("/var/log/crowdsec.log"), + Path("/var/log/crowdsec/crowdsec.log"), + ] + for log_file in log_paths: + if log_file.exists(): + try: + lines = log_file.read_text().strip().split("\n") + return "\n".join(lines[-100:]) + except Exception: + continue + + # Try journalctl as fallback + try: + result = subprocess.run( + ["journalctl", "-u", "crowdsec", "-n", "100", "--no-pager"], + capture_output=True, text=True, timeout=5 + ) + if result.stdout: + return result.stdout + except Exception: + pass + return "No CrowdSec logs found" + async def _read_dns_logs(self) -> str: + """Read DNS Guard logs.""" + log_file = Path("/var/log/secubox/dns-guard.jsonl") + if log_file.exists(): + try: + lines = log_file.read_text().strip().split("\n") + return "\n".join(lines[-100:]) + except Exception: + pass + return "No DNS logs found" + async def _read_haproxy_config(self) -> str: + """Read HAProxy configuration.""" config_file = Path("/etc/haproxy/haproxy.cfg") if config_file.exists(): return config_file.read_text() return "HAProxy config not found" + async def _read_nginx_config(self) -> str: + """Read nginx configuration.""" + config_file = Path("/etc/nginx/nginx.conf") + if config_file.exists(): + return config_file.read_text() + return "Nginx config not found" + async def _read_nftables(self) -> str: + """Read nftables firewall rules.""" try: result = subprocess.run( ["nft", "list", "ruleset"], @@ -521,26 +752,110 @@ class MCPServer: text=True, timeout=10 ) - return result.stdout + return result.stdout or "No rules found" + except FileNotFoundError: + # Try iptables as fallback + try: + result = subprocess.run( + ["iptables", "-L", "-n", "-v"], + capture_output=True, + text=True, + timeout=10 + ) + return result.stdout or "No rules found" + except Exception: + pass except Exception as e: - return f"Error: {e}" + return f"Error reading firewall rules: {e}" + return "No firewall rules found" async def _read_all_alerts(self) -> str: - alerts = {"waf": [], "crowdsec": [], "dns": [], "anomaly": []} - # Aggregate from all modules + """Aggregate alerts from all security modules.""" + alerts = { + "waf": [], + "crowdsec": [], + "dns": [], + "anomaly": [], + "iot": [], + "cve": [], + "timestamp": datetime.utcnow().isoformat() + "Z" + } + + client = await self._get_client() + + # Fetch from each module concurrently + async def fetch_alerts(module: str, path: str, key: str): + try: + url = self._get_module_url(module, path) + response = await client.get(url, timeout=5.0) + if response.status_code == 200: + data = response.json() + if isinstance(data, list): + alerts[key] = data[:20] # Limit to 20 per module + elif isinstance(data, dict) and "alerts" in data: + alerts[key] = data["alerts"][:20] + except Exception as e: + alerts[key] = [{"error": str(e)}] + + await asyncio.gather( + fetch_alerts("threat-analyst", "/alerts", "waf"), + fetch_alerts("dns-guard", "/alerts", "dns"), + fetch_alerts("network-anomaly", "/alerts", "anomaly"), + fetch_alerts("iot-guard", "/alerts", "iot"), + fetch_alerts("cve-triage", "/cves?severity=critical,high", "cve"), + return_exceptions=True + ) + + # Also try CrowdSec CLI + try: + result = subprocess.run( + ["cscli", "alerts", "list", "-o", "json", "-l", "20"], + capture_output=True, text=True, timeout=10 + ) + if result.stdout: + alerts["crowdsec"] = json.loads(result.stdout) + except Exception: + pass + return json.dumps(alerts, indent=2) async def _read_all_status(self) -> str: - modules = ["waf", "crowdsec", "dns-guard", "network-anomaly", "iot-guard"] - status = {} - for module in modules: + """Get status from all SecuBox modules.""" + status = { + "timestamp": datetime.utcnow().isoformat() + "Z", + "modules": {} + } + + client = await self._get_client() + + async def fetch_status(module: str): try: - async with httpx.AsyncClient() as client: - port = 8010 + modules.index(module) - response = await client.get(f"http://127.0.0.1:{port}/status", timeout=2.0) - status[module] = response.json() - except Exception: - status[module] = {"status": "unavailable"} + url = self._get_module_url(module, "/status") + response = await client.get(url, timeout=3.0) + if response.status_code == 200: + status["modules"][module] = response.json() + else: + status["modules"][module] = {"status": "error", "code": response.status_code} + except httpx.ConnectError: + status["modules"][module] = {"status": "unavailable"} + except Exception as e: + status["modules"][module] = {"status": "error", "error": str(e)} + + # Fetch all module statuses concurrently + await asyncio.gather( + *[fetch_status(module) for module in MODULE_PORTS.keys()], + return_exceptions=True + ) + + # Count healthy/unhealthy + healthy = sum(1 for m in status["modules"].values() + if isinstance(m, dict) and m.get("status") in ["ok", "healthy"]) + status["summary"] = { + "total": len(MODULE_PORTS), + "healthy": healthy, + "unhealthy": len(MODULE_PORTS) - healthy + } + return json.dumps(status, indent=2) async def _handle_prompts_list(self, request: MCPRequest) -> MCPResponse: @@ -701,12 +1016,74 @@ async def websocket_endpoint(websocket: WebSocket): logger.error(f"WebSocket error: {e}") +@app.post("/cache/clear", dependencies=[Depends(require_jwt)]) +async def clear_cache(): + """Clear MCP server cache.""" + mcp_server.cache.clear() + return {"status": "cleared"} + + +@app.get("/cache/stats", dependencies=[Depends(require_jwt)]) +async def cache_stats(): + """Get cache statistics.""" + memory_entries = len(mcp_server.cache._memory_cache) + file_entries = len(list(CACHE_DIR.glob("*.json"))) if CACHE_DIR.exists() else 0 + return { + "memory_entries": memory_entries, + "file_entries": file_entries, + "cache_dir": str(CACHE_DIR) + } + + +@app.get("/modules", dependencies=[Depends(require_jwt)]) +async def list_modules(): + """List all registered SecuBox modules and their ports.""" + return { + "modules": MODULE_PORTS, + "count": len(MODULE_PORTS) + } + + +@app.post("/tools/{tool_name}/call", dependencies=[Depends(require_jwt)]) +async def call_tool_direct(tool_name: str, arguments: Dict[str, Any] = {}): + """Directly call a tool without MCP protocol overhead.""" + if tool_name not in mcp_server.tools: + raise HTTPException(status_code=404, detail=f"Tool not found: {tool_name}") + + result = await mcp_server._execute_tool(tool_name, arguments) + return {"tool": tool_name, "result": result} + + +@app.get("/resources/{uri:path}/read", dependencies=[Depends(require_jwt)]) +async def read_resource_direct(uri: str): + """Directly read a resource without MCP protocol overhead.""" + full_uri = f"secubox://{uri}" + if full_uri not in mcp_server.resources: + raise HTTPException(status_code=404, detail=f"Resource not found: {full_uri}") + + content = await mcp_server._read_resource(full_uri) + return { + "uri": full_uri, + "content": content, + "mimeType": mcp_server.resources[full_uri].mimeType + } + + # ============================================================================ -# Startup +# Startup / Shutdown # ============================================================================ @app.on_event("startup") async def startup(): """Initialize on startup.""" DATA_DIR.mkdir(parents=True, exist_ok=True) - logger.info("MCP Server started") + CACHE_DIR.mkdir(parents=True, exist_ok=True) + logger.info(f"MCP Server started - {len(mcp_server.tools)} tools, {len(mcp_server.resources)} resources") + + +@app.on_event("shutdown") +async def shutdown(): + """Cleanup on shutdown.""" + if mcp_server._http_client: + await mcp_server._http_client.aclose() + logger.info("MCP Server stopped")