mirror of
http://git.nowherejezfoltodf4jiyl6r56jnzintap5vyjlia7fkirfsnfizflqd.onion/nihilist/darknet-lantern.git
synced 2025-07-01 18:56:40 +00:00
271 lines
No EOL
7.2 KiB
Python
271 lines
No EOL
7.2 KiB
Python
import random
|
|
import re
|
|
import os
|
|
import requests
|
|
from PIL import Image
|
|
import json
|
|
#from SimpleX.utils import IsUrlValid
|
|
import urllib.parse
|
|
from websockets.sync.client import connect
|
|
|
|
|
|
PURPLE = '\033[35;40m'
|
|
BOLD_PURPLE = '\033[35;40;1m'
|
|
RED = '\033[31;40m'
|
|
BOLD_RED = '\033[31;40;1m'
|
|
RESET = '\033[m'
|
|
|
|
# name should contain only up to 64 alphanumeric characters
|
|
VALID_NAME_PATTERN = re.compile(r"^[A-Za-z0-9]{1,64}$")
|
|
|
|
# pattern for regular urls (https://stackoverflow.com/a/3809435)
|
|
CLEARNET_URL_PATTERN = re.compile(
|
|
r"https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]"
|
|
r"{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*)"
|
|
)
|
|
|
|
# pattern for onion urls (56 bytes of base32 alphabet + .onion)
|
|
ONION_URL_PATTERN = re.compile(
|
|
r"^https?:\/\/([a-zA-Z0-9-]+\.)*[a-z2-7-]{56}\.onion[^\s]*$"
|
|
)
|
|
|
|
# pattern for simplex chatroom links
|
|
SIMPLEX_CHATROOM_PATTERN = re.compile(
|
|
r"(?:https?:\/\/(?:simplex\.chat|[^\/]+)|simplex:)\/(?:contact|invitation)#\/\?v=[\d-]+"
|
|
r"&smp=[^&]+(?:&[^=]+=[^&]*)*(?:&data=\{[^}]*\})?"
|
|
)
|
|
|
|
# pattern for smp or xftp simplex server ((smp|xftp):// 44 byte key @ url [:port])
|
|
SIMPLEX_SERVER_PATTERN = re.compile(
|
|
r"^(smp|xftp):\/\/([a-zA-Z0-9\-_+=]{44})@([a-z2-7]{56}\.onion|"
|
|
r"([a-zA-Z0-9\-\.]+\.[a-zA-Z0-9\-\.]+))"
|
|
r"{1,}(?::[1-9][0-9]{0,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|"
|
|
r"65[0-4][0-9]{2}|655[0-3][0-9]|6553[0-5])?$"
|
|
)
|
|
|
|
def IsSimplexChatroomValid(url: str) -> bool:
|
|
"""
|
|
Recognizes Simplex Chatroom link.
|
|
Returns True if URL is a SimpleX chatroom,
|
|
False otherwise
|
|
"""
|
|
return SIMPLEX_CHATROOM_PATTERN.match(url)
|
|
|
|
def RecognizeSimplexType(url: str) -> str:
|
|
"""
|
|
Recognizes Simplex Server URL, returns smp, xftp or invalid
|
|
"""
|
|
match = SIMPLEX_SERVER_PATTERN.match(url)
|
|
if match:
|
|
return match.group(1)
|
|
else:
|
|
return 'invalid'
|
|
|
|
# stub function
|
|
def IsXFTPServerValid(url: str) -> bool:
|
|
"""
|
|
Returns True if URL is a valid SimpleX XFTP Server URL
|
|
False otherwise
|
|
"""
|
|
return RecognizeSimplexType(url) == 'xftp'
|
|
|
|
# stub function
|
|
def IsSMPServerValid(url: str) -> bool:
|
|
"""
|
|
Returns True if URL is a valid SimpleX SMP Server URL
|
|
False otherwise
|
|
"""
|
|
return RecognizeSimplexType(url) == 'smp'
|
|
|
|
def IsClearnetLinkValid(url: str) -> bool:
|
|
"""
|
|
Returns True if URL is a valid clearnet URL
|
|
False otherwise
|
|
"""
|
|
return CLEARNET_URL_PATTERN.match(url)
|
|
|
|
def IsOnionLinkValid(url: str) -> bool:
|
|
"""
|
|
Returns True if URL is a valid onion URL
|
|
False otherwise
|
|
"""
|
|
return ONION_URL_PATTERN.match(url)
|
|
|
|
def RecognizeURLType(url: str) -> str:
|
|
"""
|
|
Recognizes URL type, can return:
|
|
- chatroom - SimpleX chatroom
|
|
- xftp - XFTP SimpleX server
|
|
- smp - SMP SimpleX server
|
|
- onion - onion URL
|
|
- clearnet - valid clearnet url
|
|
- invalid - none of the above (probably invalid)
|
|
"""
|
|
# order is important here
|
|
# (ex. simplex chatroom is also valid clearnet link)
|
|
if IsSimplexChatroomValid(url):
|
|
return 'chatroom'
|
|
if IsXFTPServerValid(url):
|
|
return 'xftp'
|
|
if IsSMPServerValid(url):
|
|
return 'smp'
|
|
if IsOnionLinkValid(url):
|
|
return 'onion'
|
|
if IsClearnetLinkValid(url):
|
|
return 'clearnet'
|
|
return 'invalid'
|
|
|
|
def IsURLValid(url: str) -> bool:
|
|
"""
|
|
Checks if given URL is valid (RecognizeURLType recognizes it)
|
|
"""
|
|
return RecognizeURLType(url) != 'invalid'
|
|
|
|
|
|
#### Checking Functions to validate that links are legit ####
|
|
|
|
def CheckUrl(url):
|
|
"""
|
|
Checks if URL is actually reachable via Tor
|
|
"""
|
|
proxies = {
|
|
'http': 'socks5h://127.0.0.1:9050',
|
|
'https': 'socks5h://127.0.0.1:9050'
|
|
}
|
|
try:
|
|
status = requests.get(url, proxies=proxies, timeout=5).status_code
|
|
return status == 200
|
|
except requests.ConnectionError:
|
|
return False
|
|
except requests.exceptions.ReadTimeout:
|
|
return False
|
|
|
|
|
|
#### PROTECTIONS AGAINST MALICIOUS CSV INPUTS ####
|
|
def IsBannerValid(path: str) -> bool:
|
|
"""
|
|
Checks if the banner.png file has the correct dimensions (240x60)
|
|
"""
|
|
try:
|
|
im = Image.open(path)
|
|
except Exception:
|
|
print("ERROR, EXCEPTION")
|
|
return False
|
|
width, height = im.size
|
|
if width != 240 or height != 60:
|
|
print("INVALID BANNER DIMENSIONS, HEIGHT=", height, " WIDTH=", width)
|
|
return False
|
|
filesizeMB = os.path.getsize(path)/1024/1024
|
|
if filesizeMB > 5:
|
|
print("Banner filesize too large (>5Mb): ",os.path.getsize(path)/1024/1024,"MB")
|
|
return False
|
|
return True
|
|
|
|
|
|
|
|
|
|
def IsStatusValid(status: str) -> bool:
|
|
"""
|
|
Checks if status contains only ['YES','NO']. Verbose only if False is returned
|
|
"""
|
|
pattern = ['YES','NO','']
|
|
status = status.strip()
|
|
if status not in pattern:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def IsScoreValid(score: str) -> bool:
|
|
"""
|
|
Check the Score is only "^[0-9.,]+$" with 8 max chars.
|
|
"""
|
|
pattern = re.compile("^[0-9.,]+$")
|
|
score = str(score)
|
|
score.strip()
|
|
if score in ['','nan']:
|
|
return True
|
|
if pattern.fullmatch(score) is None:
|
|
return False
|
|
if len(score) > 8:
|
|
return False
|
|
return True
|
|
|
|
|
|
def IsDescriptionValid(desc: str) -> bool:
|
|
"""
|
|
Check the categories are only [a-zA-Z0-9.' ] with 256 max chars.
|
|
"""
|
|
if desc == "":
|
|
return True
|
|
pattern = re.compile(r"^[A-Za-z0-9-.,' \"\(\)\/]+$")
|
|
desc = str(desc)
|
|
desc.strip()
|
|
if pattern.fullmatch(desc) is None:
|
|
return False
|
|
if desc == "DEFAULT":
|
|
return False
|
|
elif len(desc) > 256:
|
|
return False
|
|
return True
|
|
|
|
|
|
def IsCategoryValid(categories: list[str]) -> bool:
|
|
"""
|
|
Check the categories are only [a-zA-Z0-9 ] with 64 max chars.
|
|
"""
|
|
pattern = re.compile("^[A-Za-z0-9 ]+$")
|
|
for category in categories:
|
|
category.strip()
|
|
if pattern.fullmatch(category) is None:
|
|
return False
|
|
elif len(category) > 64:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def IsNameValid(name: str) -> bool:
|
|
"""
|
|
Check the parameter name only contains [a-zA-Z0-9] and is 64 chars long.
|
|
"""
|
|
try:
|
|
return bool(VALID_NAME_PATTERN.fullmatch(name.strip()))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def send_server_checks(url: str) -> tuple[str, str, str]:
|
|
"""
|
|
Sends requests to sxc websocket and retuns
|
|
response, response type and testFailure or None.
|
|
"""
|
|
with connect(f"ws://localhost:3030") as websocket:
|
|
query = f"/_server test 1 {url}"
|
|
command = {
|
|
'corrId': f"id{random.randint(0,999999)}",
|
|
'cmd': query,
|
|
}
|
|
websocket.send(json.dumps(command))
|
|
message = websocket.recv()
|
|
response = json.loads(message)
|
|
resp_type = response["resp"]["type"]
|
|
failed_response = response['resp'].get('testFailure')
|
|
|
|
return (response, resp_type, failed_response)
|
|
|
|
|
|
def print_colors(s:str=' ', bold:bool=False, is_error:bool = False, default:bool=False):
|
|
"""
|
|
Helper function to print with colors
|
|
"""
|
|
if is_error:
|
|
print(f"{RED}{s}{RESET}")
|
|
elif bold:
|
|
print(f"{BOLD_PURPLE}{s}{RESET}")
|
|
elif is_error and bold:
|
|
print(f"{BOLD_RED}{s}{RESET}")
|
|
elif default:
|
|
print(f'{s}')
|
|
else:
|
|
print(f"{PURPLE}{s}{RESET}") |