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)