implemented classes to remove global file

modules/Classes/Config.py

@@ -1,15 +1,75 @@
 import json
 
+from discord import Webhook, RequestsWebhookAdapter
+
+from modules.Classes.Driver import Driver
+from modules.Classes.Proxy import Proxy
 from modules.Classes.UserCredentials import UserCredentials
+from modules.Classes.DiscordConfig import DiscordConfig, FakeWebHook
+from modules.Classes.WordList import WordList
 
 
 class Config:
     def __init__(self, args):
-        self.args = args
-        self.UserCredentials = UserCredentials()
+        """
+        open config file
+        """
+        with open("/app/MsRewards-Reborn/user_data/discord.json", "r") as inFile:
+            discord = json.load(inFile)
+        with open("/app/MsRewards-Reborn/user_data/settings.json", "r") as inFile:
+            settings = json.load(inFile)
+        with open("/app/MsRewards-Reborn/user_data/proxy.json", "r") as inFile:
+            proxy = json.load(inFile)
+        with open("/app/MsRewards-Reborn/user_data/configs.json", "r") as inFile:
+            config = json.load(inFile)
 
+        """
+        setup standalone stuff
+        """
+        self.args = args
+        self.start = "json" if args.json else "default"
+        self.json_entry = args.json.replace("'", "\"")
+        self.wordlist = WordList("/usr/share/dict/french")
+        self.vnc = args.vnc
+        self.version = args.update_version
+        self.WebDriver = Driver()
+        self.display = None
+
+        """
+        setup UserCredential
+        """
+        self.UserCredentials = UserCredentials()
         with open("/app/MsRewards-Reborn/user_data/configs.json", "r") as inFile:
             configs = json.load(inFile)
             for i in configs[str(args.config)]["accounts"]:
                 d = configs[str(args.config)]["accounts"][i]
                 self.UserCredentials.add(d["mail"], d["pwd"], d["2fa"])
+
+        """
+        Setup discord
+        """
+        self.discord = DiscordConfig()
+        self.discord.avatar_url = settings["avatar_url"]
+        self.discord.wh_link = discord[config[args.config]["discord"]]["errorsL"]
+
+        if self.discord.wh_link != "":
+            self.discord.wh = Webhook.from_url(self.discord.wh_link, adapter=RequestsWebhookAdapter())
+        else:
+            self.discord.wh = FakeWebHook()
+
+        """
+        setup proxy
+        """
+        proxy_conf = config[args.config]["proxy"]
+        proxy_address = proxy[config[args.config]["proxy"]]["address"]
+        proxy_port = proxy[config[args.config]["proxy"]]["port"]
+        self.proxy = Proxy(proxy_conf, proxy_address, proxy_port)
+
+    def vnc_enabled(self):
+        return self.vnc != "None"
+
+    def set_display(self, display):
+        self.display = display
+
+    def has_been_updated(self):
+        return self.version != "None"

modules/Classes/DiscordConfig.py

@@ -0,0 +1,14 @@
+from modules.Tools.logger import debug
+
+
+class DiscordConfig:
+    def __init__(self):
+        self.avatar_url = ""
+        self.wh_link = None
+        self.wh = None
+
+
+class FakeWebHook:
+    def send(self, *args):
+        debug(f"Used a webhook call without webhook url with {args}")
+        

modules/Classes/DiscordLogger.py

@@ -0,0 +1,38 @@
+from discord import Embed, Colour, File
+
+from modules.Classes.Config import Config
+from modules.Classes.UserCredentials import UserCredentials
+from modules.Tools.logger import info, warning, error, critical
+from modules.Tools.tools import format_error
+
+
+class DiscordLogger:
+    def __init__(self, config: Config):
+        self.config = config
+
+    def send(self, message: str):
+        driver = self.config.WebDriver.driver
+        if type(message) is not str:
+            message = format_error(message)
+        error(message)
+        with open("page.html", "w") as f:
+            try:
+                f.write(driver.page_source)
+            except Exception as e:
+                error(e)
+                f.write("the driver has closed or crashed. Can't access page content")
+        img = self.config.display.waitgrab()
+        img.save("screenshot.png")
+
+        embed = Embed(
+            title="An Error has occured",
+            description=str(message),
+            colour=Colour.red(),
+        )
+        file = File("screenshot.png")
+        embed.set_image(url="attachment://screenshot.png")
+        embed.set_footer(text=self.config.UserCredentials.creds.get_mail())
+
+        self.config.discord.wh.send(embed=embed, username="error", file=file, avatar_url=self.config.discord.avatar_url)
+        self.config.discord.wh.send(username="error", file=File("page.html"), avatar_url=self.config.discord.avatar_url)
+

modules/Classes/Driver.py

@@ -0,0 +1,22 @@
+class Driver:
+    def __init__(self):
+        self.pc_driver = None
+        self.mobile_driver = None
+        self.driver = None
+
+    def set_pc_driver(self, pc_driver):
+        self.pc_driver = pc_driver
+
+    def set_mobile_driver(self, mobile_driver):
+        self.mobile_driver = mobile_driver
+
+    def switch_to_driver(self, driver: str):
+        match driver:
+            case "pc" | "PC" | "Pc":
+                self.driver = self.pc_driver
+
+            case "mobile" | "Mobile":
+                self.driver = self.mobile_driver
+
+            case _:
+                raise ValueError("The driver must be either pc or mobile")

modules/Classes/Proxy.py

@@ -0,0 +1,9 @@
+class Proxy:
+    def __init__(self, enabled: str, ip: str = None, port: str = None):
+        self.ip = ip
+        self.port = port
+        self.enabled = enabled != "-1"
+
+    def is_enabled(self):
+        return self.enabled
+    

modules/Classes/UserCredentials.py

@@ -1,4 +1,7 @@
 import json
+
+from pyotp import TOTP
+
 from modules.Tools.logger import debug, warning
 
 
@@ -29,7 +32,7 @@ class UserCredentials:
     def get_tfa(self):
         if not self.tfa_enable():
             warning("Warning: TFA is not enabled. Calling get_tfa is an expected behaviour.")
-        return self.data[self.current]["tfa"]
+        return TOTP(self.data[self.current]["tfa"])
 
     def next_account(self):
         self.current += 1

modules/Classes/WordList.py

@@ -0,0 +1,13 @@
+import random
+
+
+class WordList:
+    def __init__(self, path):
+        with open(path, "r", encoding="utf-8") as h:
+            lines = h.readlines()
+            self.words = [x.replace('\n', "") for x in lines]
+
+        random.shuffle(self.words)
+
+    def get_word(self):
+        return self.words.pop(0)

modules/Tools/tools.py

@@ -0,0 +1,32 @@
+from time import sleep
+
+
+# return current page domain
+def get_domain(driver):
+    return driver.current_url.split("/")[2]
+
+
+def custom_sleep(temps):
+    try:
+        if True:  # todo: change this awful condition
+            points = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"]
+            passe = 0
+            for i in range(int(temps)):
+                for _ in range(8):
+                    sleep(0.125)
+                    passe += 0.125
+                    print(f"{points[i]}  -  {round(float(temps) - passe, 3)}", end="\r")
+            print("                        ", end="\r")
+        else:
+            sleep(temps)
+    except KeyboardInterrupt:
+        print("attente annulée")
+
+
+def format_error(e) -> str:
+    tb = e.__traceback__
+    txt = ""
+    while tb is not None:
+        txt = txt + f" -> {tb.tb_frame.f_code.co_name} ({tb.tb_lineno}) "
+        tb = tb.tb_next
+    return txt + "\n" + str(e)

modules/config.py

@@ -1,14 +1,5 @@
 #!/usr/bin/python3.10
-from modules.driver_tools import *
 from modules.imports import *
-import modules.globals as g
-import json
-
-
-class FakeWebHook:
-    def send(self, text="", username='', avatar_url='', embed="", file=""):
-        print(text)
-
 
 parser = argparse.ArgumentParser()
 
@@ -39,69 +30,4 @@ parser.add_argument(
     default=""
 )
 
-with open("/app/MsRewards-Reborn/user_data/discord.json", "r") as inFile:
-    discord = json.load(inFile)
-with open("/app/MsRewards-Reborn/user_data/settings.json", "r") as inFile:
-    settings = json.load(inFile)
-with open("/app/MsRewards-Reborn/user_data/proxy.json", "r") as inFile:
-    proxy = json.load(inFile)
-with open("/app/MsRewards-Reborn/user_data/configs.json", "r") as inFile:
-    config = json.load(inFile)
-
 args = parser.parse_args()
-
-g.json_start = args.json
-
-g.vnc_enabled = args.vnc != "None"
-g.vnc_port = args.vnc
-g.update_version = args.update_version
-# global variables used later in the code
-
-g.start_time = time()
-
-# path configurations
-g.mot_path = "/usr/share/dict/french"
-g.credential_path = "/app/MsRewards-Reborn/user_data/login.csv"
-
-discord_conf = config[args.config]["discord"]
-
-# discord configuration
-g.discord_success_link = discord[discord_conf]["successL"]
-g.discord_error_link = discord[discord_conf]["errorsL"]
-g.discord_enabled_error = discord[discord_conf]["errorsT"] == "True"
-g.discord_enabled_success = discord[discord_conf]["successT"] == "True"
-
-g.avatar_url = settings["avatarlink"]
-
-if not g.json_start:
-    if g.discord_enabled_error:
-        webhookFailure = Webhook.from_url(g.discord_error_link, adapter=RequestsWebhookAdapter())
-    if g.discord_enabled_success:
-        webhookSuccess = Webhook.from_url(g.discord_success_link, adapter=RequestsWebhookAdapter())
-else:
-    webhookFailure = FakeWebHook()
-    webhookSuccess = FakeWebHook()
-
-
-# base settings
-g.discord_embed = False  # send new point value in an embed, fixed for now
-g.headless = False
-
-# proxy settings
-g.proxy_enabled = config[args.config]["proxy"] != "-1"
-if g.proxy_enabled:
-    g.proxy_address = proxy[config[args.config]["proxy"]]["address"]
-    g.proxy_port = proxy[config[args.config]["proxy"]]["port"]
-
-
-# list of words
-with open(g.mot_path, "r", encoding="utf-8") as h:
-    lines = h.readlines()
-    if len(lines) < 3:
-        Liste_de_mot = list(lines[0].split(","))
-    else:
-        Liste_de_mot = [x.replace('\n', "") for x in lines]
-
-
-if g.proxy_enabled:
-    setup_proxy(g.proxy_address, g.proxy_port)

modules/db.py

@@ -1,67 +1,66 @@
 import sqlite3
 
-#Create a new row, for the account [compte] whith [points] points
-def add_row(compte, points, mycursor, mydb):
+
+# Create a new row, for the account [compte] whith [points] points
+def add_row(account, points, mycursor, mydb):
     sql = "INSERT INTO daily (compte, points, date) VALUES (?, ?, date())"
-    val = (compte, points)
+    val = (account, points)
     mycursor.execute(sql, val)
     mydb.commit()
-    #printf(mycursor.rowcount, "record created.")
+    # printf(mycursor.rowcount, "record created.")
 
 
-#update the ammount of points for the account [compte] 
-def update_row(compte, points, mycursor, mydb):
-    sql = f"UPDATE daily SET points = {points} WHERE compte = '{compte}' AND date = date() ;"
+# update the ammount of points for the account [compte]
+def update_row(account, points, mycursor, mydb):
+    sql = f"UPDATE daily SET points = {points} WHERE compte = '{account}' AND date = date() ;"
     mycursor.execute(sql)
     mydb.commit()
-    #printf(mycursor.rowcount, "record(s) updated")
+
 
 # update the value of last_pts for the table comptes
-def update_last(compte, points, mycursor, mydb):
-    sql1 = f"UPDATE comptes SET last_pts = {points} WHERE compte = '{compte}';"
-    sql2 = f"select * from comptes where compte = '{compte}'"
-    sql3 = f"INSERT INTO comptes (compte, last_pts,banned) VALUES ('{compte}', {points}, 0)"
+def update_last(account, points, mycursor, mydb):
+    sql1 = f"UPDATE comptes SET last_pts = {points} WHERE compte = '{account}';"
+    sql2 = f"select * from comptes where compte = '{account}'"
+    sql3 = f"INSERT INTO comptes (compte, last_pts,banned) VALUES ('{account}', {points}, 0)"
     cmd = mycursor.execute(sql2)
     if len(list(cmd)) == 0:
         mycursor.execute(sql3)
-    else :
+    else:
         mycursor.execute(sql1)
     mydb.commit()
-    #printf(mycursor.rowcount, "record(s) updated")
 
-# if return if there already is a line in the database for the account [compte]. if same_point is enabled, the line must also have the same number of points
+
+# Return if there already is a line in the database for the account [account].
+# if same_point is enabled, the line must also have the same number of points
 # SQLITE
-def get_row(compte, points, mycursor, same_points = True): 
-    if same_points :
-        mycursor.execute(f"SELECT * FROM daily WHERE points = {points} AND compte = '{compte}' AND date = date() ;")
-    else :
-        mycursor.execute(f"SELECT * FROM daily WHERE compte = '{compte}' AND date = date() ;")
+def get_row(account, points, mycursor, same_points=True):
+    if same_points:
+        mycursor.execute(f"SELECT * FROM daily WHERE points = {points} AND compte = '{account}' AND date = date() ;")
+    else:
+        mycursor.execute(f"SELECT * FROM daily WHERE compte = '{account}' AND date = date() ;")
     myresult = mycursor.fetchall()
-    return(len(myresult) == 1)
+    return (len(myresult) == 1)
 
 
-def add_to_database(compte, points, save_if_fail=True):
+def add_to_database(account, points):
     if points is None:
         pass
     else:
         mydb = sqlite3.connect("/app/MsRewards-Reborn/MsRewards.db")
         mycursor = mydb.cursor()
-        if get_row(compte, points,mycursor, True): #check if the row exist with the same ammount of points and do nothind if it does
-            #printf("les points sont deja bon")
-            #return(0)
+
+        if get_row(account, points, mycursor, True):
+            # check if the row exist with the same amount of points and do nothing if it does
             pass
-        elif get_row(compte, points,mycursor, False) : #check if the row exist, but without the same ammount of points and update the point account then
-            update_row(compte, points,mycursor,mydb)
-            #printf("row updated")
-            #return(1)
-        else : # if the row don't exist, create it with the good ammount of points
-            add_row(compte, points,mycursor,mydb)
-            #return(2) #printf("row added")
-        if int(points) > 10 :
-            update_last(compte, points, mycursor, mydb)
+
+        # check if the row exist, but without the same amount of points and update the point account then
+        elif get_row(account, points, mycursor, False):
+            update_row(account, points, mycursor, mydb)
+
+        else:  # if the row don't exist, create it with the good amount of points
+            add_row(account, points, mycursor, mydb)
+
+        if int(points) > 10:
+            update_last(account, points, mycursor, mydb)
         mycursor.close()
         mydb.close()
-
-
-
-

modules/driver_tools.py

@@ -1,7 +1,12 @@
-from modules.imports import *
-from modules.config import *
-from modules.tools import *
-import modules.globals as g
+from random import uniform
+
+from selenium.common import TimeoutException
+from selenium.webdriver import ActionChains, Keys
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support import expected_conditions
+from selenium.webdriver.support.wait import WebDriverWait
+
+from modules.Tools.tools import *
 
 
 def set_language(ldriver):
@@ -16,10 +21,10 @@ def set_language(ldriver):
     # scroll down
     action.reset_actions()
     elm = ldriver.find_element(By.XPATH, "/html/body")
-    ActionChains(ldriver)\
-        .send_keys("french")\
-        .pause(0.5)\
-        .send_keys(Keys.TAB + Keys.TAB + Keys.ENTER + Keys.TAB + Keys.TAB + Keys.ENTER)\
+    ActionChains(ldriver) \
+        .send_keys("french") \
+        .pause(0.5) \
+        .send_keys(Keys.TAB + Keys.TAB + Keys.ENTER + Keys.TAB + Keys.TAB + Keys.ENTER) \
         .perform()
     x_coord = 1163
     y_coord = 717
@@ -33,64 +38,34 @@ def set_language(ldriver):
     action.click().perform()
 
 
-def setup_proxy(ip: str, port: str) -> None:
-    PROXY = f"{ip}:{port}"
-    webdriver.DesiredCapabilities.FIREFOX['proxy'] = {
-        "httpProxy": PROXY,
-        "sslProxy": PROXY,
-        "proxyType": "MANUAL",
-    }
-
-
-#Deal with RGPD popup as well as some random popup like 'are you satisfied' one
-def rgpd_popup(driver) -> None:
-    for i in ["bnp_btn_accept", "bnp_hfly_cta2", "bnp_hfly_close"] :
+# Deal with RGPD popup as well as some random popup like 'are you satisfied' one
+def rgpd_popup(config) -> None:
+    for i in ["bnp_btn_accept", "bnp_hfly_cta2", "bnp_hfly_close"]:
         try:
-            driver.find_element(By.ID, i).click()
+            config.WebDriver.driver.find_element(By.ID, i).click()
         except:
             pass
 
 
-# save webdriver cookies
-def save_cookies(driver) -> None:
-    if g.dev:
-        f = open(f"{'/'.join(__file__.split('/')[:-2])}/user_data/cookies/{g._mail}_unsafe.pkl", "w")
-        for i in driver.get_cookies():
-            f.write(str(i) + "\n")
-        f.close()
-    else :
-        pickle.dump(driver.get_cookies(), open(f"{'/'.join(__file__.split('/')[:-2])}/user_data/cookies/{g._mail}.pkl", "wb"))
-
-
-# load cookies previously saved to the driver
-def load_cookies(driver) -> None:
-    if g.dev:
-        f = open(f"{'/'.join(__file__.split('/')[:-2])}/user_data/cookies/{g._mail}_unsafe.pkl", "r")
-        lines = f.readlines()
-        f.close()
-        cookies = [literal_eval(x) for x in lines]
-    else :
-        cookies = pickle.load(open(f"{'/'.join(__file__.split('/')[:-2])}/user_data/cookies/{g._mail}.pkl", "rb"))
-    for cookie in cookies:
-        driver.add_cookie(cookie)
-
 """
-send_keys_wait([selenium element:element, str:keys]) send the different keys to the field element, with a random time between each press to simulate human action.
+send_keys_wait([selenium element:element, str:keys]) send the different keys to the field element, with a random 
+time between each press to simulate human action.
 keys can be an string, but also selenium keys
 """
+
+
 def send_keys_wait(element, keys: str) -> None:
     for i in keys:
         element.send_keys(i)
         sleep(uniform(0.1, 0.3))
 
 
-
 # Wait for the presence of the element identifier or [timeout]s
-def wait_until_visible(search_by: str, identifier: str, timeout = 20, browser = None) -> None:
-    try :
-        WebDriverWait(browser, timeout).until(EC.visibility_of_element_located((search_by,identifier)), "element not found")
-        return(True)
+def wait_until_visible(search_by: str, identifier: str, timeout: int = 20, browser=None) -> bool:
+    try:
+        WebDriverWait(browser, timeout).until(
+            expected_conditions.visibility_of_element_located((search_by, identifier)), "element not found")
+        return True
     except TimeoutException as e:
         printf(f"element {identifier} not found after {timeout}s")
-        return(False)
-
+        return False

modules/error.py

@@ -1,8 +1,10 @@
 class Banned(Exception):
     pass
 
+
 class NotBanned(Exception):
     pass
 
+
 class Identity(Exception):
-    pass
+    pass

modules/globals.py

@@ -1,30 +0,0 @@
-driver = None
-display = None
-log = False
-full_log = False
-vnc_enabled = False
-vnc_port = 2345
-points_file = "/"
-update_version = False
-start_time = 0
-mot_path = "/"
-credential_path = "/"
-discord_success_link = "https://example.com"
-discord_error_link = "https://example.com"
-discord_enabled_error = False
-discord_enabled_success = False
-avatar_url = ""
-fidelity_link = "None"
-discord_embed = False
-headless = False
-proxy_enabled = False
-proxy_address = "0.0.0.0"
-proxy_port = "0"
-sql_enabled = False
-sql_usr = "None"
-sql_pwd = "azerty"
-sql_host = "https://example.com"
-sql_database = "MsRewards"
-norvege = False
-database_error_override = False
-json_start = ""

modules/tools.py

@@ -1,82 +0,0 @@
-from modules.imports import *
-from modules.config import *
-from modules.db import *
-import modules.globals as g
-# add the time arround the text given in [text]&
-def Timer(text: str) -> str:
-    return(f"[{g._mail.split('@')[0]} - {datetime.today().strftime('%d/%m')} - {timedelta(seconds = round(float(time() - g.start_time)))}] " + str(text))
-
-
-# replace the function print, with more options
-# [txt] : string, [driver] : selenium webdriver
-def printf(txt):
-    print(Timer(txt))
-
-
-# return current page domain
-def get_domain(driver):
-    return(driver.current_url.split("/")[2])
-
-    
-# check if the user is using IPV4 using ipify.org
-# [driver] : selenium webdriver
-# never used here
-# can be useful as Ms had issues with IPV6 at some point
-def check_ipv4(driver):
-    driver.get("https://api64.ipify.org")
-    elm = driver.find_element(By.TAG_NAME, "body")
-    if len(elm.text.split('.')) == 4 :
-        return True
-    return False
-
-
-def custom_sleep(temps):
-    try : 
-        if g.log : #only print sleep when user see it
-            points = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"]
-            passe = 0
-            for i in range(int(temps)):
-                for i in range(8):
-                    sleep(0.125)
-                    passe += 0.125
-                    print(f"{points[i]}  -  {round(float(temps) - passe, 3)}", end="\r")
-            print("                        ", end="\r")
-        else:
-            sleep(temps)
-    except KeyboardInterrupt :
-        print("attente annulée")
-
-
-def format_error(e) -> str:
-    tb = e.__traceback__
-    txt = ""
-    while tb != None :
-        txt = txt + f" -> {tb.tb_frame.f_code.co_name} ({tb.tb_lineno}) "
-        tb = tb.tb_next
-    return(txt + "\n" + str(e))
-
-
-def progressBar(current, total=30, barLength=20, name="Progress"):
-    percent = float(current + 1) * 100 / total
-    arrow = "-" * int(percent / 100 * barLength - 1) + ">"
-    spaces = " " * (barLength - len(arrow))
-    print(name + ": [%s%s] %d %%" % (arrow, spaces, percent), end="\r")
-
-
-def save_points_from_file(file):
-    with open(file) as f:
-        read = reader(f)
-        points_list = list(read)
-    for item in points_list:
-        compte, points = item[0], item[1]
-        add_to_database(compte, points, g.sql_host,g.sql_usr,g.sql_pwd,g.sql_database, save_if_fail=False)
-    with open(file, "w") as f:
-        f.write("")
-
-
-def select_accounts(multiple = True):
-    system("clear")  # clear from previous command to allow a clean choice
-    emails = [x[0] for x in g._cred]  # list of all email adresses
-    emails_selected = enquiries.choose(f"quel{'s' if multiple else ''} compte{'s' if multiple else ''} ?", emails, multi=multiple)
-    return([x for x in g._cred if x[0] in emails_selected])
-