Skip to content

Beautify and code cleanup #81

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
aCaboose merged 6 commits into from
Mar 18, 2026
Merged

Beautify and code cleanup #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
dc2cf39
Firewall endpoint added and IP_REP implemented
cgedris Feb 18, 2026
f511310
Merge branch 'main' into engine_work
aCaboose Mar 4, 2026
2435e3c
WHOIS: registrar scoring
cgedris Mar 5, 2026
7ababdc
remove method scan
cgedris Mar 11, 2026
8756c02
Beautified Code - comments and cleanup
cgedris Mar 17, 2026
f788b07
Merge branch 'main' into engine_work
aCaboose Mar 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions Scoring Engine/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand All @@ -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",
Expand All @@ -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",
Expand All @@ -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",
Expand All @@ -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"
]
24 changes: 12 additions & 12 deletions Scoring Engine/data_fetch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -28,24 +28,24 @@ 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)
return result.stdout.strip()

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]:
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Loading
Loading