From dc2cf390c04d3011c389e0a7e0afbf7bd53de908 Mon Sep 17 00:00:00 2001 From: cgedris Date: Wed, 18 Feb 2026 09:20:29 -0700 Subject: [PATCH 1/4] Firewall endpoint added and IP_REP implemented --- Scoring Engine/.gitignore | 3 +++ Scoring Engine/config.py | 13 +++++++------ Scoring Engine/scoring_logic.py | 17 ++++++++++++++--- 3 files changed, 24 insertions(+), 9 deletions(-) diff --git a/Scoring Engine/.gitignore b/Scoring Engine/.gitignore index c7e1059..b677159 100644 --- a/Scoring Engine/.gitignore +++ b/Scoring Engine/.gitignore @@ -4,3 +4,6 @@ __pycache__/ batch_run.py url_batch.txt all_scores.txt +bad_ips.txt +radix_test.py +.venv diff --git a/Scoring Engine/config.py b/Scoring Engine/config.py index 923f903..77243f7 100644 --- a/Scoring Engine/config.py +++ b/Scoring Engine/config.py @@ -19,12 +19,12 @@ # Weights for each component in the final score calculation WEIGHTS = { 'Connection_Security': 18, - 'Certificate_Health': 16, - 'DNS_Record_Health': 15, - 'Domain_Reputation': 23, + 'Certificate_Health': 15, + 'DNS_Record_Health': 14, + 'Domain_Reputation': 24, 'WHOIS_Pattern': 10, #unused currently - 'IP_Reputation': 0, #unused currently (probably won't be used) - 'Credential_Safety': 18 + 'IP_Reputation': 2, #unused currently (probably won't be used) + 'Credential_Safety': 17 } @@ -35,7 +35,8 @@ 'hval', 'mail', 'method', - 'rdap' + 'rdap', + 'firewall' ] # Default target hostname used if no argument is provided diff --git a/Scoring Engine/scoring_logic.py b/Scoring Engine/scoring_logic.py index 74d9541..ecc6689 100644 --- a/Scoring Engine/scoring_logic.py +++ b/Scoring Engine/scoring_logic.py @@ -377,12 +377,18 @@ def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): #TODO: IMPLE if app_config.VERBOSE: print("Cred Safety Score: Significant Deduction - HSTS header missing. (CRED_SAFETY)", file=sys.stderr) -def score_ip_rep(dns_data:dict, hval_data:dict, scores:dict): #PAUSED: Further investigation needed to determine if helpful +def score_ip_rep(firewall_data:dict, scores:dict): #PAUSED: Further investigation needed to determine if helpful """Placeholder for IP Reputation scoring function. Currently unused, but can be implemented in the future. """ - - pass + blocked = firewall_data.get("Block", False) + if blocked: + scores['IP_Reputation'] -= 100 + if app_config.VERBOSE: + print("IP Reputation Score: CRITICAL - IP is listed on a firewall blocklist. (IP_REP)", file=sys.stderr) + else: + if app_config.VERBOSE: + print("IP Reputation Score: No deduction - IP is not listed on firewall blocklist. (IP_REP)", file=sys.stderr) def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO: IMPLEMENT """Placeholder for WHOIS Pattern scoring function. @@ -595,6 +601,11 @@ def calculate_security_score(all_scans: dict, scan_date: datetime) -> dict: #CHA except Exception as e: print(f"Error in whois_pattern scan: {e}", file=sys.stderr) + try: + score_ip_rep(all_scans['firewall_scan'], scores) + except Exception as e: + print(f"Error in ip_rep scan: {e}", file=sys.stderr) + # 3. Clamp scores between 1 and 100 after all deductions for key in scores: scores[key] = max(1, min(100, scores[key])) From 2435e3cbad839ab2c5565dbd36f18b5d3c14713b Mon Sep 17 00:00:00 2001 From: cgedris Date: Thu, 5 Mar 2026 09:00:17 -0700 Subject: [PATCH 2/4] WHOIS: registrar scoring --- Scoring Engine/config.py | 14 ++++++++++++++ Scoring Engine/scoring_logic.py | 16 +++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/Scoring Engine/config.py b/Scoring Engine/config.py index 77243f7..ec7b791 100644 --- a/Scoring Engine/config.py +++ b/Scoring Engine/config.py @@ -70,3 +70,17 @@ "support","tel", "to", "today", "top", "tr", "tv", "ua", "us", "vip", "wiki", "world", "ws", "xn--q9jyb4c", "xyz" ] + +MAL_REGISTRARS = [ + "StanCo", "Istanco", "Hangzhou Yunji", "FlokiNET", "NauNet", "OPENPROV-RU", "DomainDelights", "Navicosoft Pty", "Shock Hosting", + "nicenic.net", "DropCatch.com 1422", "Dynu Systems Incorporated", "RegRU", "Hello Internet Corp", "PortlandNames.com", "Dynadot", + "Sav.com", "Gname", "WebNic.cc", "Mat Bao Corporation", "Immaterialism Limited", "MAXNAME-RU", "FE-RU", "温州市中网计算机技术服务有限公司", + "Namecheap", "Registrar R01.ru", "SPRINTNAMES-RU", "RegRU", "NIC.UA", "Namecheap", "NameSilo", "WebNic.cc", "Name SRS AB", "XServer", + "PDR", "OwnRegistrar", "Trunkoz", "Hostinger", "GMO", "Tucows", "Sav.com", "Realtime Register", "RU-Center", "Name.com", "Openprovider", + "Dominet", "GoDaddy", "Ultahost", "WebNic.cc", "河北识道网络科技有限公司", "MainReg Inc.", "Todaynic", "Eranet International" "长春市智绘网络科技有限公司", + "厦门三五互联信息有限公司", "南昌知乐远科技有限公司", "长沙小豆网络科技有限公司", "西部数码国际有限公司", "GKG NET", "成都垦派科技有限公司", + "四川域趣网络科技有限公司", "海口智慧康网络科技有限公司", "Global Domain Group", "Beijing Dongfang Ruipeng Digital Information Technology Co.", + "武汉物与伦比科技有限公司", "厦门纳网科技股份有限公司", "成都西维数码科技有限公司", "west263.com", "rocket" +] + + diff --git a/Scoring Engine/scoring_logic.py b/Scoring Engine/scoring_logic.py index 82bfaaa..f6e9572 100644 --- a/Scoring Engine/scoring_logic.py +++ b/Scoring Engine/scoring_logic.py @@ -541,7 +541,21 @@ def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO if app_config.VERBOSE: print("WHO_IS Score: Registrar name is ", registrar_name, file=sys.stderr) - #TODO: put registrar scoring here + # 2. Check for registrary matches from MAL_REGISTRARS list (case-insensitive substring match) + found_matches = False + for reg in app_config.MAL_REGISTRARS: + # We convert BOTH to lowercase to ensure "NameSilo" matches "namesilo" + if reg.lower() in registrar_name.lower(): + found_matches = True + break + + if found_matches: + scores['WHOIS_Pattern'] -= 10 + if app_config.VERBOSE: + print("WHO_IS Score: Suspicious Registrar found: ", registrar_name, file=sys.stderr) + else: + if app_config.VERBOSE: + print("WHO_IS Score: Trustworthy Registrar found: ", registrar_name, file=sys.stderr) def calculate_final_score(weights, scores): #CHANGE """ From 7ababdc7165c082325d32fe657af5077fb861c16 Mon Sep 17 00:00:00 2001 From: cgedris Date: Wed, 11 Mar 2026 09:10:24 -0600 Subject: [PATCH 3/4] remove method scan --- Scoring Engine/config.py | 8 -------- Scoring Engine/scoring_logic.py | 29 ++--------------------------- 2 files changed, 2 insertions(+), 35 deletions(-) diff --git a/Scoring Engine/config.py b/Scoring Engine/config.py index ec7b791..3b7bc86 100644 --- a/Scoring Engine/config.py +++ b/Scoring Engine/config.py @@ -2,13 +2,6 @@ # --- Configuration and Constants --- -# Bitmasks for Method Scan (Flag) -METHOD_FLAGS = { - 'HEAD': 1, 'GET': 2, 'POST': 4, - 'PUT': 8, 'PATCH': 16, 'DELETE': 32, - 'TRACE': 64, 'CONNECT': 128 -} - # Bitmasks for HVAL Scan (Security Flag) SECURITY_FLAGS = { 'HSTS': 1, 'CSP': 2, 'XCTO': 4, @@ -34,7 +27,6 @@ 'dns', 'hval', 'mail', - 'method', 'rdap', 'firewall' ] diff --git a/Scoring Engine/scoring_logic.py b/Scoring Engine/scoring_logic.py index f6e9572..185ac84 100644 --- a/Scoring Engine/scoring_logic.py +++ b/Scoring Engine/scoring_logic.py @@ -234,7 +234,7 @@ def score_conn_sec(hval_data: dict, cert_data: dict, scores: dict): if app_config.VERBOSE: print(f"HVAL Score: Significant Deduction - Outdated TLS version: {tls_version}. (CONN_SEC)", file=sys.stderr) -def score_dom_rep(mail_data: dict, method_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION +def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION """Unifies Domain Reputation scoring from Mail, Method, and RDAP scans.""" #ADD: tld scoring (list of top 20 suspicious, add points for gov/edu?) # --- Mail Scan --- @@ -294,31 +294,6 @@ def score_dom_rep(mail_data: dict, method_data: dict, rdap_data: dict, scores: d if app_config.VERBOSE: print(f"Mail Score: Deduction - SPF policy is too permissive ('{spf_string[-4:]}'). (DOM_REP)", file=sys.stderr) - # --- Method Scan --- - # 1. Check for Dangerous Methods (Major Deductions) - flag = method_data.get("flag", 0) - - # CONNECT AND PATCH (128, 16) - Tunneling/Modification Risk - if flag & (app_config.METHOD_FLAGS['CONNECT'] | app_config.METHOD_FLAGS['PATCH']): - scores['Domain_Reputation'] -= 7 - if app_config.VERBOSE: - print("Method Score: Deduction - possible modification/tunneling risk (CONNECT and/or PATCH). (DOM_REP)", file=sys.stderr) - - # PUT, DELETE, and TRACE (8, 32, 64) - Editing Risk - if flag & (app_config.METHOD_FLAGS['TRACE'] | app_config.METHOD_FLAGS['DELETE'] | app_config.METHOD_FLAGS['PUT']): - scores['Domain_Reputation'] -= 20 - if app_config.VERBOSE: - print("Method Score: Major Deduction - DELETE, TRACE, and/or PUT methods enabled. (DOM_REP)", file=sys.stderr) - - # 2. Optimal Check (Positive Bonus) - # Optimal for a public web page is usually only HEAD (1) and GET (2), resulting in flag 3. - if flag == 3: - if app_config.VERBOSE: - print("Method Score: Optimal - Only HEAD and GET methods enabled. (DOM_REP)", file=sys.stderr) - elif flag == 7: - if app_config.VERBOSE: - print("Method Score: Acceptable - HEAD, GET, and POST methods enabled. (DOM_REP)", file=sys.stderr) - # --- RDAP Scan --- nameservers = rdap_data[0].get("nameserver", []) @@ -644,7 +619,7 @@ def calculate_security_score(all_scans: dict, scan_date: datetime) -> dict: #CHA print(f"Error in conn_sec scan: {e}", file=sys.stderr) try: - score_dom_rep(all_scans['mail_scan'], all_scans['method_scan'], all_scans['rdap_scan'], scores) + score_dom_rep(all_scans['mail_scan'], all_scans['rdap_scan'], scores) except Exception as e: print(f"Error in dom_rep scan: {e}", file=sys.stderr) From 8756c027f0e3834a9bd4270f2a12b58ac4cffa90 Mon Sep 17 00:00:00 2001 From: cgedris Date: Tue, 17 Mar 2026 08:50:14 -0600 Subject: [PATCH 4/4] Beautified Code - comments and cleanup --- Scoring Engine/config.py | 20 +++- Scoring Engine/data_fetch.py | 24 ++-- Scoring Engine/scoring_logic.py | 205 ++++++++++++++------------------ Scoring Engine/scoring_main.py | 25 ++-- 4 files changed, 136 insertions(+), 138 deletions(-) diff --git a/Scoring Engine/config.py b/Scoring Engine/config.py index 3b7bc86..c61018c 100644 --- a/Scoring Engine/config.py +++ b/Scoring Engine/config.py @@ -15,12 +15,12 @@ 'Certificate_Health': 15, 'DNS_Record_Health': 14, 'Domain_Reputation': 24, - 'WHOIS_Pattern': 10, #unused currently - 'IP_Reputation': 2, #unused currently (probably won't be used) + 'WHOIS_Pattern': 10, + 'IP_Reputation': 2, 'Credential_Safety': 17 } - +# NetSTAR endpoints that are used to fetch information BASE_URL = 'https://w4.netstar.dev/' API_ENDPOINTS = [ 'cert', @@ -37,6 +37,7 @@ # Verbose mode flag VERBOSE = False +# List of TLDs from Spamhaus (includes .com and other very common TLDs) MAL_TLDS = [ "ac", "ai", "app", "at", "autos", "biz", "bond", "br", "bz", "ca", "cc", "cfd", "claims", "click", "cn", "co", "com", "coupons", "courses", @@ -50,6 +51,7 @@ "us", "vip", "wiki", "world", "ws" ,"xn--q9jyb4c" ,"xyz" ] +# List of TLDs from Spamhaus ( Listed TLDs below removed) MAL_TLDS_SLIM = [ #removed co, com, eu, uk, org, net "ac", "ai", "app", "at", "autos", "biz", "bond", "br", "bz", "ca", "cc", "cfd", "claims", "click", "cn", "coupons", "courses", @@ -63,6 +65,7 @@ "us", "vip", "wiki", "world", "ws", "xn--q9jyb4c", "xyz" ] +# List of registrars from Spamhaus MAL_REGISTRARS = [ "StanCo", "Istanco", "Hangzhou Yunji", "FlokiNET", "NauNet", "OPENPROV-RU", "DomainDelights", "Navicosoft Pty", "Shock Hosting", "nicenic.net", "DropCatch.com 1422", "Dynu Systems Incorporated", "RegRU", "Hello Internet Corp", "PortlandNames.com", "Dynadot", @@ -75,4 +78,13 @@ "武汉物与伦比科技有限公司", "厦门纳网科技股份有限公司", "成都西维数码科技有限公司", "west263.com", "rocket" ] - +CLIENT_LOCKS = [ + "client delete prohibited", + "client transfer prohibited", + "client update prohibited" + ] +SERVER_LOCKS = [ + "server delete prohibited", + "server transfer prohibited", + "server update prohibited" +] diff --git a/Scoring Engine/data_fetch.py b/Scoring Engine/data_fetch.py index efdeb0d..8112d6c 100644 --- a/Scoring Engine/data_fetch.py +++ b/Scoring Engine/data_fetch.py @@ -14,7 +14,7 @@ def execute_curl_command(command: List[str]) -> Optional[str]: #KEEP Handles potential errors during execution. """ if app_config.VERBOSE: - print(f"Executing command: {' '.join(command)}", file=sys.stderr) + print(f"Executing command: {' '.join(command)}") try: # Run the command, capture stdout, and decode as text result = subprocess.run( @@ -28,8 +28,8 @@ def execute_curl_command(command: List[str]) -> Optional[str]: #KEEP if result.returncode != 0: # Report the error code and stderr if the command failed if app_config.VERBOSE: - print(f"Error executing command. Return code: {result.returncode}", file=sys.stderr) - print(f"Standard Error:\n{result.stderr.strip()}", file=sys.stderr) + print(f"Error executing command. Return code: {result.returncode}") + print(f"Standard Error:\n{result.stderr.strip()}") return None # The output is returned as a string (JSON) @@ -37,15 +37,15 @@ def execute_curl_command(command: List[str]) -> Optional[str]: #KEEP except FileNotFoundError: if app_config.VERBOSE: - print("Error: The 'curl' command was not found. Make sure it is installed and in your system PATH.", file=sys.stderr) + print("Error: The 'curl' command was not found. Make sure it is installed and in your system PATH.") return None except subprocess.TimeoutExpired: if app_config.VERBOSE: - print("Error: Command execution timed out.", file=sys.stderr) + print("Error: Command execution timed out.") return None except Exception as e: if app_config.VERBOSE: - print(f"An unexpected error occurred during execution: {e}", file=sys.stderr) + print(f"An unexpected error occurred during execution: {e}") return None def process_single_endpoint(host: str, endpoint: str) -> tuple[str | None, dict | None]: @@ -69,7 +69,7 @@ def process_single_endpoint(host: str, endpoint: str) -> tuple[str | None, dict # 3. Define the cURL command (with exception for RDAP POST) if app_config.VERBOSE: - print(f"\n[Processing Endpoint: {endpoint.upper()}]", file=sys.stderr) + print(f"\n[Processing Endpoint: {endpoint.upper()}]") CURL_COMMAND = [] # Initialize command list @@ -100,7 +100,7 @@ def process_single_endpoint(host: str, endpoint: str) -> tuple[str | None, dict if output is None: if app_config.VERBOSE: - print(f"--> Endpoint {endpoint.upper()} failed execution. Skipping.", file=sys.stderr) + print(f"--> Endpoint {endpoint.upper()} failed execution. Skipping.") return (None, None) # 5. Parse the JSON output @@ -110,11 +110,11 @@ def process_single_endpoint(host: str, endpoint: str) -> tuple[str | None, dict return (scan_key, data) except json.JSONDecodeError: if app_config.VERBOSE: - print(f"--> Endpoint {endpoint.upper()} returned invalid JSON. Skipping.", file=sys.stderr) + print(f"--> Endpoint {endpoint.upper()} returned invalid JSON. Skipping.") return (None, None) except Exception as e: if app_config.VERBOSE: - print(f"--> An error occurred processing {endpoint.upper()}: {e}", file=sys.stderr) + print(f"--> An error occurred processing {endpoint.upper()}: {e}") return (None, None) def fetch_scan_data_concurrent(host: str) -> dict: @@ -124,7 +124,7 @@ def fetch_scan_data_concurrent(host: str) -> dict: """ all_scans = {} if app_config.VERBOSE: - print(f"\n--- Fetching live data for {host} from NetStar API (via concurrent cURL) ---", file=sys.stderr) + print(f"\n--- Fetching live data for {host} from NetStar API (via concurrent cURL) ---") print(f"\"url\": \"{host}\"") # Use ThreadPoolExecutor to run tasks in parallel @@ -145,5 +145,5 @@ def fetch_scan_data_concurrent(host: str) -> dict: all_scans[scan_key] = data if app_config.VERBOSE: - print("\n--- Data fetching complete ---", file=sys.stderr) + print("\n--- Data fetching complete ---") return all_scans diff --git a/Scoring Engine/scoring_logic.py b/Scoring Engine/scoring_logic.py index 185ac84..a385583 100644 --- a/Scoring Engine/scoring_logic.py +++ b/Scoring Engine/scoring_logic.py @@ -1,14 +1,10 @@ -# from asyncio import events -# from logging import config -import math from datetime import datetime -# from typing import Dict import config as app_config import sys # --- Scoring Functions --- -def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Change data to cert_data and fix any waterfall affect from it +def score_cert_health(cert_data: dict, scan_date: datetime, scores: dict): #TODO: Change data to cert_data and fix any waterfall affect from it """Calculates the score for the Certificate Scan (Max Score: 100). Focuses on validity and time to expiration. @@ -17,14 +13,14 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha """ try: # --- MODIFIED: The certificate chain is the 'data' parameter itself (a list) --- - connection_data = data.get("connection", {}) - verification_data = data.get("verification", {}) - cert_list = data.get("certs", []) + connection_data = cert_data.get("connection", {}) + verification_data = cert_data.get("verification", {}) + cert_list = cert_data.get("certs", []) # Check if the list exists and is not empty if not cert_list or not isinstance(cert_list, list) or len(cert_list) == 0: if app_config.VERBOSE: - print("Cert Score: CRITICAL - No certificates found in response. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: CRITICAL - No certificates found in response. (CERT_HEALTH)") # Note: Deducting from Certificate_Health initialized at 100 scores['Certificate_Health'] -= 50 return # Exit the function if no certs are found @@ -40,11 +36,10 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha # Check if they are None or empty if not not_after_str or not not_before_str: if app_config.VERBOSE: - print("Cert Score: CRITICAL - Certificate date fields missing or invalid. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: CRITICAL - Certificate date fields missing or invalid. (CERT_HEALTH)") scores['Certificate_Health'] -= 9 return - # --- MODIFIED: Remove .split('.') since the new format ends in 'Z' (e.g., 2024-11-20T14:00:00Z) --- # The 'Z' indicates UTC and is handled directly by fromisoformat. not_after = datetime.fromisoformat(not_after_str.replace('Z', '+00:00')) not_before = datetime.fromisoformat(not_before_str.replace('Z', '+00:00')) @@ -52,7 +47,7 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha except (ValueError, TypeError) as e: # Handles errors from fromisoformat if the string is malformed if app_config.VERBOSE: - print(f"Cert Score: CRITICAL - Certificate date fields are malformed or missing (Error: {e}). (CERT_HEALTH)", file=sys.stderr) + print(f"Cert Score: CRITICAL - Certificate date fields are malformed or missing (Error: {e}). (CERT_HEALTH)") scores['Certificate_Health'] -= 8 return @@ -60,12 +55,12 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha if scan_date.replace(tzinfo=not_after.tzinfo) > not_after: # Expired if app_config.VERBOSE: - print("Cert Score: CRITICAL - Certificate has expired. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: CRITICAL - Certificate has expired. (CERT_HEALTH)") scores['Certificate_Health'] -= 50 if scan_date.replace(tzinfo=not_before.tzinfo) < not_before: # Not yet valid if app_config.VERBOSE: - print("Cert Score: CRITICAL - Certificate not yet valid. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: CRITICAL - Certificate not yet valid. (CERT_HEALTH)") scores['Certificate_Health'] -= 50 # 2. Expiration Time Check (Gradient and Buckets) @@ -76,7 +71,7 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha if days_until_expiration > 30: # No deduction for >30 days if app_config.VERBOSE: - print(f"Cert Score: Standard Warning - Expires in {days_until_expiration} days.", file=sys.stderr) + print(f"Cert Score: Standard Warning - Expires in {days_until_expiration} days.") else: # 1 <= days_until_expiration <= 30 # High-Risk Gradient: Deduction scales from 0 at 30 days to 30 at 0 days. MAX_GRADIENT_DEDUCTION = 30 @@ -87,7 +82,7 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha scores['Certificate_Health'] -= deduction if app_config.VERBOSE: - print(f"Cert Score: High Risk Gradient - Expires in {days_until_expiration} days. Deduction: -{deduction} (CERT_HEALTH)", file=sys.stderr) + print(f"Cert Score: High Risk Gradient - Expires in {days_until_expiration} days. Deduction: -{deduction} (CERT_HEALTH)") # 3. Check Verification Status hostname_matches = verification_data.get("hostname_matches", False) @@ -96,20 +91,19 @@ def score_cert_health(data: dict, scan_date: datetime, scores: dict): #TODO: Cha if not hostname_matches: scores['Certificate_Health'] -= 10 if app_config.VERBOSE: - print("Cert Score: Significant Deduction - Hostname does not match certificate. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: Significant Deduction - Hostname does not match certificate. (CERT_HEALTH)") if not chain_verified: scores['Certificate_Health'] -= 10 if app_config.VERBOSE: - print("Cert Score: Significant Deduction - Certificate chain not verified. (CERT_HEALTH)", file=sys.stderr) + print("Cert Score: Significant Deduction - Certificate chain not verified. (CERT_HEALTH)") -def score_dns_rec_health(dns_data: dict, rdap_scan:dict, scores: dict): +def score_dns_rec_health(dns_data: dict, scores: dict): """Calculates the score for the DNS Scan (Max Score: 100). Focuses on record coverage (rcode) and redundancy (A/AAAA counts). """ rcode = dns_data.get("rcode", 0) a_count = len(dns_data.get("a", [])) aaaa_count = len(dns_data.get("aaaa", [])) - cname = dns_data.get("cname", []) # --- 2. RCODE Completeness Check (New Banded Scoring) --- # Goal: Ensure a wide set of requested record types are returned. @@ -121,12 +115,12 @@ def score_dns_rec_health(dns_data: dict, rdap_scan:dict, scores: dict): # Missing several key types (e.g., TXT/MX if NS is present) scores['DNS_Record_Health'] -= 10 if app_config.VERBOSE: - print(f"DNS Score: Minor Deduction - rcode {rcode} is incomplete (Missing advanced types). (DNS_REC_HEALTH)", file=sys.stderr) + print(f"DNS Score: Minor Deduction - rcode {rcode} is incomplete (Missing advanced types). (DNS_REC_HEALTH)") elif rcode >= 1: # 1 <= rcode <= 7 # Missing foundational types (e.g., NS) scores['DNS_Record_Health'] -= 15 if app_config.VERBOSE: - print(f"DNS Score: Significant Deduction - rcode {rcode} is low (Missing foundational types). (DNS_REC_HEALTH)", file=sys.stderr) + print(f"DNS Score: Significant Deduction - rcode {rcode} is low (Missing foundational types). (DNS_REC_HEALTH)") # 2. Redundancy Check @@ -134,19 +128,18 @@ def score_dns_rec_health(dns_data: dict, rdap_scan:dict, scores: dict): if a_count < 2: scores['DNS_Record_Health'] -= 10 if app_config.VERBOSE: - print("DNS Score: Minor Deduction - Only one IPv4 address (SPOF). (DNS_REC_HEALTH)", file=sys.stderr) + print("DNS Score: Minor Deduction - Only one IPv4 address (SPOF). (DNS_REC_HEALTH)") # IPv6 Redundancy if aaaa_count == 0: scores['DNS_Record_Health'] -= 5 if app_config.VERBOSE: - print("DNS Score: Minor Deduction - No IPv6 support. (DNS_REC_HEALTH)", file=sys.stderr) + print("DNS Score: Minor Deduction - No IPv6 support. (DNS_REC_HEALTH)") elif aaaa_count < 2: scores['DNS_Record_Health'] -= 5 if app_config.VERBOSE: - print("DNS Score: Minor Deduction - Only one IPv6 address (SPOF). (DNS_REC_HEALTH)", file=sys.stderr) + print("DNS Score: Minor Deduction - Only one IPv6 address (SPOF). (DNS_REC_HEALTH)") - # TODO: look into how to check CNAME and nameserver info def score_conn_sec(hval_data: dict, cert_data: dict, scores: dict): """Calculates the score for the HVAL Scan (Max Score: 100). @@ -165,17 +158,17 @@ def score_conn_sec(hval_data: dict, cert_data: dict, scores: dict): # CHECK - correct functionality for desired outcome? if final_status == 403: if app_config.VERBOSE: - print("HVAL Notice: Final connection returned 403 Forbidden. Skipping HTTPS enforcement check. (CONN_SEC)", file=sys.stderr) + print("HVAL Notice: Final connection returned 403 Forbidden. Skipping HTTPS enforcement check. (CONN_SEC)") pass elif final_status == -1: if app_config.VERBOSE: - print("HVAL Score: CRITICAL - No response from server (connection failed). (CONN_SEC)", file=sys.stderr) + print("HVAL Score: CRITICAL - No response from server (connection failed). (CONN_SEC)") scores['Connection_Security'] -= 10 # return 1 elif not (200 <= final_status < 207) or not final_url.startswith("https"): # Fails to load or loads over HTTP if app_config.VERBOSE: - print("HVAL Score: CRITICAL - Final connection not 200 HTTPS. (CONN_SEC)", file=sys.stderr) + print("HVAL Score: CRITICAL - Final connection not 200 HTTPS. (CONN_SEC)") scores['Connection_Security'] -= 45 # return 1 @@ -185,14 +178,13 @@ def score_conn_sec(hval_data: dict, cert_data: dict, scores: dict): elif "TLS_ECDHE_RSA" in tls_cipher: scores['Connection_Security'] -= 10 if app_config.VERBOSE: - print(f"HVAL Score: Minor Deduction - Moderate cipher used: {tls_cipher}. (CONN_SEC)", file=sys.stderr) + print(f"HVAL Score: Minor Deduction - Moderate cipher used: {tls_cipher}. (CONN_SEC)") else: scores['Connection_Security'] -= 45 if app_config.VERBOSE: - print(f"HVAL Score: Significant Deduction - Weak/no cipher used: {tls_cipher}. (CONN_SEC)", file=sys.stderr) + print(f"HVAL Score: Significant Deduction - Weak/no cipher used: {tls_cipher}. (CONN_SEC)") # 3. Security Header Check (Bitwise Flag Analysis) - #TODO: Create functionality if security flag is missing (skip this step?) # The required headers are HSTS (1), CSP (2), XCTO (4). Total = 7. REQUIRED_FLAGS = app_config.SECURITY_FLAGS['HSTS'] | app_config.SECURITY_FLAGS['CSP'] | app_config.SECURITY_FLAGS['XCTO'] @@ -209,30 +201,30 @@ def score_conn_sec(hval_data: dict, cert_data: dict, scores: dict): missing_count = is_hsts_missing + is_csp_missing + is_xcto_missing if missing_count == 0: - # HSTS, CSP, and XCTO present: +0 (or a small bonus) + # HSTS, CSP, and XCTO present: +0 if app_config.VERBOSE: - print("HVAL Score: HSTS, CSP, XCTO all present.", file=sys.stderr) + print("HVAL Score: HSTS, CSP, XCTO all present.") elif missing_count == 1: # Missing one of the three: -20 scores['Connection_Security'] -= 20 if app_config.VERBOSE: - print(f"HVAL Score: Deduction - Missing 1 critical header (HSTS/CSP/XCTO). -20 pts. (CONN_SEC)", file=sys.stderr) + print(f"HVAL Score: Deduction - Missing 1 critical header (HSTS/CSP/XCTO). -20 pts. (CONN_SEC)") elif missing_count >= 2: # Missing two or more of the three: -40 scores['Connection_Security'] -= 40 if app_config.VERBOSE: - print(f"HVAL Score: Major Deduction - Missing {missing_count} critical headers. -40 pts. (CONN_SEC)", file=sys.stderr) + print(f"HVAL Score: Major Deduction - Missing {missing_count} critical headers. -40 pts. (CONN_SEC)") # Check Dangerous/Advanced Headers (Minor Deductions) advanced_flags = app_config.SECURITY_FLAGS['COOP'] | app_config.SECURITY_FLAGS['CORP'] | app_config.SECURITY_FLAGS['COEP'] if (security_flag & advanced_flags) != advanced_flags: scores['Connection_Security'] -= 5 # Minor deduction for incomplete advanced security. if app_config.VERBOSE: - print("HVAL Score: Minor Deduction - Missing one or more advanced security headers (COOP/CORP/COEP). (CONN_SEC)", file=sys.stderr) + print("HVAL Score: Minor Deduction - Missing one or more advanced security headers (COOP/CORP/COEP). (CONN_SEC)") if tls_version not in ['TLS 1.2', 'TLS 1.3']: scores['Connection_Security'] -= 20 if app_config.VERBOSE: - print(f"HVAL Score: Significant Deduction - Outdated TLS version: {tls_version}. (CONN_SEC)", file=sys.stderr) + print(f"HVAL Score: Significant Deduction - Outdated TLS version: {tls_version}. (CONN_SEC)") def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION """Unifies Domain Reputation scoring from Mail, Method, and RDAP scans.""" @@ -243,21 +235,21 @@ def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION if mx_count == 0: scores['Domain_Reputation'] -= 20 if app_config.VERBOSE: - print("Mail Score: CRITICAL - No MX records (cannot receive mail). (DOM_REP)", file=sys.stderr) + print("Mail Score: CRITICAL - No MX records (cannot receive mail). (DOM_REP)") elif mx_count < 2: scores['Domain_Reputation'] -= 5 if app_config.VERBOSE: - print("Mail Score: Significant Deduction - Only one MX record (SPOF). (DOM_REP)", file=sys.stderr) + print("Mail Score: Significant Deduction - Only one MX record (SPOF). (DOM_REP)") else: if app_config.VERBOSE: - print("Mail Score: MX redundancy is good.", file=sys.stderr) + print("Mail Score: MX redundancy is good.") # 2. DMARC Policy (Highest Impact) dmarc_data = mail_data.get("dmarc", []) if not dmarc_data: scores['Domain_Reputation'] -= 22 if app_config.VERBOSE: - print("Mail Score: Major Deduction - DMARC record is missing (high spoofing risk). (DOM_REP)", file=sys.stderr) + print("Mail Score: Major Deduction - DMARC record is missing (high spoofing risk). (DOM_REP)") else: # Parse the DMARC string (e.g., "v=DMARC1; p=reject;...") dmarc_policy = next((part.split('=')[1] for part in dmarc_data[0].split(';') if part.strip().startswith('p=')), 'none') @@ -265,21 +257,21 @@ def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION if dmarc_policy.strip() != 'reject' and dmarc_policy.strip() != 'quarantine': scores['Domain_Reputation'] -= 7 # Optimal is reject or quarantine if app_config.VERBOSE: - print(f"Mail Score: Significant Deduction - DMARC policy is '{dmarc_policy}' (no active enforcement). (DOM_REP)", file=sys.stderr) + print(f"Mail Score: Significant Deduction - DMARC policy is '{dmarc_policy}' (no active enforcement). (DOM_REP)") # Check Subdomain policy (sp=) sp_policy = next((part.split('=')[1] for part in dmarc_data[0].split(';') if part.strip().startswith('sp=')), dmarc_policy) if sp_policy.strip() != 'reject' and sp_policy.strip() != 'quarantine': scores['Domain_Reputation'] -= 2 # Optimal is reject or quarantine if app_config.VERBOSE: - print(f"Mail Score: Minor Deduction - DMARC subdomain policy is '{sp_policy}' (no active enforcement). (DOM_REP)", file=sys.stderr) + print(f"Mail Score: Minor Deduction - DMARC subdomain policy is '{sp_policy}' (no active enforcement). (DOM_REP)") # 3. SPF Policy spf_data = mail_data.get("spf", []) if not spf_data or not any("v=spf1" in s for s in spf_data): scores['Domain_Reputation'] -= 10 if app_config.VERBOSE: - print("Mail Score: Major Deduction - SPF record is missing. (DOM_REP)", file=sys.stderr) + print("Mail Score: Major Deduction - SPF record is missing. (DOM_REP)") else: # Extract the SPF mechanism (e.g., "~all" or "-all") spf_string = next(s for s in spf_data if "v=spf1" in s) @@ -288,11 +280,11 @@ def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION elif "~all" in spf_string: scores['Domain_Reputation'] -= 5 # SoftFail (like medium.com) if app_config.VERBOSE: - print("Mail Score: Minor Deduction - SPF policy is '~all' (SoftFail). (DOM_REP)", file=sys.stderr) + print("Mail Score: Minor Deduction - SPF policy is '~all' (SoftFail). (DOM_REP)") elif "?all" in spf_string or "+all" in spf_string: scores['Domain_Reputation'] -= 12 if app_config.VERBOSE: - print(f"Mail Score: Deduction - SPF policy is too permissive ('{spf_string[-4:]}'). (DOM_REP)", file=sys.stderr) + print(f"Mail Score: Deduction - SPF policy is too permissive ('{spf_string[-4:]}'). (DOM_REP)") # --- RDAP Scan --- nameservers = rdap_data[0].get("nameserver", []) @@ -301,11 +293,11 @@ def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION if len(nameservers) < 2: scores['Domain_Reputation'] -= 15 if app_config.VERBOSE: - print("RDAP Score: CRITICAL - Less than 2 nameservers (SPOF). (DOM_REP)", file=sys.stderr) + print("RDAP Score: CRITICAL - Less than 2 nameservers (SPOF). (DOM_REP)") elif len(nameservers) == 2: scores['Domain_Reputation'] -= 2 if app_config.VERBOSE: - print("RDAP Score: Deduction - Only 2 nameservers (limited redundancy). (DOM_REP)", file=sys.stderr) + print("RDAP Score: Deduction - Only 2 nameservers (limited redundancy). (DOM_REP)") elif len(nameservers) >= 3: pass # Good redundancy, no deduction @@ -321,28 +313,28 @@ def score_dom_rep(mail_data: dict, rdap_data: dict, scores: dict): #NEW FUNCTION # Example: both are *.cloudflare.com scores['Domain_Reputation'] -= 2 if app_config.VERBOSE: - print(f"RDAP Score: Minor Deduction - All nameservers on the same vendor ({list(providers)[0]}). (DOM_REP)", file=sys.stderr) + print(f"RDAP Score: Minor Deduction - All nameservers on the same vendor ({list(providers)[0]}). (DOM_REP)") elif len(providers) > 1: pass # Good diversity, no deduction #functionality to check tlds against malicious list target_tld = rdap_data[0].get("host", "").split('.')[-1] - print(f"Target TLD: {target_tld}", file=sys.stderr) + print(f"Target TLD: {target_tld}") if target_tld in app_config.MAL_TLDS_SLIM: scores['Domain_Reputation'] -= 10 if app_config.VERBOSE: - print(f"RDAP Score: Minor Deduction - TLD '{target_tld}' is associated with malicious websites. -10 (DOM_REP)", file=sys.stderr) + print(f"RDAP Score: Minor Deduction - TLD '{target_tld}' is associated with malicious websites. -10 (DOM_REP)") else: - pass # No deduction for TLD reputation + pass -def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): #TODO: IMPLEMENT +def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): """Scores Credential Safety of a site based on TLS, certificate, and crypto algorithms.""" cert_list = cert_data.get("certs", []) # Check if certificate list exists and is not empty if not cert_list or not isinstance(cert_list, list) or len(cert_list) == 0: if app_config.VERBOSE: - print("Credential Safety: CRITICAL - No certificates found in response. (CRED_SAFETY)", file=sys.stderr) + print("Credential Safety: CRITICAL - No certificates found in response. (CRED_SAFETY)") scores['Credential_Safety'] -= 50 return @@ -357,19 +349,19 @@ def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): #TODO: IMPLE if tls_version not in ['TLS 1.2', 'TLS 1.3']: scores['Credential_Safety'] -= 50 if app_config.VERBOSE: - print(f"Cred Safety Score: CRITICAL - Outdated TLS version: {tls_version}. (CRED_SAFETY)", file=sys.stderr) + print(f"Cred Safety Score: CRITICAL - Outdated TLS version: {tls_version}. (CRED_SAFETY)") # 2. HSTS Header if (sec_flag & app_config.SECURITY_FLAGS['HSTS']) == 0: scores['Credential_Safety'] -= 20 if app_config.VERBOSE: - print("Cred Safety Score: Significant Deduction - HSTS header missing. (CRED_SAFETY)", file=sys.stderr) + print("Cred Safety Score: Significant Deduction - HSTS header missing. (CRED_SAFETY)") # 3. Key Algorithm if key_algorithm not in ['ECDSA', 'RSA']: scores['Credential_Safety'] -= 20 if app_config.VERBOSE: - print("Cred Safety Score: Significant Deduction - Certificate key algorithm missing. (CRED_SAFETY)", file=sys.stderr) + print("Cred Safety Score: Significant Deduction - Certificate key algorithm missing. (CRED_SAFETY)") # 4. ECDSA Checks if (key_algorithm == "ECDSA"): @@ -377,11 +369,11 @@ def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): #TODO: IMPLE if (curve_name not in ['P-256', 'P-384', 'P-521']): scores['Credential_Safety'] -= 15 if app_config.VERBOSE: - print(f"Cred Safety Score: Significant Deduction - Outdated curve: {curve_name}. (CRED_SAFETY)", file=sys.stderr) + print(f"Cred Safety Score: Significant Deduction - Outdated curve: {curve_name}. (CRED_SAFETY)") if (sig_algorithm not in ['ECDSA-SHA256', 'ECDSA-SHA384', 'ECDSA-SHA512']): scores['Credential_Safety'] -= 5 if app_config.VERBOSE: - print(f"Cred Safety Score: Minor Deduction - Weak signature algorithm: {sig_algorithm}. (CRED_SAFETY)", file=sys.stderr) + print(f"Cred Safety Score: Minor Deduction - Weak signature algorithm: {sig_algorithm}. (CRED_SAFETY)") # 5. RSA Checks if (key_algorithm == "RSA"): @@ -389,67 +381,54 @@ def score_cred_safety(cert_data:dict, hval_data:dict, scores:dict): #TODO: IMPLE if (key_size not in [2048, 3072, 4096]): scores['Credential_Safety'] -= 15 if app_config.VERBOSE: - print(f"Cred Safety Score: Significant Deduction - Outdated key size: {key_size}. (CRED_SAFETY)", file=sys.stderr) + print(f"Cred Safety Score: Significant Deduction - Outdated key size: {key_size}. (CRED_SAFETY)") if (sig_algorithm not in ['SHA256-RSA', 'SHA384-RSA', 'SHA512-RSA']): scores['Credential_Safety'] -= 5 if app_config.VERBOSE: - print(f"Cred Safety Score: Minor Deduction - Weak signature algorithm: {sig_algorithm}. (CRED_SAFETY)", file=sys.stderr) + print(f"Cred Safety Score: Minor Deduction - Weak signature algorithm: {sig_algorithm}. (CRED_SAFETY)") -def score_ip_rep(firewall_data:dict, scores:dict): #PAUSED: Further investigation needed to determine if helpful - """Placeholder for IP Reputation scoring function. - Currently unused, but can be implemented in the future. - """ +def score_ip_rep(firewall_data:dict, scores:dict): + """IP Reputation scoring function. Checks if IP is on firewall list by FireHOL and scores accordingly.""" blocked = firewall_data.get("Block", False) if blocked: scores['IP_Reputation'] -= 99 if app_config.VERBOSE: - print("IP Reputation Score: CRITICAL - IP is listed on a firewall blocklist. (IP_REP)", file=sys.stderr) + print("IP Reputation Score: CRITICAL - IP is listed on a firewall blocklist. (IP_REP)") else: if app_config.VERBOSE: - print("IP Reputation Score: No deduction - IP is not listed on firewall blocklist. (IP_REP)", file=sys.stderr) + print("IP Reputation Score: No deduction - IP is not listed on firewall blocklist. (IP_REP)") -def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO: IMPLEMENT - """Placeholder for WHOIS Pattern scoring function. - Currently unused, but can be implemented in the future. - """ +def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): + """WHOIS Pattern scoring function. Checks for domain age, registrar reputation, and domain status patterns.""" domain_data = rdap_data[0].get('domain', {}) host = rdap_data[0].get("host","") nameservers = rdap_data[0].get("nameserver", []) events = domain_data.get('events', []) - status = domain_data.get('status', []) #place to check for "client delete prohibited", + status = domain_data.get('status', []) + #place to check for "client delete prohibited", # "client transfer prohibited", # "client update prohibited", # "server delete prohibited", # "server transfer prohibited", # "server update prohibited" and "add period" (means newly registered) - client_locks = [ - "client delete prohibited", - "client transfer prohibited", - "client update prohibited" - ] - server_locks = [ - "server delete prohibited", - "server transfer prohibited", - "server update prohibited" - ] if "add period" in status: scores['WHOIS_Pattern'] -= 30 if app_config.VERBOSE: - print("WHOIS Score: CRITICAL - Domain is newly registered (add period). (WHOIS_PATTERN)", file=sys.stderr) + print("WHOIS Score: CRITICAL - Domain is newly registered (add period). (WHOIS_PATTERN)") # Check for client and server locks - for lock in client_locks: + for lock in app_config.CLIENT_LOCKS: if lock not in status: scores['WHOIS_Pattern'] -= 5 if app_config.VERBOSE: - print(f"WHOIS Score: Minor Deduction - {lock} is not set. (WHOIS_PATTERN)", file=sys.stderr) + print(f"WHOIS Score: Minor Deduction - {lock} is not set. (WHOIS_PATTERN)") - for lock in server_locks: + for lock in app_config.SERVER_LOCKS: if lock not in status: scores['WHOIS_Pattern'] -= 5 if app_config.VERBOSE: - print(f"WHOIS Score: Minor Deduction - {lock} is not set. (WHOIS_PATTERN)", file=sys.stderr) + print(f"WHOIS Score: Minor Deduction - {lock} is not set. (WHOIS_PATTERN)") # Grabs the registration_date from the events list registration_date = None #If new, bad. If old, good. @@ -473,21 +452,21 @@ def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO if days_old < 30: # We use max(0, days_old) to handle cases where clock skew makes age negative if app_config.VERBOSE: - print(f"WHOIS Score: WARNING - Domain is very new ({max(0, days_old)} days old). (WHO_IS)", file=sys.stderr) + print(f"WHOIS Score: WARNING - Domain is very new ({max(0, days_old)} days old). (WHO_IS)") scores['WHOIS_Pattern'] -= 30 else: if app_config.VERBOSE: - print(f"WHOIS Score: INFO - Domain age is {days_old} days. (WHO_IS)", file=sys.stderr) + print(f"WHOIS Score: INFO - Domain age is {days_old} days. (WHO_IS)") except (ValueError, TypeError) as e: # Handles malformed RDAP date strings if app_config.VERBOSE: - print(f"WHOIS Score: ERROR - Registration date malformed: {e}. (WHO_IS)", file=sys.stderr) + print(f"WHOIS Score: ERROR - Registration date malformed: {e}. (WHO_IS)") scores['WHOIS_Pattern'] -= 5 else: # If no registration event was found in the RDAP data if app_config.VERBOSE: - print("WHOIS Score: WARNING - No registration date found. (WHO_IS)", file=sys.stderr) + print("WHOIS Score: WARNING - No registration date found. (WHO_IS)") scores['WHOIS_Pattern'] -= 10 # 1. Access the vcard list safely @@ -514,7 +493,7 @@ def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO registrar_name = extracted_fn if app_config.VERBOSE: - print("WHO_IS Score: Registrar name is ", registrar_name, file=sys.stderr) + print("WHO_IS Score: Registrar name is ", registrar_name) # 2. Check for registrary matches from MAL_REGISTRARS list (case-insensitive substring match) found_matches = False @@ -527,12 +506,12 @@ def score_whois_pattern(rdap_data:dict, scan_date: datetime, scores:dict): #TODO if found_matches: scores['WHOIS_Pattern'] -= 10 if app_config.VERBOSE: - print("WHO_IS Score: Suspicious Registrar found: ", registrar_name, file=sys.stderr) + print("WHO_IS Score: Suspicious Registrar found: ", registrar_name) else: if app_config.VERBOSE: - print("WHO_IS Score: Trustworthy Registrar found: ", registrar_name, file=sys.stderr) + print("WHO_IS Score: Trustworthy Registrar found: ", registrar_name) -def calculate_final_score(weights, scores): #CHANGE +def calculate_final_score(weights, scores): """ Calculates the final score using the Weighted Harmonic Mean formula: Final Score = (Sum of Weights) / (Sum of (Weight / Score)) @@ -546,7 +525,7 @@ def calculate_final_score(weights, scores): #CHANGE sum_of_ratios = 0.0 if app_config.VERBOSE: - print("\n--- Individual Component Ratios (Wi / Scorei) ---", file=sys.stderr) + print("\n--- Individual Component Ratios (Wi / Scorei) ---") # We only include components in the calculation *if* we have a score for them. for tool_name, score in scores.items(): @@ -558,7 +537,7 @@ def calculate_final_score(weights, scores): #CHANGE # If any score is zero or negative, the Harmonic Mean approaches zero. # We return 1 immediately as a zero score on a critical factor indicates failure. if app_config.VERBOSE: - print(f"CRITICAL ERROR: {tool_name} score is 0 or less. Returning Final Score of 1.", file=sys.stderr) + print(f"CRITICAL ERROR: {tool_name} score is 0 or less. Returning Final Score of 1.") return 1 # Calculate the ratio Wi / Scorei @@ -567,18 +546,18 @@ def calculate_final_score(weights, scores): #CHANGE # Display the components for clarity if app_config.VERBOSE: - print(f" {tool_name:15}: {weight} / {score:.2f} = {ratio:.4f}", file=sys.stderr) + print(f" {tool_name:15}: {weight} / {score:.2f} = {ratio:.4f}") if app_config.VERBOSE: - print("--------------------------------------------------", file=sys.stderr) - print(f"Sum of Weights (Numerator): {sum_of_weights}", file=sys.stderr) - print(f"Sum of Ratios (Denominator): {sum_of_ratios:.4f}", file=sys.stderr) + print("--------------------------------------------------") + print(f"Sum of Weights (Numerator): {sum_of_weights}") + print(f"Sum of Ratios (Denominator): {sum_of_ratios:.4f}") # 3. Calculate the Final Score if sum_of_ratios == 0: # This happens if no scores were provided. if app_config.VERBOSE: - print("No valid scores found. Cannot calculate score.", file=sys.stderr) + print("No valid scores found. Cannot calculate score.") return 0.0 final_score = sum_of_weights / sum_of_ratios @@ -586,7 +565,7 @@ def calculate_final_score(weights, scores): #CHANGE # --- Main Scoring Orchestrator --- -def calculate_security_score(all_scans: dict, scan_date: datetime) -> dict: #CHANGE +def calculate_security_score(all_scans: dict, scan_date: datetime) -> dict: """Runs all scoring functions and calculates the average security score.""" scores = { @@ -600,43 +579,43 @@ def calculate_security_score(all_scans: dict, scan_date: datetime) -> dict: #CHA } if app_config.VERBOSE: - print(f"\n--- Calculating Scores (Reference Date: {scan_date.strftime('%Y-%m-%d')}) ---", file=sys.stderr) + print(f"\n--- Calculating Scores (Reference Date: {scan_date.strftime('%Y-%m-%d')}) ---") # 2. Run each scan function, which will modify the scores dictionary try: score_cert_health(all_scans['cert_scan'], scan_date, scores) except Exception as e: - print(f"Error in cert_health scan: {e}", file=sys.stderr) + print(f"Error in cert_health scan: {e}") try: - score_dns_rec_health(all_scans['dns_scan'], all_scans['rdap_scan'], scores) + score_dns_rec_health(all_scans['dns_scan'], scores) except Exception as e: - print(f"Error in dns_rec_health scan: {e}", file=sys.stderr) + print(f"Error in dns_rec_health scan: {e}") try: score_conn_sec(all_scans['hval_scan'], all_scans['cert_scan'], scores) except Exception as e: - print(f"Error in conn_sec scan: {e}", file=sys.stderr) + print(f"Error in conn_sec scan: {e}") try: score_dom_rep(all_scans['mail_scan'], all_scans['rdap_scan'], scores) except Exception as e: - print(f"Error in dom_rep scan: {e}", file=sys.stderr) + print(f"Error in dom_rep scan: {e}") try: score_cred_safety(all_scans['cert_scan'], all_scans['hval_scan'], scores) except Exception as e: - print(f"Error in cred_safety scan: {e}", file=sys.stderr) + print(f"Error in cred_safety scan: {e}") try: score_whois_pattern(all_scans['rdap_scan'], scan_date, scores) except Exception as e: - print(f"Error in whois_pattern scan: {e}", file=sys.stderr) + print(f"Error in whois_pattern scan: {e}") try: score_ip_rep(all_scans['firewall_scan'], scores) except Exception as e: - print(f"Error in ip_rep scan: {e}", file=sys.stderr) + print(f"Error in ip_rep scan: {e}") # 3. Clamp scores between 1 and 100 after all deductions for key in scores: diff --git a/Scoring Engine/scoring_main.py b/Scoring Engine/scoring_main.py index 883b3c9..3529f29 100644 --- a/Scoring Engine/scoring_main.py +++ b/Scoring Engine/scoring_main.py @@ -112,19 +112,26 @@ ) # The -t/--target value is expected to be a pre-sanitized hostname # provided by the server (no scheme, no path, no "www." prefix). - # All normalization and validation happens in server.js before this - # script is invoked. See Docs/url-sanitization-policy.md. + # All normalization and validation happens before this + # script is invoked. parser.add_argument( '-t', '--target', type=str, default=app_config.DEFAULT_URL, help=f"The target hostname to scan (e.g., {app_config.DEFAULT_URL})" ) + # The -v/--verbose flag is for debugging and clarity when running the script manually. + # It will print out more information about the scoring process and what reasons the + # different categories got docked points for. parser.add_argument( '-v', '--verbose', action='store_true', help="Enable verbose output for clarity." ) + #The --use-test-data flag allows us to run the scoring logic on the internal + # test_scans data instead of making live API calls. This is useful for development + # and testing purposes, especially when we want consistent results without relying + # on external factors. parser.add_argument( '--use-test-data', action='store_true', @@ -145,23 +152,23 @@ # 2. Decide whether to use test data or fetch live data if args.use_test_data: if app_config.VERBOSE: - print(f"--- Running analysis on TEST DATA ---", file=sys.stderr) + print(f"--- Running analysis on TEST DATA ---") all_scans = test_scans # For reproducible results, we'll set a fixed date for the expiration checks. # Cert Sample Expiration: 2025-12-15. scan_date = datetime(2025, 10, 15) else: if app_config.VERBOSE: - print(f"--- Running analysis on LIVE DATA for {args.target} ---", file=sys.stderr) + print(f"--- Running analysis on LIVE DATA for {args.target} ---") # For live data, use the real date! scan_date = datetime.now() - # *** CHANGED TO THE CONCURRENT FETCH FUNCTION *** + # Pass the URL into the curl functions all_scans = fetch_scan_data_concurrent(args.target) # 3. Check if we have data, then calculate and print scores if not all_scans: if app_config.VERBOSE: - print("No scan data was retrieved. Exiting.", file=sys.stderr) + print("No scan data was retrieved. Exiting.") sys.exit(1) final_scores = calculate_security_score(all_scans, scan_date) @@ -178,7 +185,7 @@ print(json.dumps(output, indent=2)) if app_config.VERBOSE: - print("-------------------------------------------", file=sys.stderr) - print(f"Total execution time: {elapsed_time:.2f} seconds", file=sys.stderr) - print("-------------------------------------------", file=sys.stderr) + print("-------------------------------------------") + print(f"Total execution time: {elapsed_time:.2f} seconds") + print("-------------------------------------------")