cxykevin
/
peinjector
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

补充注释，解决path未刷新问题

This commit is contained in:
cxykevin 2024-02-12 12:27:11 +08:00
parent 60db1e7aa4
commit c2d7b0fe51
8 changed files with 154 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/action.py
View File

@ -116,17 +116,23 @@ def save_action(actions_list: list) -> None:
return tmp_filename
def do_action(actions_list: list) -> str:
def do_action(actions_list: list) -> tuple:
removeinfo = {}
actionlogs = "----- Action Log Start -----\r\n"
thispack = "__null__"
for i in actions_list:
if i[0] == 0:
actionlogs = actionlogs+f"Load [{i[1]}]\r\n"
thispack = i[1]
removeinfo[thispack] = []
elif i[0] == 1:
try:
os.mkdir(i[1], mode=0o777)
except Exception as exp:
actionlogs = actionlogs + \
f" mkdir \"{i[1]}\" failed: {repr(exp)}\r\n"
else:
removeinfo[thispack].insert(0, i)
elif i[0] == 2:
try:
shutil.copyfile(i[1], i[2])
@ -175,4 +181,5 @@ def do_action(actions_list: list) -> str:
except:
actionlogs = actionlogs + \
f" add reg \"{i[1]}\" failed: {repr(exp)}\r\n"
return actionlogs
print(removeinfo)
return (actionlogs, removeinfo)

32
src/hook.pyw
View File

@ -10,6 +10,9 @@ import sys
import os
import time
import pickle
import traceback
# 导入路径
sys.path.append("X:\\PEinjector\\src")
sys.path.append("X:\\PEinjector\\tool")
try:
@ -20,10 +23,12 @@ except:
sys.path.append("src")
sys.path.append("tool")
# 导入模块
log = __import__("log")
config = __import__("config")
action = __import__("action")
utils = __import__("utils")
regwrite = __import__("regwrite")
# logo
logo = """\033[1m\033[34m
@ -36,9 +41,11 @@ logo = """\033[1m\033[34m
\033[0m\033[?25l"""
print(logo)
# 初始化日志系统
log.init()
log.info("hook start")
try:
# 等待动画
sys.stdout.write("\033[3A \033[37m")
for j in range(5):
for i in range(41):
@ -51,6 +58,7 @@ for j in range(5):
sys.stdout.write("\n\033[1A \033[37m")
sys.stdout.write("\033[0m\033[?25h\n\n\n")
# 搜索磁盘和用户名
sysdrive = "X:"
if not os.path.exists(sysdrive+"\\"):
sysdrive = "C:"
@ -59,18 +67,28 @@ for i in os.listdir(f"{sysdrive}\\Users"):
if i not in ("Default", "Default User", "Public", "All Users") and not os.path.isfile(f"{sysdrive}\\Users\\{i}"):
username = i
# wait
# 等待
while 1:
time.sleep(1)
if os.path.exists(f"{sysdrive}\\Users\\{username}\\Desktop"):
break
# load action list
# 加载
log.info("load action \"onload\"")
with open(config.TEMP_DIR.replace("{TEMP}", "X:\\PEinjector")+"\\"+"PEinjector.tmp", "rb") as file:
lists = pickle.load(file)
lists = pickle.load(file) # 读取文件
alog = action.do_action(lists)
with open(config.ACTIONLOGPATH.replace("{DISK}", utils.find_disk()), "w") as file:
file.write(alog)
with open(config.ACTIONLOGPATH.replace("{DISK}", utils.find_disk()), "a") as file:
file.write(alog[0]) # 追加写入日志
with open(config.TEMP_DIR.replace("{TEMP}", "X:\\PEinjector")+"\\"+"remove.tmp", "wb") as file:
pickle.dump(alog[1], file) # 写入卸载文件
# 刷新path
regwrite.refresh_path()
# 加载完成
log.info("done")
except Exception as exp: # 未知错误
log.break_err("Exception \n"+str(traceback.format_exc(exp)))
raise exp
print("This window will close after 3 seconds.")
time.sleep(3)

121
src/loader.py
View File

@ -11,10 +11,10 @@ import traceback
import config
import action
loaded_package = []
loaderr_pkgs = []
disk = ""
lists = []
loaded_package = [] # 加载过包
loaderr_pkgs = [] # 加载错误的包
disk = "" # PEinjector 安装盘
lists = [] # 软件包列表
def __version_parse(version: str) -> list:
@ -31,6 +31,7 @@ def __version_compare(ver1: str, ver2: str) -> int:
def version_check(configuration: dict, pkg_name: str) -> int:
# 消息格式
ERROR_MESSAGE = "load moudle [{}] failed: PEinjector version too {}, need [{}]"
try:
if not "compatibility" in configuration or not "injector" in configuration["compatibility"]:
@ -40,17 +41,17 @@ def version_check(configuration: dict, pkg_name: str) -> int:
if "min" in configuration["compatibility"]["injector"]:
plugver = configuration["compatibility"]["injector"]["min"]
if __version_compare(version, plugver) == -1:
if __version_compare(version, plugver) == -1: # 版本过低
log.warn(ERROR_MESSAGE.format(pkg_name, "low", plugver))
return 5
if "max" in configuration["compatibility"]["injector"]:
plugver = configuration["compatibility"]["injector"]["max"]
if __version_compare(version, plugver) == 1:
if __version_compare(version, plugver) == 1: # 版本或高
log.warn(ERROR_MESSAGE.format(pkg_name, "high", plugver))
return 6
except Exception as e:
except Exception as e: # 未知错误
log.warn(f"load moudle [{pkg_name}] failed: {e}")
return 7
return 0
@ -59,16 +60,21 @@ def version_check(configuration: dict, pkg_name: str) -> int:
def file_check(file_json: dict, pkg_name: str) -> int:
try:
for i in file_json.get("compatibility", {}).get("file", {}).get("must", []):
if not os.path.exists(i):
if not os.path.exists(i): # 找不到文件
log.warn(f"load moudle [{pkg_name}] failed: " +
f"Cannot find file: [{i}]")
return 8
for i in file_json.get("compatibility", {}).get("file", {}).get("mustnot", []):
if os.path.exists(i):
if os.path.exists(i): # 找到不兼容文件
log.warn(f"load moudle [{pkg_name}] failed: " +
f"Find incompatible file: [{i}]")
return 9
except:
for i in file_json.get("compatibility", {}).get("file", {}).get("loaded", []):
if os.path.exists(i): # 已经无需加载
log.info(f"jump moudle [{pkg_name}]: " +
f"Find compatible file: [{i}]")
return -1
except: # 未知错误
log.warn(f"load moudle [{pkg_name}] failed: " +
"Unknown error in file check")
return 10
@ -81,14 +87,14 @@ actions = {"onboot": [], "onload": []}
def load_package(pkg_name: str) -> int:
pkg_path = f"{disk}/PEinjector/package/{pkg_name}"
loaded_package.append(pkg_name)
if "manifest.json" not in os.listdir(pkg_path):
if "manifest.json" not in os.listdir(pkg_path): # 连manifest都没有
log.warn(f"load moudle [{pkg_name}] failed: " +
"Cannot find manifest.json")
return 1
try:
with open(pkg_path+"/"+"manifest.json", "r", encoding="utf-8") as file:
file_json = json.load(file)
except json.decoder.JSONDecodeError:
except json.decoder.JSONDecodeError: # json格式错误
log.warn(f"load moudle [{pkg_name}] failed: " +
"Json syntax error")
return 3
@ -96,32 +102,34 @@ def load_package(pkg_name: str) -> int:
log.warn(f"load moudle [{pkg_name}] failed: " +
"Unknown error in read file ("+repr(exp)+")")
return 2
for i in ("version", "name", "author", "introduce"):
for i in ("version", "name", "author", "introduce"): # 检查必须字段
if i not in file_json:
log.warn(f"load moudle [{pkg_name}] failed: " +
f"\"{i}\" key not in manifest.json")
return 4
retvar = version_check(file_json, pkg_name)
retvar = version_check(file_json, pkg_name) # 检查版本号
if retvar != 0:
return retvar
retvar = file_check(file_json, pkg_name)
retvar = file_check(file_json, pkg_name) # 检查所需文件
if retvar == -1:
return 0
if retvar != 0:
return retvar
if "dependence" in file_json:
if "dependence" in file_json: # 依赖加载
try:
for i in file_json["dependence"]:
if i not in lists:
if i not in lists: # 找不到依赖
log.warn(f"load moudle [{pkg_name}] failed: " +
f"Cannot find dependence [{i}]")
return 11
if retvar in loaderr_pkgs:
if i in loaderr_pkgs: # 软件包已经加载过但出错
log.warn(f"load moudle [{pkg_name}] failed: " +
f"dependence [{i}] loaded failed")
return 12
if retvar not in loaded_package:
if i not in loaded_package: # 软件包未加载
log.info(f"load moudle [{i}] from [{pkg_name}]")
retvar = load_package(i)
if retvar != 0:
retvar = load_package(i) # （递归）加载软件包
if retvar != 0: # 加载出错
loaderr_pkgs.append(i)
log.warn(f"load moudle [{pkg_name}] failed: " +
f"dependence [{i}] loaded failed")
@ -130,14 +138,15 @@ def load_package(pkg_name: str) -> int:
log.warn(f"load moudle [{pkg_name}] failed: " +
f"Cannot find dependence [{i}]")
data_list = []
if "data" in file_json:
j = -1
prep_flag = False
if "data" in file_json: # 对于数据的预处理，实际在load处理链接
j = -1 # 计数
prep_flag = False # 判断 data 是否初始化
if pkg_name not in os.listdir(config.DATAPATH.replace("{DISK}", utils.find_disk())):
log.info(f"prep moudle [{pkg_name}] data")
prep_flag = True
for i in file_json["data"]:
j += 1
# 语法检查缺少成分
flag = False
for event in ("from", "to"):
if event not in i or i[event] == "":
@ -147,7 +156,9 @@ def load_package(pkg_name: str) -> int:
break
if flag:
continue
# 添加
data_list.append(i)
# 初始化 data
if prep_flag:
data_dir = config.DATAPATH.replace(
"{DISK}", utils.find_disk())+"/"+pkg_name
@ -158,32 +169,38 @@ def load_package(pkg_name: str) -> int:
else:
shutil.copytree(
pkg_path+"/"+i["from"], data_dir+"/"+i["to"])
add_action_head = False
# 添加过头部标记
onload_add_action_head = False
onboot_add_action_head = False
if "load" in file_json:
open_symlink = config.USE_SYMLINK
if "symlink" in file_json["load"] and file_json["load"] == False:
open_symlink = config.USE_SYMLINK # 启用链接加载
if "symlink" in file_json["load"] and file_json["load"] == False: # 软件包明确禁用
open_symlink = False
if "mode" in file_json["load"]:
for event in ("onboot", "onload"):
if event not in file_json["load"]["mode"]:
continue
# 添加软件头
actions[event].append((0, pkg_name))
if event == "onload":
add_action_head = True
onload_add_action_head = True
if event == "onboot":
onboot_add_action_head = True
# 语法类型检查
if type(file_json["load"]["mode"][event]) != list:
log.warn(f"load moudle [{pkg_name}] failed: " +
f"Load commands syntax error on {event} (must be a list)")
return 13
j = -1
j = -1 # 软件包action计数，从 0 开始，用于log
for i in file_json["load"]["mode"][event]:
j += 1
if "type" not in i:
if "type" not in i: # 未知类型
log.warn(f"load moudle [{pkg_name}] warning: " +
f"Load commands syntax error on {event}[{j+1}] (lost \"type\"), igrone")
continue
# 强制复制（复制）
if i["type"] == "force_copy" or (i["type"] == "copy" and open_symlink == False):
flag = False
# 语法检查缺少成分
for check in ("from", "to"):
if check not in i:
flag = True
@ -192,13 +209,17 @@ def load_package(pkg_name: str) -> int:
break
if flag:
continue
# 检查路径有效性
if not os.path.exists(pkg_path+"/"+i["from"]):
log.warn(f"load moudle [{pkg_name}] warning: " +
f"Load commands file error on {event}[{j+1}] (cannot find \"{i['from']}\"), igrone")
continue
# 添加
actions[event] += action.force_copy(pkg_path,
pkg_name, data_list, i["from"], i["to"])
# 复制（链接）
elif i["type"] == "copy":
# 语法检查缺少成分
flag = False
for check in ("from", "to"):
if check not in i:
@ -208,13 +229,16 @@ def load_package(pkg_name: str) -> int:
break
if flag:
continue
# 检查路径有效性
if not os.path.exists(pkg_path+"/"+i["from"]):
log.warn(f"load moudle [{pkg_name}] warning: " +
f"Load commands file error on {event}[{j+1}] (cannot find \"{i['from']}\"), igrone")
continue
# 添加
actions[event] += action.copy(pkg_path, pkg_name, data_list,
i["from"], i["to"])
elif i["type"] == "start":
elif i["type"] == "start": # 非阻塞启动程序
# 语法检查缺少成分
flag = False
for check in ("command", ):
if check not in i:
@ -224,38 +248,50 @@ def load_package(pkg_name: str) -> int:
break
if flag:
continue
# 添加
actions[event] += action.start(
pkg_name, data_list, i["command"])
else:
log.warn(f"load moudle [{pkg_name}] warning: " +
f"Load commands syntax error on {event}[{j+1}] (unknown type), igrone")
if "start" in file_json:
if not add_action_head:
actions[event].append((0, pkg_name))
if "icon" in file_json["start"]:
# 补充包开始加载声明
if not onload_add_action_head:
actions["onload"].append((0, pkg_name))
# 语法检查缺少成分
for i in ("command", "name", "icon"):
if i not in file_json["start"]["icon"]:
log.warn(f"load moudle [{pkg_name}] warning: " +
f"Load icon syntax error (lost \"{i}\"), igrone")
# 相对路径转绝对路径
if len(file_json["start"]["icon"]["icon"]) < 2 or file_json["start"]["icon"]["icon"][1] != ':':
file_json["start"]["icon"]["icon"] = pkg_path + \
"/"+file_json["start"]["icon"]["icon"]
if len(file_json["start"]["icon"]["command"]) < 2 or file_json["start"]["icon"]["command"][1] != ':':
file_json["start"]["icon"]["command"] = pkg_path + \
"/"+file_json["start"]["icon"]["command"]
# 添加
actions["onload"].append(
(6, file_json["start"]["icon"]["command"], file_json["start"]["icon"]["icon"], file_json["start"]["icon"]["name"]))
if "path" in file_json["start"]:
for i in file_json["start"]["path"]:
# 补充包开始加载声明
if not onload_add_action_head:
actions["onload"].append((0, pkg_name))
for i in file_json["start"]["path"]: # 加载多个path
# 相对路径转绝对路径
if len(i) < 2 or i[1] != ':':
i = pkg_path + \
"/"+i
# 添加
actions["onload"].append((7, i))
if "reg" in file_json:
for i in file_json["reg"]:
if "reg" in file_json: # 添加注册表
for i in file_json["reg"]: # 多个注册表文件
# 相对路径转绝对路径
if len(i) < 2 or i[1] != ':':
i = pkg_path + \
"/"+i
# 添加
actions["onload"].append((8, i))
@ -265,18 +301,21 @@ def load():
log.info("start load")
disk = utils.find_disk()
lists = os.listdir(f"{disk}/PEinjector/package")
# 读取禁用包列表
with open(config.DISABLEPATH.replace("{DISK}", disk), "r") as file:
disable_packages = [i.rstrip("\n\r") for i in file.readlines()]
for packs in lists:
for packs in lists: # 加载包
if packs not in loaded_package and packs not in disable_packages:
log.info(f"load moudle [{packs}]")
retvar = load_package(packs)
if retvar == 0:
loaderr_pkgs.append(packs)
# 保存并执行action
action.save_action(actions["onload"])
alog = action.do_action(actions["onboot"])
alog = action.do_action(actions["onboot"])[0]
# 写action日志
with open(config.ACTIONLOGPATH.replace("{DISK}", utils.find_disk()), "w") as file:
file.write(alog)
except Exception as exp:
except Exception as exp: # 未知错误
log.break_err("Exception \n"+str(traceback.format_exc(exp)))
raise exp

4
src/makelnk.py
View File

@ -8,6 +8,7 @@ import pylnk3
# Thank the project "pylnk3"
# 搜索系统磁盘用户名
sysdrive = "X:"
if not os.path.exists(sysdrive+"\\"):
sysdrive = "C:"
@ -17,8 +18,9 @@ for i in os.listdir(f"{sysdrive}\\Users"):
username = i
def makelnk(name: str, exepath: str, iconpath: str) -> None:
def makelnk(name: str, exepath: str, iconpath: str) -> None: # 创建快捷方式
log.info(f"make shortcut \"{name}\"")
# 等价于 pylnk3 create
pylnk3.for_file(exepath,
lnk_name=f"{sysdrive}\\Users\\{username}" +
f"\\Desktop\\{name}.lnk",

12
src/regwrite.py
View File

@ -7,7 +7,7 @@ import log
import os
def add_path(path: str) -> None:
def add_path(path: str) -> None: # 写注册表添加path
log.info(f"add path \"{path}\"")
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
r"SYSTEM\CurrentControlSet\Control\Session Manager\Environment", 0, winreg.KEY_READ) as key:
@ -20,6 +20,12 @@ def add_path(path: str) -> None:
winreg.SetValueEx(key, "Path", 0, winreg.REG_EXPAND_SZ, new_path)
def add_reg(regpath: str) -> None:
def add_reg(regpath: str) -> None: # 导入注册表文件
log.info(f"add reg \"{regpath}\"")
assert os.system(f"reg import \"{regpath.replace("/", "\\")}\"") == 0
assert os.system(f"reg import \"{regpath.replace("/", "\\")}\">nul") == 0
def refresh_path() -> None: # Thank for bbs.bathome.net/thread-62029-1-1.html and dllcall.exe
log.info("refresh path env")
os.system(
f"start /I /B {os.path.dirname(__file__).replace("src", "tool")}\\dllcalls.bat>nul")

1
src/utils.py
View File

@ -5,6 +5,7 @@
import os
# 找磁盘
def find_disk() -> str: # This code from dashedgeless
disk_list = "CDEFGHIJKABLMNOPQRSTUVWYZ"
for i in disk_list:

BIN
tool/dllcall.exe Normal file
View File

Binary file not shown.

2
tool/dllcalls.bat Normal file
View File

@ -0,0 +1,2 @@
@echo off
X:\PEinjector\tool\dllcall SendMessageTimeoutA,0xffff,0x1a,0,"Environment",2,5000,0,user32.dll