mirror of
https://github.com/XiaoliChan/wmiexec-Pro.git
synced 2026-10-18 22:50:54 +08:00
XiaoliChan
244 lines
13 KiB
Python
244 lines
13 KiB
Python
import logging
|
|
import json
|
|
import sys
|
|
|
|
from io import StringIO
|
|
from impacket.dcerpc.v5.dtypes import NULL
|
|
|
|
from lib.checkError import checkError
|
|
from lib.module_base import ModuleBase
|
|
|
|
|
|
class Firewall_Toolkit(ModuleBase):
|
|
name = "firewall"
|
|
description = "Firewall abusing."
|
|
|
|
@staticmethod
|
|
def register_parser(subparsers):
|
|
p = subparsers.add_parser(Firewall_Toolkit.name, help=Firewall_Toolkit.description)
|
|
p.add_argument("-search-port", action="store", metavar="port num", help="Search rules associate with the port.")
|
|
p.add_argument("-dump", action="store", metavar="FILENAME", help="Dump all firewall rules to file as json format.")
|
|
p.add_argument("-rule-id", action="store", metavar="ID", help='Specify firewall rule instance id to do operation in "-rule-op"')
|
|
p.add_argument("-action", action="store", default="disable", choices=["enable", "disable", "remove"],
|
|
help="Action of firewall rule which you specify.")
|
|
p.add_argument("-firewall-profile", action="store", choices=["enable", "disable"],
|
|
help="Use it on your own risk if you try to do this one.")
|
|
return p
|
|
|
|
@staticmethod
|
|
def run(iWbemLevel1Login, dcom, options, **kwargs):
|
|
toolkit = Firewall_Toolkit(iWbemLevel1Login)
|
|
if options.search_port:
|
|
toolkit.port_Searcher(options.search_port)
|
|
if options.dump:
|
|
toolkit.dump_FirewallRules(options.dump)
|
|
if options.rule_id and options.action:
|
|
toolkit.rule_Controller(ID=options.rule_id, flag=options.action)
|
|
if options.firewall_profile:
|
|
toolkit.FirewallProfile_Controller(options.firewall_profile)
|
|
|
|
def __init__(self, iWbemLevel1Login):
|
|
self.iWbemLevel1Login = iWbemLevel1Login
|
|
self.logger = logging.getLogger("wmiexec-pro")
|
|
|
|
def port_Searcher(self, port, returnID = False):
|
|
self.logger.info(f"Searching rule include the specified port: {port}")
|
|
id_List = []
|
|
iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/StandardCimv2", NULL, NULL)
|
|
self.iWbemLevel1Login.RemRelease()
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery("SELECT InstanceID, LocalPort, Protocol, RemotePort FROM MSFT_NetProtocolPortFilter")
|
|
while True:
|
|
try:
|
|
firewall_PortClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
record = dict(firewall_PortClass.getProperties())
|
|
for i in record["LocalPort"]["value"]:
|
|
if str(port) == i and record["InstanceID"]["value"]:
|
|
result = self.instanceID_Searcher(record["InstanceID"]["value"], iWbemServices)
|
|
if not returnID:
|
|
result = "[+] Rule id: {}, DisplayName: {}, Direction: {}, Port: {}, Action: {}, Status: {}".format(
|
|
record["InstanceID"]["value"],
|
|
result["DisplayName"],
|
|
result["Direction"],
|
|
record["LocalPort"]["value"],
|
|
result["Action"],
|
|
result["Status"]
|
|
)
|
|
self.logger.log(100, result)
|
|
if returnID:
|
|
id_List.append(record["InstanceID"]["value"])
|
|
except Exception as e:
|
|
if str(e).find("S_FALSE") < 0:
|
|
pass
|
|
else:
|
|
break
|
|
iEnumWbemClassObject.RemRelease()
|
|
if returnID:
|
|
return id_List
|
|
|
|
def instanceID_Searcher(self, ID, iWbemServices=None):
|
|
# If user do NTLMLogin operate to many time, then will cause RPC_E_DISCONNECTED exception(DCOM connection rate limit)
|
|
# So, we can keep using iWbemServices object from port_Searcher when doing enumeration, and do newlogin when doing invoke this function.
|
|
if not iWbemServices:
|
|
iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/StandardCimv2", NULL, NULL)
|
|
self.iWbemLevel1Login.RemRelease()
|
|
tmp_dict = {}
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery(f'SELECT DisplayName, InstanceID, Direction, Action, Enabled FROM MSFT_NetFirewallRule where InstanceID = "{ID}"')
|
|
firewall_RuleClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
record = dict(firewall_RuleClass.getProperties())
|
|
tmp_dict["DisplayName"] = record["DisplayName"]["value"]
|
|
tmp_dict["Direction"] = "Inbound" if record["Direction"]["value"] == 1 else "Outbound"
|
|
tmp_dict["Action"] = "Allow" if record["Action"]["value"] == 2 else ("AllowBypass" if record["Action"]["value"] == 3 else "Block")
|
|
tmp_dict["Status"] = "Enabled" if record["Enabled"]["value"] == 1 else "Disabled"
|
|
iEnumWbemClassObject.RemRelease()
|
|
return tmp_dict
|
|
|
|
def rule_Controller(self, ID, flag):
|
|
# Can"t invoke disable and enable function directly
|
|
# firewall_RuleClass.Enable/Disable not wotking
|
|
# But we can force push firewall rule :)
|
|
iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/StandardCimv2", NULL, NULL)
|
|
self.iWbemLevel1Login.RemRelease()
|
|
#DisplayName, InstanceID, Direction, Action, Enabled
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery(f'SELECT * FROM MSFT_NetFirewallRule where InstanceID = "{ID}"')
|
|
firewall_RuleClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
#firewall_RuleClass.Enable
|
|
record = dict(firewall_RuleClass.getProperties())
|
|
if flag in ["enable","disable"]:
|
|
firewall_Instance = firewall_RuleClass.SpawnInstance()
|
|
# windows only accept encoded string with latin-1, if chinese character in value, will get exception in impacket/structure.py line 250
|
|
# Have no idea how to solve this, let"s submit uuid instead of original character
|
|
firewall_Instance.ElementName = ""
|
|
firewall_Instance.DisplayName = ""
|
|
firewall_Instance.Description = ""
|
|
firewall_Instance.DisplayGroup = ""
|
|
"""
|
|
try:
|
|
b(record["ElementName"]["value"])
|
|
except:
|
|
firewall_Instance.ElementName = "windows-object-" + str(uuid.uuid4())
|
|
else:
|
|
firewall_Instance.ElementName = record["ElementName"]["value"]
|
|
|
|
try:
|
|
b(record["DisplayName"]["value"])
|
|
except:
|
|
firewall_Instance.DisplayName = "windows-object-" + str(uuid.uuid4())
|
|
else:
|
|
firewall_Instance.DisplayName = record["DisplayName"]["value"]
|
|
|
|
try:
|
|
b(record["Description"]["value"])
|
|
except:
|
|
firewall_Instance.Description = "windows-object-" + str(uuid.uuid4())
|
|
else:
|
|
firewall_Instance.Description = "" if record["Description"]["value"] == None else record["Description"]["value"]
|
|
|
|
try:
|
|
b(record["DisplayGroup"]["value"])
|
|
except:
|
|
firewall_Instance.DisplayGroup = "windows-object-" + str(uuid.uuid4())
|
|
else:
|
|
firewall_Instance.DisplayGroup = "" if record["DisplayGroup"]["value"] == None else record["DisplayGroup"]["value"]
|
|
"""
|
|
# allow=2, allowBypass=3, Block=4
|
|
firewall_Instance.Action = record["Action"]["value"]
|
|
firewall_Instance.Caption = "" if not record["Caption"]["value"] else record["Caption"]["value"]
|
|
firewall_Instance.CommonName = "" if not record["CommonName"]["value"] else record["CommonName"]["value"]
|
|
# Enable = 1, disable = 2
|
|
firewall_Instance.Enabled = 2 if flag == "disable" else 1
|
|
firewall_Instance.LocalOnlyMapping = False if record["LocalOnlyMapping"]["value"] == "False" else True
|
|
firewall_Instance.LooseSourceMapping = False if record["LooseSourceMapping"]["value"] == "False" else True
|
|
firewall_Instance.Mandatory = "" if not record["Mandatory"]["value"] else record["Mandatory"]["value"]
|
|
firewall_Instance.Owner = "" if not record["Owner"]["value"] else record["Owner"]["value"]
|
|
firewall_Instance.RuleGroup = "" if not record["RuleGroup"]["value"] else record["RuleGroup"]["value"]
|
|
firewall_Instance.RuleUsage = "" if not record["RuleUsage"]["value"] else record["RuleUsage"]["value"]
|
|
# Status attribute must be hardcode
|
|
firewall_Instance.Status = "The rule was parsed successfully from the store."
|
|
current=sys.stdout
|
|
sys.stdout = StringIO()
|
|
marshalled = firewall_Instance.marshalMe()
|
|
sys.stdout = current
|
|
checkError(iWbemServices.PutInstance(marshalled))
|
|
else:
|
|
checkError(iWbemServices.DeleteInstance('MSFT_NetFirewallRule.CreationClassName="{}",PolicyRuleName="{}",SystemCreationClassName="{}",SystemName="{}"'.format(
|
|
record["CreationClassName"]["value"],
|
|
record["PolicyRuleName"]["value"],
|
|
record["SystemCreationClassName"]["value"],
|
|
record["SystemName"]["value"]
|
|
)))
|
|
|
|
def dump_FirewallRules(self, save_FileName):
|
|
self.logger.info("Dumpping...")
|
|
iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/StandardCimv2", NULL, NULL)
|
|
self.iWbemLevel1Login.RemRelease()
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery("SELECT DisplayName, InstanceID, Direction, Action, Enabled FROM MSFT_NetFirewallRule")
|
|
firewall_RuleRecord = {}
|
|
while True:
|
|
try:
|
|
tmp_dict = {}
|
|
firewall_RuleClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
record = dict(firewall_RuleClass.getProperties())
|
|
tmp_dict["DisplayName"] = record["DisplayName"]["value"]
|
|
tmp_dict["Direction"] = "Inbound" if record["Direction"]["value"] == 1 else "Outbound"
|
|
tmp_dict["Action"] = "Allow" if record["Action"]["value"] == 2 else ("AllowBypass" if record["Action"]["value"] == 3 else "Block")
|
|
tmp_dict["Status"] = "Enabled" if record["Enabled"]["value"] == 1 else "Disabled"
|
|
firewall_RuleRecord[record["InstanceID"]["value"]] = tmp_dict
|
|
except Exception as e:
|
|
if str(e).find("S_FALSE") < 0:
|
|
pass
|
|
else:
|
|
break
|
|
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery("SELECT InstanceID, LocalPort, Protocol, RemotePort FROM MSFT_NetProtocolPortFilter")
|
|
firewall_RuleDetailRecord = {}
|
|
while True:
|
|
try:
|
|
tmp_dict = {}
|
|
firewall_PortClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
record = dict(firewall_PortClass.getProperties())
|
|
tmp_dict["Protocol"] = record["Protocol"]["value"]
|
|
tmp_dict["LocalPort"] = record["LocalPort"]["value"]
|
|
tmp_dict["RemotePort"] = record["RemotePort"]["value"]
|
|
firewall_RuleDetailRecord[record["InstanceID"]["value"]] = tmp_dict
|
|
except Exception as e:
|
|
if str(e).find("S_FALSE") < 0:
|
|
pass
|
|
else:
|
|
break
|
|
|
|
test = dict(firewall_RuleDetailRecord, **firewall_RuleRecord)
|
|
for key in test.keys():
|
|
if key in firewall_RuleDetailRecord:
|
|
test[key] = dict(test[key], **firewall_RuleDetailRecord[key])
|
|
|
|
with open(save_FileName,"w") as f:
|
|
f.write(json.dumps(test, indent=4))
|
|
|
|
self.logger.log(100, f"[+] Whole the firewall rules are dumped to {save_FileName}")
|
|
|
|
def FirewallProfile_Controller(self, flag):
|
|
# Use it on your own risk :)
|
|
iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/StandardCimv2", NULL, NULL)
|
|
self.iWbemLevel1Login.RemRelease()
|
|
iEnumWbemClassObject = iWbemServices.ExecQuery("SELECT * FROM MSFT_NetFirewallProfile")
|
|
while True:
|
|
try:
|
|
firewall_ProfileClass = iEnumWbemClassObject.Next(0xffffffff,1)[0]
|
|
record = dict(firewall_ProfileClass.getProperties())
|
|
firewall_ProfileInstance = firewall_ProfileClass.SpawnInstance()
|
|
firewall_ProfileInstance.DisabledInterfaceAliases = ""
|
|
firewall_ProfileInstance.Caption = "" if not record["Caption"]["value"] else record["Caption"]["value"]
|
|
firewall_ProfileInstance.Enabled = 1 if flag == "enable" else 0
|
|
firewall_ProfileInstance.Description = "" if not record["Caption"]["value"] else record["Caption"]["value"]
|
|
current=sys.stdout
|
|
sys.stdout = StringIO()
|
|
iWbemServices.PutInstance(firewall_ProfileInstance.marshalMe())
|
|
sys.stdout = current
|
|
except Exception as e:
|
|
if str(e).find("S_FALSE") < 0:
|
|
self.logger.error(str(e))
|
|
raise
|
|
else:
|
|
self.logger.log(100, f"All firewall profile has been {flag}")
|
|
break
|
|
iEnumWbemClassObject.RemRelease() |