phishing_detector.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Phishing URL Detector - Terminal Application
This application checks whether user-entered URLs are phishing websites.
It can fetch real-time data from online phishing databases.
"""

import urllib.parse
import urllib.request
import json
import re
from datetime import datetime, timedelta


class PhishingDetector:
    """Main class for detecting phishing URLs with online data fetching"""

    def __init__(self, enable_online_check=True):
        """Initialize known phishing domains list and suspicious characters"""
        self.enable_online_check = enable_online_check
        self.last_update_time = None
        self.update_interval = timedelta(hours=1)  # Update every hour

        # Base known phishing domains (fallback list)
        self.known_phishing_domains = {
            'paypal-security.com',
            'paypal-verification.net',
            'amazon-security.org',
            'google-verify.com',
            'microsoft-login.net',
            'apple-id-verify.com',
            'facebook-security.org',
            'instagram-help.net',
            'twitter-verify.com',
            'linkedin-security.org',
            'netflix-billing.com',
            'spotify-premium.net',
            'dropbox-storage.org',
            'github-security.com',
            'steam-community.net',
            'ebay-secure.org',
            'whatsapp-verify.com',
            'telegram-security.net',
            'discord-nitro.org',
            'zoom-meeting.net'
        }

        self.suspicious_characters = ['@', '..', '--', '__']

        # Online phishing databases
        self.phishing_data_sources = [
            {
                'name': 'PhishTank API',
                'url': 'http://data.phishtank.com/data/online-valid.json',
                'format': 'json',
                'enabled': True
            },
            {
                'name': 'OpenPhish',
                'url': 'https://openphish.com/feed.txt',
                'format': 'txt',
                'enabled': True
            },
            {
                'name': 'Anti-Phishing Working Group',
                'url': 'https://apwg.org/phishing-attack-trends-reports/',
                'format': 'html',
                'enabled': False  # Requires custom parsing
            }
        ]

        # Try to load fresh data on initialization
        if self.enable_online_check:
            self.update_phishing_database()

    def update_phishing_database(self):
        """Update phishing database from online sources"""
        if not self.enable_online_check:
            return False

        # Check if update is needed
        if self.last_update_time:
            time_diff = datetime.now() - self.last_update_time
            if time_diff < self.update_interval:
                return False

        print("🔄 Updating phishing database from online sources...")

        new_domains = set()
        successful_updates = 0

        for source in self.phishing_data_sources:
            if not source['enabled']:
                continue

            try:
                domains = self._fetch_from_source(source)
                if domains:
                    new_domains.update(domains)
                    successful_updates += 1
                    print(f"✅ Updated from {source['name']}: "
                          f"{len(domains)} domains")
                else:
                    print(f"⚠️  No data from {source['name']}")

            except Exception as e:
                print(f"❌ Failed to update from {source['name']}: {str(e)}")

        if new_domains:
            # Add new domains to existing list
            self.known_phishing_domains.update(new_domains)
            self.last_update_time = datetime.now()
            print(f"✅ Database updated! Total domains: "
                  f"{len(self.known_phishing_domains)}")
            return True
        else:
            print("⚠️  No new domains found in online sources")
            return False

    def _fetch_from_source(self, source):
        """Fetch phishing domains from a specific source"""
        try:
            # Set a reasonable timeout
            request = urllib.request.Request(
                source['url'],
                headers={
                    'User-Agent': 'Mozilla/5.0 (Phishing-Detector/1.0)'
                }
            )

            with urllib.request.urlopen(request, timeout=10) as response:
                data = response.read()

                if source['format'] == 'json':
                    return self._parse_json_source(data)
                elif source['format'] == 'txt':
                    return self._parse_txt_source(data)
                else:
                    return set()

        except Exception as e:
            print(f"Error fetching from {source['name']}: {str(e)}")
            return set()

    def _parse_json_source(self, data):
        """Parse JSON format phishing data"""
        try:
            json_data = json.loads(data.decode('utf-8'))
            domains = set()

            # Handle PhishTank format
            if isinstance(json_data, list):
                for entry in json_data:
                    if isinstance(entry, dict) and 'url' in entry:
                        try:
                            parsed = urllib.parse.urlparse(entry['url'])
                            if parsed.netloc:
                                domains.add(parsed.netloc.lower())
                        except Exception:
                            continue

            return domains

        except Exception as e:
            print(f"Error parsing JSON data: {str(e)}")
            return set()

    def _parse_txt_source(self, data):
        """Parse text format phishing data (one URL per line)"""
        try:
            text_data = data.decode('utf-8')
            domains = set()

            for line in text_data.strip().split('\n'):
                line = line.strip()
                if line and not line.startswith('#'):
                    try:
                        # If line doesn't start with protocol, add http://
                        if not line.startswith(('http://', 'https://')):
                            line = 'http://' + line

                        parsed = urllib.parse.urlparse(line)
                        if parsed.netloc:
                            domains.add(parsed.netloc.lower())
                    except Exception:
                        continue

            return domains

        except Exception as e:
            print(f"Error parsing text data: {str(e)}")
            return set()

    def check_online_reputation(self, domain):
        """Check domain reputation using online services"""
        if not self.enable_online_check:
            return None

        # Simple online check using public DNS
        try:
            # Check if domain resolves (basic validation)
            import socket
            socket.gethostbyname(domain)
            return True  # Domain exists
        except Exception:
            return False  # Domain doesn't resolve

    def force_database_update(self):
        """Force an immediate database update"""
        self.last_update_time = None
        return self.update_phishing_database()

    def parse_url(self, url):
        """
        Parse URL and extract domain part

        Args:
            url (str): URL to check

        Returns:
            tuple: (scheme, domain, is_valid)
        """
        try:
            # Add http:// if URL doesn't start with scheme
            if not url.startswith(('http://', 'https://', 'ftp://')):
                url = 'http://' + url

            parsed_url = urllib.parse.urlparse(url)
            domain = parsed_url.netloc.lower()
            scheme = parsed_url.scheme.lower()

            # If domain is not empty and contains valid characters
            if domain and self._is_valid_domain(domain):
                return scheme, domain, True
            else:
                return scheme, domain, False

        except Exception:
            return None, None, False

    def _is_valid_domain(self, domain):
        """
        Check if domain is valid

        Args:
            domain (str): Domain to check

        Returns:
            bool: True if domain is valid
        """
        # Simple domain validation
        domain_pattern = re.compile(
            r'^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?'
            r'(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*$'
        )
        return bool(domain_pattern.match(domain))

    def check_known_phishing_domain(self, domain):
        """
        Check if domain is in the known phishing list

        Args:
            domain (str): Domain to check

        Returns:
            bool: True if it's a phishing domain
        """
        return domain in self.known_phishing_domains

    def check_suspicious_characters(self, url):
        """
        Check if URL contains suspicious characters

        Args:
            url (str): URL to check

        Returns:
            tuple: (has_suspicious, suspicious_chars_found)
        """
        found_chars = []
        for char in self.suspicious_characters:
            if char in url:
                found_chars.append(char)

        return len(found_chars) > 0, found_chars

    def check_https_scheme(self, scheme):
        """
        Check if URL uses HTTPS

        Args:
            scheme (str): URL scheme

        Returns:
            bool: True if using HTTPS
        """
        return scheme == 'https'

    def analyze_url(self, url):
        """
        Comprehensively analyze URL and perform risk assessment

        Args:
            url (str): URL to analyze

        Returns:
            dict: Analysis results
        """
        # Parse URL
        scheme, domain, is_valid = self.parse_url(url)

        if not is_valid:
            return self._create_invalid_url_result(domain, scheme)

        # Perform all checks
        security_results = self._perform_security_checks(url, domain, scheme)
        reasons, risk_factors = security_results

        # Determine final risk level
        risk_level, is_suspicious = self._calculate_risk_level(risk_factors)

        if not is_suspicious:
            reasons.append("No suspicious features detected")

        return {
            'is_suspicious': is_suspicious,
            'is_valid': True,
            'risk_level': risk_level,
            'reasons': reasons,
            'domain': domain,
            'scheme': scheme,
            'risk_factors': risk_factors,
            'online_check_enabled': self.enable_online_check
        }

    def _create_invalid_url_result(self, domain, scheme):
        """Create result object for invalid URLs"""
        return {
            'is_suspicious': True,
            'is_valid': False,
            'risk_level': 'HIGH',
            'reasons': ['Invalid or malformed URL format'],
            'domain': domain,
            'scheme': scheme,
            'online_check_enabled': self.enable_online_check
        }

    def _perform_security_checks(self, url, domain, scheme):
        """Perform all security checks and return reasons and risk factors"""
        reasons = []
        risk_factors = 0

        # 1. Known phishing domain check (includes online data)
        if self.check_known_phishing_domain(domain):
            reasons.append(f"'{domain}' is a known phishing domain")
            risk_factors += 3

        # 2. Suspicious character check
        suspicious_result = self.check_suspicious_characters(url)
        has_suspicious, suspicious_chars = suspicious_result
        if has_suspicious:
            reasons.append(
                f"Suspicious characters found: {', '.join(suspicious_chars)}")
            risk_factors += 2

        # 3. HTTPS check
        if not self.check_https_scheme(scheme):
            reasons.append("Not using secure HTTPS connection")
            risk_factors += 1

        # 4. Online checks
        if self.enable_online_check:
            online_reasons, online_risk = self._perform_online_checks(domain)
            reasons.extend(online_reasons)
            risk_factors += online_risk

        return reasons, risk_factors

    def _perform_online_checks(self, domain):
        """Perform online reputation and database checks"""
        reasons = []
        risk_factors = 0

        # Online reputation check
        reputation = self.check_online_reputation(domain)
        if reputation is False:
            reasons.append("Domain does not resolve (may be inactive)")
            risk_factors += 1
        elif reputation is True:
            reasons.append("Domain is active and resolves")

        # Database update check
        should_update = (
            self.last_update_time is None
            or datetime.now() - self.last_update_time > self.update_interval
        )

        if should_update:
            try:
                updated = self.update_phishing_database()
                if updated and self.check_known_phishing_domain(domain):
                    phishing_msg = f"'{domain}' is a known phishing domain"
                    if phishing_msg not in reasons:
                        reasons.append(
                            f"'{domain}' found in updated phishing database")
                        risk_factors += 3
            except Exception as e:
                reasons.append(f"Could not update online database: {str(e)}")

        return reasons, risk_factors

    def _calculate_risk_level(self, risk_factors):
        """Calculate risk level based on risk factors"""
        if risk_factors >= 4:
            return 'HIGH', True
        elif risk_factors >= 2:
            return 'MEDIUM', True
        elif risk_factors >= 1:
            return 'LOW', True
        else:
            return 'SAFE', False


def print_banner():
    """Print application header"""
    print("=" * 60)
    print("           🛡️  PHISHING URL DETECTOR  🛡️")
    print("=" * 60)
    print("This application checks if URLs are safe or potentially dangerous.")
    print("Features:")
    print("  • Real-time online phishing database updates")
    print("  • Local fallback database with 20+ known threats")
    print("  • Advanced suspicious character detection")
    print("  • HTTPS security validation")
    print("Type 'q' or 'exit' to quit.")
    print("-" * 60)


def print_analysis_result(analysis_result, url):
    """
    Print analysis results in a user-friendly way

    Args:
        analysis_result (dict): Analysis results
        url (str): Analyzed URL
    """
    print(f"\n📊 URL Analysis Result: {url}")
    print("-" * 50)

    if not analysis_result['is_valid']:
        print("❌ RESULT: INVALID URL")
        print("🔍 DOMAIN: Could not be detected")
        print("⚠️  RISK LEVEL: HIGH")
    else:
        # Result
        if analysis_result['is_suspicious']:
            print("❌ RESULT: SUSPICIOUS/DANGEROUS")
        else:
            print("✅ RESULT: SAFE")

        # Details
        print(f"🔍 DOMAIN: {analysis_result['domain']}")
        print(f"🔒 SCHEME: {analysis_result['scheme'].upper()}")
        print(f"⚠️  RISK LEVEL: {analysis_result['risk_level']}")

    # Reasons
    print("\n📋 DETAILS:")
    for i, reason in enumerate(analysis_result['reasons'], 1):
        if analysis_result['is_suspicious']:
            print(f"   {i}. ⚠️  {reason}")
        else:
            print(f"   {i}. ✅ {reason}")

    print("-" * 50)


def get_user_input():
    """
    Get URL input from user

    Returns:
        str: URL entered by user
    """
    while True:
        try:
            url = input("\n🌐 Enter URL to check: ").strip()
            if url:
                return url
            else:
                print("⚠️  Please enter a valid URL!")
        except (EOFError, KeyboardInterrupt):
            print("\n\n👋 Program terminated. Goodbye!")
            return None


def handle_special_commands(url, detector):
    """Handle special commands like update and stats."""
    if url.lower() == 'update':
        if detector.enable_online_check:
            print("🔄 Forcing database update...")
            updated = detector.force_database_update()
            if updated:
                domain_count = len(detector.known_phishing_domains)
                print(f"✅ Database updated! Total domains: {domain_count}")
            else:
                print("⚠️  No updates available")
        else:
            print("❌ Online checking is disabled")
        return True

    if url.lower() == 'stats':
        print_statistics(detector)
        return True

    return False


def setup_detector():
    """Set up and initialize the phishing detector."""
    print_banner()

    # Ask user about online features
    online_choice = get_online_preference()

    # Create phishing detector with user preference
    detector = PhishingDetector(enable_online_check=online_choice)

    # Show database info
    print(f"📊 Loaded {len(detector.known_phishing_domains)} phishing domains")
    if online_choice:
        print("🌐 Online checking enabled")
        if detector.last_update_time:
            time_format = '%Y-%m-%d %H:%M'
            last_update_str = detector.last_update_time.strftime(time_format)
            print(f"📅 Last update: {last_update_str}")
    else:
        print("🔒 Offline mode - using local database only")
    print("-" * 60)

    return detector


def process_url_analysis(url, detector):
    """Process URL analysis and display results."""
    try:
        result = detector.analyze_url(url)
        print_analysis_result(result, url)
    except Exception as e:
        print(f"\n❌ Error during analysis: {str(e)}")
        print("Please use a valid URL format.")


def main():
    """Main program function"""
    detector = setup_detector()

    while True:
        # Get URL from user
        url = get_user_input()

        # Exit check
        if url is None or url.lower() in ['q', 'quit', 'exit']:
            print("\n👋 Program terminated. Goodbye!")
            break

        # Handle special commands
        if handle_special_commands(url, detector):
            continue

        # Analyze URL
        process_url_analysis(url, detector)

        # Ask to continue
        print("\nWould you like to check another URL?")
        print("(Press Enter to continue, 'update' to refresh database,")
        print(" 'stats' for statistics, or 'q' to quit)")


def get_online_preference():
    """Ask user if they want to enable online checking"""
    while True:
        try:
            prompt = "\n🌐 Enable online phishing database updates? (y/N): "
            choice = input(prompt).strip().lower()
            if choice in ['y', 'yes']:
                return True
            elif choice in ['n', 'no', '']:
                return False
            else:
                print("Please enter 'y' for yes or 'n' for no")
        except (EOFError, KeyboardInterrupt):
            return False


def print_statistics(detector):
    """Print detector statistics"""
    print("\n📊 Phishing Detector Statistics")
    print("-" * 40)
    print(f"Total phishing domains: {len(detector.known_phishing_domains)}")
    online_status = 'Enabled' if detector.enable_online_check else 'Disabled'
    print(f"Online checking: {online_status}")
    if detector.last_update_time:
        time_format = '%Y-%m-%d %H:%M:%S'
        last_update_str = detector.last_update_time.strftime(time_format)
        print(f"Last update: {last_update_str}")
        time_since = datetime.now() - detector.last_update_time
        print(f"Time since update: {time_since}")
    else:
        print("Last update: Never")
    print(f"Suspicious characters checked: {detector.suspicious_characters}")
    print("-" * 40)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n👋 Program terminated by user. Goodbye!")
    except Exception as e:
        print(f"\n❌ Unexpected error: {str(e)}")
        print("Program terminated.")