quantum-ai-v3/agents/utils.py

"""
Utility functions for the agents app.
Contains webhook validation, message formatting, and other helper functions.
"""

import ipaddress
from urllib.parse import urlparse


def validate_webhook_url(url):
    """
    Validate webhook URL to prevent SSRF attacks.
    Only allows HTTPS URLs to external, non-private networks.
    """
    try:
        parsed = urlparse(url)
        
        # Only allow HTTP/HTTPS protocols
        if parsed.scheme not in ['http', 'https']:
            raise ValueError("Only HTTP/HTTPS URLs are allowed")
        
        # Get hostname
        hostname = parsed.hostname
        if not hostname:
            raise ValueError("Invalid hostname in URL")
        
        # For localhost development, allow localhost URLs first
        if hostname in ['localhost', '127.0.0.1'] and parsed.port in [5678, 8000, 8080]:
            return True  # Allow N8N development server
        
        # Check if hostname is an IP address
        try:
            ip = ipaddress.ip_address(hostname)
            # Block private, loopback, and reserved IP ranges
            if (ip.is_private or ip.is_loopback or ip.is_reserved or 
                ip.is_link_local or ip.is_multicast):
                raise ValueError("Internal/private IP addresses are not allowed")
        except ValueError as e:
            if "does not appear to be an IPv4 or IPv6 address" not in str(e):
                raise  # Re-raise if it's not just a "not an IP" error
            # If it's not an IP, it's a domain name - that's fine
            
        return True
        
    except Exception as e:
        raise ValueError(f"Invalid webhook URL: {str(e)}")


def format_agent_message(agent_slug, input_data):
    """Format input data into a message for N8N webhook based on agent type"""
    if agent_slug == 'social-ads-generator':
        description = input_data.get('description', '')
        platform = input_data.get('social_platform', '')
        emoji = input_data.get('include_emoji', 'yes')
        language = input_data.get('language', 'English')
        
        return f"Execute Social Media Ad Creator with the following parameters:. Describe what you'd like to generate: {description}. Include Emoji: {emoji.title()}. For Social Media Platform: {platform.title()}. Language: {language}."
    
    elif agent_slug == 'job-posting-generator':
        job_title = input_data.get('job_title', '')
        company_name = input_data.get('company_name', '')
        description = input_data.get('job_description', '')
        seniority = input_data.get('seniority_level', '')
        contract = input_data.get('contract_type', '')
        location = input_data.get('location', '')
        language = input_data.get('language', 'English')
        
        return f"Create a professional job posting for: {job_title} at {company_name}. Description: {description}. Seniority: {seniority}. Contract: {contract}. Location: {location}. Language: {language}. Make it comprehensive and attractive to candidates."
    
    # Default formatting for other agents
    params = [f"{key}: {value}" for key, value in input_data.items() if value]
    return f"Execute {agent_slug.replace('-', ' ').title()} with parameters: {'. '.join(params)}."


class AgentCompat:
    """
    Compatibility class to convert file-based agent data to object format
    for templates and views that expect object attributes.
    """
    def __init__(self, data):
        self.slug = data['slug']
        self.name = data['name']
        self.price = float(data['price'])
        self.webhook_url = data['webhook_url']
        self.access_url_name = data.get('access_url_name', '')
        self.display_url_name = data.get('display_url_name', '')
        self.id = data['slug']  # Use slug as ID for file-based agents
        self.message_limit = data.get('message_limit', 50)