IoT_PreTester_Demo/V4.5/django_project/IoTAnalyzer/analyze/analyze_url.py

import dns.resolver
import requests
import ssl
import socket
import nmap
from django.db import transaction
import os
import re

from IoTAnalyzer.db_create import create_analyzed_url
from IoTAnalyzer.models import Tool, TlsVersion, CipherSuite, UrlRaw, AnalyzedDataset
from urllib.parse import urlparse
from datetime import datetime

LOG_FILE = "errors.log"  # Log file for errors


def log_error(message):
    """
    Logs an error message to the log file with a timestamp.
    """
    with open(LOG_FILE, "a") as log:
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log.write(f"[{timestamp}] {message}\n")


def validate_url(url):
    """
    Validates if a given string is a properly formatted URL or domain.
    """
    try:
        result = urlparse(url)
        return result.netloc or result.path
    except Exception:
        return False


def resolve_dns(query):
    """
    Resolves a DNS query and returns a list of resolved IPs.
    """
    try:
        result = dns.resolver.resolve(query, 'A')
        resolved_ips = [str(rdata) for rdata in result]
        if not resolved_ips:
            log_error(f"No IPs resolved for DNS query: {query}")
        return resolved_ips
    except dns.resolver.NXDOMAIN:
        log_error(f"NXDOMAIN: No such domain {query}")
    except dns.resolver.Timeout:
        log_error(f"Timeout: DNS query for {query} timed out")
    except dns.resolver.NoNameservers:
        log_error(f"NoNameservers: No name servers found for {query}")
    except Exception as e:
        log_error(f"General DNS error for {query}: {e}")
    return []


def scan_ports(target):
    """
    Scans the given IP for open ports using Nmap.
    Returns a list of open ports.
    """
    nm = nmap.PortScanner()
    open_ports = []

    try:
        nm.scan(hosts=target, arguments='-sT -Pn -p 22,80,443,8000,8080 --scan-delay 500ms --max-retries 1')
        #nm.scan(hosts=ip, arguments='-sS -Pn -F')
        if target in nm.all_hosts():
            for port in nm[target].all_tcp():
                if 'open' in nm[target]['tcp'][port]['state']:
                    open_ports.append(port)
    except Exception as e:
        log_error(f"Error scanning IP {target}: {e}")

    if not open_ports:
        print("Python-Nmap failed, using OS-level Nmap...")
        os.system(f"nmap -sT -Pn -p 22,80,443,8000,8080 {target} > nmap_output.txt")
        with open("nmap_output.txt", "r") as f:
            output = f.read()
            if "open" in output:
                open_ports = []
                for port in [22, 80, 443, 8000, 8080]:
                    pattern = rf"^{port}/tcp\s+open\s"
                    if re.search(pattern, output, re.MULTILINE):
                        open_ports.append(port)

    return open_ports


def scan_ssl(target):
    """
    Performs an SSL scan on the given domain or IP.
    Retrieves TLS version, cipher suite, and certificate presence.
    """
    context = ssl.create_default_context()
    context.check_hostname = True
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.verify_mode = ssl.CERT_REQUIRED
    context.load_default_certs()
    context.set_alpn_protocols(["http/1.1", "h2"])
    context.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1  # Disable old protocols

    scan_result = {
        'tls_version': None,
        'cipher': None,
        'certificate': None,
    }

    try:
        with socket.create_connection((target, 443), timeout=5) as sock:
            with context.wrap_socket(sock, server_hostname=target) as ssock:
                cert = ssock.getpeercert()
                cipher = ssock.cipher()
                tls_version = ssock.version()

                scan_result['certificate'] = 'Present' if cert else 'Not Present'
                scan_result['cipher'] = cipher[0] if cipher else None
                scan_result['tls_version'] = tls_version
    except ssl.SSLError as e:
        log_error(f"SSL Error during scan for {target}: {e}")
    except socket.timeout:
        log_error(f"Timeout during SSL scan for {target}")
    except Exception as e:
        log_error(f"General error during SSL scan for {target}: {e}")

    return scan_result


def fetch_important_headers(url):
    """
    Fetches only important headers (e.g., Server, Content-Type).
    """
    try:
        response = requests.head(url, timeout=5)
        headers = response.headers
        important_headers = {
            "Server": headers.get("Server", "N/A"),
            "Content-Type": headers.get("Content-Type", "N/A"),
            "Strict-Transport-Security": headers.get("Strict-Transport-Security", "N/A"),
        }
        return important_headers
    except Exception as e:
        log_error(f"Error fetching headers for {url}: {e}")
        return {}


def analyze_target(query, targets, analyzed_dataset):
    """
    Ensures each URL is analyzed only once.
    Combines SSL, HTTP headers, and port scanning for a DNS query and its targets.
    Saves the results into the database with detailed logging.
    """
    target = query.replace("http://", "").replace("https://", "").split('/')[0]
    ssl_info = scan_ssl(target)
    headers = fetch_important_headers(f"https://{target}")
    open_ports = scan_ports(target)

    print(f"Analyzing URL: {query}")
    print(f"SSL Info: {ssl_info}")
    print(f"Headers: {headers}")
    print(f"Open Ports: {open_ports}")

    with transaction.atomic():
        tool, _ = Tool.objects.get_or_create(tool_name="DNS Analyzer", version="1.0")

        if ssl_info['tls_version'] == None:
            tls_version, _ = TlsVersion.objects.get_or_create(tls_version="N/A")
        else:
            tls_version = ssl_info['tls_version']

        if ssl_info['cipher'] == None:
            cipher, _ = CipherSuite.objects.get_or_create(cipher_suite="N/A")
        else:
            cipher = ssl_info['cipher']

        create_analyzed_url(analyzed_dataset.id, tool.id, query, open_ports, [tls_version], [cipher])

        log_error(f"Data saved for URL: {query}")



def get_urls_from_db(raw_url_id):
    """
    Main function to handle the URL analysis for a given raw_url_id.
    """
    try:
        raw_url = UrlRaw.objects.get(id=raw_url_id)
    except UrlRaw.DoesNotExist:
        log_error(f"Raw URL with ID {raw_url_id} not found.")
        return

    queries = [raw_url.url]
    analysis_targets = {}

    for query in queries:
        if not validate_url(query):
            log_error(f"Invalid URL: {query}")
            continue

        ips = resolve_dns(query)
        analysis_targets[query] = ips if ips else [query]

    # analyzed_dataset = raw_url.raw_dataset.file.analyzeddataset_set.last()
    analyzed_dataset, _ = AnalyzedDataset.objects.get_or_create(file=raw_url.raw_dataset.file)
    if not analyzed_dataset:
        log_error(f"No analyzed dataset found for the file associated with URL: {raw_url.url}")
        return

    for query, targets in analysis_targets.items():
        analyze_target(query, targets, analyzed_dataset)