MCP HTTP TAVILY DATE OAUTH

#!/usr/bin/env python3 """ Control de acceso para el servidor MCP """ import os from typing import List, Set try: from .logger import get_logger except ImportError: from logger import get_logger logger = get_logger("access-control") class AccessControl: """Gestión de control de acceso""" def __init__(self): # Cargar configuración desde variables de entorno self.allowed_emails = self._load_allowed_emails() self.allowed_domains = self._load_allowed_domains() self.require_approval = os.getenv("REQUIRE_MANUAL_APPROVAL", "false").lower() == "true" logger.info(f"Control de acceso configurado:") logger.info(f" - Emails permitidos: {len(self.allowed_emails)}") logger.info(f" - Dominios permitidos: {len(self.allowed_domains)}") logger.info(f" - Requiere aprobación manual: {self.require_approval}") def _load_allowed_emails(self) -> Set[str]: """Cargar emails permitidos desde variable de entorno""" emails_str = os.getenv("ALLOWED_EMAILS", "") if emails_str: emails = [email.strip().lower() for email in emails_str.split(",")] return set(emails) return set() def _load_allowed_domains(self) -> Set[str]: """Cargar dominios permitidos desde variable de entorno""" domains_str = os.getenv("ALLOWED_DOMAINS", "") if domains_str: domains = [domain.strip().lower() for domain in domains_str.split(",")] return set(domains) return set() def is_email_allowed(self, email: str) -> tuple[bool, str]: """ Verificar si un email está permitido Returns: (is_allowed, reason) """ email = email.lower().strip() # Si no hay restricciones configuradas, permitir todos if not self.allowed_emails and not self.allowed_domains: return True, "Sin restricciones configuradas" # Permitir emails de GitHub (para usuarios sin email público) if email.endswith("@github.com"): return True, "Email de GitHub permitido" # Verificar email específico if self.allowed_emails and email in self.allowed_emails: return True, "Email en lista blanca" # Verificar dominio if self.allowed_domains: domain = email.split("@")[1] if "@" in email else "" if domain in self.allowed_domains: return True, f"Dominio {domain} permitido" # Si llegamos aquí, no está permitido if self.allowed_emails: return False, f"Email {email} no está en la lista de emails permitidos" if self.allowed_domains: domain = email.split("@")[1] if "@" in email else "" return False, f"Dominio {domain} no está en la lista de dominios permitidos" return False, "Acceso denegado" def check_user_access(self, email: str, user_exists: bool = False) -> tuple[bool, str]: """ Verificar acceso completo del usuario Args: email: Email del usuario user_exists: Si el usuario ya existe en el sistema Returns: (is_allowed, message) """ # 1. Verificar si el email está permitido email_allowed, email_reason = self.is_email_allowed(email) if not email_allowed: logger.warning(f"Acceso denegado para {email}: {email_reason}") return False, email_reason # 2. Si requiere aprobación manual y es usuario nuevo if self.require_approval and not user_exists: logger.info(f"Nuevo usuario requiere aprobación: {email}") return False, "Tu cuenta está pendiente de aprobación por el administrador" # 3. Todo OK logger.info(f"Acceso permitido para {email}: {email_reason}") return True, f"Acceso permitido: {email_reason}" def add_allowed_email(self, email: str): """Agregar email a la lista permitida (para uso programático)""" self.allowed_emails.add(email.lower().strip()) logger.info(f"Email agregado a lista permitida: {email}") def remove_allowed_email(self, email: str): """Remover email de la lista permitida""" self.allowed_emails.discard(email.lower().strip()) logger.info(f"Email removido de lista permitida: {email}") def get_stats(self) -> dict: """Obtener estadísticas de control de acceso""" return { "allowed_emails_count": len(self.allowed_emails), "allowed_domains_count": len(self.allowed_domains), "require_approval": self.require_approval, "has_restrictions": bool(self.allowed_emails or self.allowed_domains) } # Instancia global access_control = AccessControl()