mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-23 08:02:24 +02:00
Jack
a24f2e23c6
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
730 lines
29 KiB
Python
730 lines
29 KiB
Python
#!/www/server/panel/pyenv/bin/python3.7
|
|
# coding: utf-8
|
|
# -------------------------------------------------------------------
|
|
# aapanel
|
|
# -------------------------------------------------------------------
|
|
# Copyright (c) 2014-2099 aapanel(http://www.aapanel.com) All rights reserved.
|
|
# -------------------------------------------------------------------
|
|
# Author: wzz <wzz@bt.cn>
|
|
# -------------------------------------------------------------------
|
|
|
|
# ------------------------------
|
|
# 系统防火墙模型 - ufw封装库
|
|
# ------------------------------
|
|
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
if "/www/server/panel/class" not in sys.path:
|
|
sys.path.insert(0, "/www/server/panel/class")
|
|
import public
|
|
# import re
|
|
from firewallModel.app.appBase import Base
|
|
|
|
class Ufw(Base):
|
|
def __init__(self):
|
|
self.cmd_str = self._set_cmd_str()
|
|
|
|
def _set_cmd_str(self):
|
|
return "ufw"
|
|
|
|
# 2024/3/19 下午 5:00 获取系统防火墙的运行状态
|
|
def status(self):
|
|
'''
|
|
@name 获取系统防火墙的运行状态
|
|
@author wzz <2024/3/19 下午 5:00>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run([self.cmd_str, "status"], capture_output=True, text=True, check=True)
|
|
if "Status: active" in result.stdout:
|
|
return "running"
|
|
elif "状态： 激活" in result.stdout:
|
|
return "running"
|
|
else:
|
|
return "not running"
|
|
except subprocess.CalledProcessError:
|
|
return "not running"
|
|
except Exception as e:
|
|
return "not running"
|
|
|
|
# 2024/3/19 下午 5:00 获取系统防火墙的版本号
|
|
def version(self):
|
|
'''
|
|
@name 获取系统防火墙的版本号
|
|
@author wzz <2024/3/19 下午 5:00>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run([self.cmd_str, "version"], capture_output=True, text=True, check=True)
|
|
info = result.stdout.replace("\n", "")
|
|
return info.replace("ufw ", "")
|
|
except Exception as e:
|
|
return "未知版本"
|
|
|
|
# 2024/3/19 下午 5:00 启动防火墙
|
|
def start(self):
|
|
'''
|
|
@name 启动防火墙
|
|
@author wzz <2024/3/19 下午 5:00>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
stdout, stderr = public.ExecShell("echo y | {} enable".format(self.cmd_str))
|
|
if stderr:
|
|
return self._result(False, "启动防火墙失败:{}".format(stderr))
|
|
return self._result(True, "启动防火墙成功")
|
|
except Exception as e:
|
|
return self._result(False, "启动防火墙失败:{}".format(str(e)))
|
|
|
|
# 2024/3/19 下午 5:00 停止防火墙
|
|
def stop(self):
|
|
'''
|
|
@name 停止防火墙
|
|
@author wzz <2024/3/19 下午 5:00>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
stdout, stderr = public.ExecShell("{} disable".format(self.cmd_str))
|
|
if stderr:
|
|
return self._result(False, "停止防火墙失败:{}".format(stderr))
|
|
return self._result(True, "停止防火墙成功")
|
|
except Exception as e:
|
|
return self._result(False, "停止防火墙失败:{}".format(str(e)))
|
|
|
|
# 2024/3/19 下午 4:59 重启防火墙
|
|
def restart(self):
|
|
'''
|
|
@name 重启防火墙
|
|
@author wzz <2024/3/19 下午 4:59>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
self.stop()
|
|
self.start()
|
|
except Exception as e:
|
|
return self._result(False, "重启防火墙失败:{}".format(str(e)))
|
|
|
|
# 2024/3/19 下午 4:59 重载防火墙
|
|
def reload(self):
|
|
'''
|
|
@name 重载防火墙
|
|
@author wzz <2024/3/19 下午 4:59>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
subprocess.run([self.cmd_str, "reload"], check=True, stdout=subprocess.PIPE)
|
|
except Exception as e:
|
|
return self._result(False, "重载防火墙失败:{}".format(str(e)))
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有端口规则
|
|
def list_port(self):
|
|
'''
|
|
@name 列出防火墙中所有端口规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
item_fire = self._load_info(line, "port")
|
|
if item_fire.get("Port") and item_fire["Port"] != "Anywhere" and "." not in item_fire["Port"]:
|
|
item_fire["Port"] = item_fire["Port"].replace(":", "-")
|
|
item_fire["Address"] = "all" if item_fire["Address"] == "Anywhere" else item_fire["Address"]
|
|
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有input端口规则
|
|
def list_input_port(self):
|
|
'''
|
|
@name 列出防火墙中所有input端口规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
item_fire = self._load_info(line, "port")
|
|
if item_fire.get("Port") and item_fire["Port"] != "Anywhere" and "." not in item_fire["Port"]:
|
|
item_fire["Port"] = item_fire["Port"].replace(":", "-")
|
|
|
|
if item_fire["Chain"] == "INPUT":
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有output端口规则
|
|
def list_output_port(self):
|
|
'''
|
|
@name 列出防火墙中所有output端口规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
item_fire = self._load_info(line, "port")
|
|
if item_fire.get("Port") and item_fire["Port"] != "Anywhere" and "." not in item_fire["Port"]:
|
|
item_fire["Port"] = item_fire["Port"].replace(":", "-")
|
|
|
|
if item_fire["Chain"] == "OUTPUT":
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有的ip规则
|
|
def list_address(self):
|
|
'''
|
|
@name 列出防火墙中所有的ip规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
item_fire = self._load_info(line, "address")
|
|
if "Port" in item_fire: continue
|
|
if item_fire.get("Address"):
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有input的ip规则
|
|
def list_input_address(self):
|
|
'''
|
|
@name 列出防火墙中所有input的ip规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
if " IN" not in line:
|
|
continue
|
|
item_fire = self._load_info(line, "address")
|
|
if "Port" in item_fire: continue
|
|
if item_fire.get("Address"):
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 上午 10:39 列出防火墙中所有output的ip规则
|
|
def list_output_address(self):
|
|
'''
|
|
@name 列出防火墙中所有output的ip规则
|
|
@author wzz <2024/3/19 上午 10:39>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
result = subprocess.run(
|
|
[self.cmd_str, "status", "verbose"], capture_output=True, text=True, check=True
|
|
)
|
|
port_infos = result.stdout.split("\n")
|
|
datas = []
|
|
is_start = False
|
|
for line in port_infos:
|
|
if line.startswith("-"):
|
|
is_start = True
|
|
continue
|
|
if not is_start:
|
|
continue
|
|
if " OUT" not in line:
|
|
continue
|
|
item_fire = self._load_info(line, "address")
|
|
if "Port" in item_fire: continue
|
|
if item_fire.get("Address"):
|
|
datas.append(item_fire)
|
|
return datas
|
|
except Exception as e:
|
|
return []
|
|
|
|
# 2024/3/19 下午 4:59 添加端口规则
|
|
def input_port(self, info, operation):
|
|
'''
|
|
@name 添加端口规则
|
|
@author wzz <2024/3/19 下午 4:59>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
if info["Strategy"] == "accept":
|
|
info["Strategy"] = "allow"
|
|
elif info["Strategy"] == "drop":
|
|
info["Strategy"] = "deny"
|
|
|
|
if info["Port"].find('-') != -1:
|
|
info["Port"] = info["Port"].replace('-', ':')
|
|
|
|
if operation == "add":
|
|
if info['Protocol'].find("/") != -1:
|
|
rich_rule = self.cmd_str + " insert 1 {} {}".format(info['Strategy'], info['Port'])
|
|
stdout, stderr = public.ExecShell(rich_rule)
|
|
elif info['Protocol'] == "tcp/udp":
|
|
stdout, stderr = public.ExecShell(self.cmd_str + " allow " + info['Port'])
|
|
else:
|
|
stdout, stderr = public.ExecShell(self.cmd_str + " allow " + info['Port'] + "/" + info['Protocol'])
|
|
else:
|
|
if info['Protocol'].find("/") != -1:
|
|
rich_rule = "{} delete {} {}".format(self.cmd_str, info['Strategy'], info['Port'])
|
|
stdout, stderr = public.ExecShell(rich_rule)
|
|
elif info['Protocol'] == "tcp/udp":
|
|
stdout, stderr = public.ExecShell(self.cmd_str + " delete allow " + info['Port'])
|
|
else:
|
|
stdout, stderr = public.ExecShell(self.cmd_str + " delete allow " + info['Port'] + "/" + info['Protocol'])
|
|
|
|
if stderr:
|
|
if "setlocale" in stderr:
|
|
return self._result(True, "设置端口规则成功")
|
|
return self._result(False, "设置端口规则失败:{}".format(stderr))
|
|
|
|
return self._result(True, "设置端口规则成功")
|
|
|
|
except Exception as e:
|
|
if "setlocale" in str(e):
|
|
return self._result(True, "设置端口规则成功")
|
|
return self._result(False, "设置端口规则失败:{}".format(str(e)))
|
|
|
|
# 2024/3/24 下午 11:28 设置output端口策略
|
|
def output_port(self, info, operation):
|
|
'''
|
|
@name 设置output端口策略
|
|
@param info: 端口号
|
|
@param operation: 操作
|
|
@return None
|
|
'''
|
|
try:
|
|
if info["Strategy"] == "accept":
|
|
info["Strategy"] = "allow"
|
|
elif info["Strategy"] == "drop":
|
|
info["Strategy"] = "deny"
|
|
|
|
if operation == "add":
|
|
if info['Protocol'].find('/') != -1:
|
|
cmd = "{} {} out {}".format(self.cmd_str, info['Strategy'], info['Port'])
|
|
else:
|
|
cmd = "{} {} out {}/{}".format(self.cmd_str, info['Strategy'], info['Port'], info['Protocol'])
|
|
else:
|
|
if info['Protocol'].find('/') != -1:
|
|
cmd = "{} delete {} out {}".format(self.cmd_str, info['Strategy'], info['Port'])
|
|
else:
|
|
cmd = "{} delete {} out {}/{}".format(self.cmd_str, info['Strategy'], info['Port'], info['Protocol'])
|
|
stdout, stderr = public.ExecShell(cmd)
|
|
if stderr:
|
|
if "setlocale" in stderr:
|
|
return self._result(True, "设置端口规则成功")
|
|
return self._result(False, "设置output端口规则失败:{}".format(stderr))
|
|
return self._result(True, "设置output端口规则成功")
|
|
except Exception as e:
|
|
if "setlocale" in str(e):
|
|
return self._result(True, "设置output端口规则成功")
|
|
return self._result(False, "设置output端口规则失败:{}".format(str(e)))
|
|
|
|
# 2024/3/19 下午 4:58 复杂一些的规则管理
|
|
def rich_rules(self, info, operation):
|
|
'''
|
|
@name 复杂一些的规则管理
|
|
@author wzz <2024/3/19 下午 4:58>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
try:
|
|
if info["Strategy"] == "accept":
|
|
info["Strategy"] = "allow"
|
|
elif info["Strategy"] == "drop":
|
|
info["Strategy"] = "deny"
|
|
else:
|
|
return self._result(False, "未知的策略参数:{}".format(info["Strategy"]))
|
|
|
|
rule_str = "{} insert 1 {} ".format(self.cmd_str, info["Strategy"])
|
|
if "Address" in info and public.is_ipv6(info['Address']):
|
|
rule_str = "{} {} ".format(self.cmd_str, info["Strategy"])
|
|
if operation == "remove":
|
|
rule_str = "{} delete {} ".format(self.cmd_str, info["Strategy"])
|
|
|
|
if "Address" in info and info['Address'] != "all":
|
|
rule_str += "from {} ".format(info['Address'])
|
|
if len(info.get("Protocol", "")) != 0 and "/" not in info['Protocol']:
|
|
rule_str += "proto {} ".format(info['Protocol'])
|
|
if len(info.get("Port", "")) != 0:
|
|
rule_str += "to any port {} ".format(info['Port'])
|
|
stdout, stderr = public.ExecShell(rule_str)
|
|
if stderr:
|
|
if "Rule added" in stdout or "Rule deleted" in stdout or "Rule updated" in stdout or "Rule inserted" in stdout or "Skipping adding existing rule" in stdout:
|
|
return self._result(True, "设置规则成功")
|
|
if "setlocale" in stderr:
|
|
return self._result(True, "设置规则成功")
|
|
return self._result(False, "规则设置失败:{}".format(stderr))
|
|
return self._result(True, "设置规则成功")
|
|
except Exception as e:
|
|
if "setlocale" in str(e):
|
|
return self._result(True, "设置规则成功")
|
|
return self._result(False, "规则设置失败:{}".format(e))
|
|
|
|
# 2024/3/24 下午 11:29 设置output rich_rules
|
|
def output_rich_rules(self, info, operation):
|
|
'''
|
|
@name 设置output rich_rules
|
|
@param info: 规则
|
|
@param operation: 操作
|
|
@return None
|
|
'''
|
|
try:
|
|
if info["Strategy"] == "accept":
|
|
info["Strategy"] = "allow"
|
|
elif info["Strategy"] == "drop":
|
|
info["Strategy"] = "deny"
|
|
else:
|
|
return self._result(False, "未知的策略: {}".format(info["Strategy"]))
|
|
|
|
rule_str = "{} insert 1 {} ".format(self.cmd_str, info["Strategy"])
|
|
if "Address" in info and public.is_ipv6(info['Address']):
|
|
rule_str = "{} {} ".format(self.cmd_str, info["Strategy"])
|
|
if operation == "remove":
|
|
rule_str = "{} delete {} ".format(self.cmd_str, info["Strategy"])
|
|
|
|
if len(info.get("Address", "")) != 0:
|
|
rule_str += "out from {} ".format(info['Address'])
|
|
if len(info.get("Protocol", "")) != 0:
|
|
rule_str += "proto {} ".format(info['Protocol'])
|
|
if len(info.get("Port", "")) != 0:
|
|
rule_str += "to any port {} ".format(info['Port'])
|
|
stdout, stderr = public.ExecShell(rule_str)
|
|
if stderr:
|
|
if "Rule added" in stdout or "Rule deleted" in stdout or "Rule updated" in stdout or "Rule inserted" in stdout or "Skipping adding existing rule" in stdout:
|
|
return self._result(True, "设置output规则成功")
|
|
if "setlocale" in stderr:
|
|
return self._result(True, "设置规则成功")
|
|
return self._result(False, "outpu规则设置失败:{}".format(stderr))
|
|
return self._result(True, "设置output规则成功")
|
|
except Exception as e:
|
|
if "setlocale" in str(e):
|
|
return self._result(True, "设置output规则成功")
|
|
return self._result(False, "outpu规则设置失败:{}".format(e))
|
|
|
|
# 2024/3/19 下午 5:01 解析防火墙规则信息，返回字典格式数据，用于添加或删除防火墙规则
|
|
def _load_info(self, line, fire_type):
|
|
'''
|
|
@name 解析防火墙规则信息，返回字典格式数据，用于添加或删除防火墙规则
|
|
@author wzz <2024/3/19 上午 10:38>
|
|
@param "data":{"参数名":""} <数据类型> 参数描述
|
|
@return dict{"status":True/False,"msg":"提示信息"}
|
|
'''
|
|
fields = line.split()
|
|
item_info = {}
|
|
if "LIMIT" in line or "ALLOW FWD" in line:
|
|
return item_info
|
|
if len(fields) < 4:
|
|
return item_info
|
|
if fields[0] == "Anywhere" and fire_type != "port":
|
|
item_info["Strategy"] = "drop"
|
|
|
|
if fields[1] != "(v6)":
|
|
if fields[1] == "ALLOW":
|
|
item_info["Strategy"] = "accept"
|
|
if fields[2] == "IN":
|
|
item_info["Chain"] = "INPUT"
|
|
elif fields[2] == "OUT":
|
|
item_info["Chain"] = "OUTPUT"
|
|
item_info["Address"] = fields[3]
|
|
item_info["Family"] = "ipv4"
|
|
else:
|
|
if fields[2] == "ALLOW":
|
|
item_info["Strategy"] = "accept"
|
|
if fields[3] == "IN":
|
|
item_info["Chain"] = "INPUT"
|
|
elif fields[3] == "OUT":
|
|
item_info["Chain"] = "OUTPUT"
|
|
item_info["Address"] = fields[4]
|
|
item_info["Family"] = "ipv6"
|
|
|
|
return item_info
|
|
|
|
if "/" in fields[0]:
|
|
item_info["Port"] = fields[0].split("/")[0]
|
|
item_info["Protocol"] = fields[0].split("/")[1]
|
|
else:
|
|
item_info["Port"] = fields[0]
|
|
item_info["Protocol"] = "tcp/udp"
|
|
|
|
if "v6" in fields[1]:
|
|
item_info["Family"] = "ipv6"
|
|
if fields[2] == "ALLOW":
|
|
item_info["Strategy"] = "accept"
|
|
else:
|
|
item_info["Strategy"] = "drop"
|
|
|
|
if fields[3] == "IN":
|
|
item_info["Chain"] = "INPUT"
|
|
elif fields[3] == "OUT":
|
|
item_info["Chain"] = "OUTPUT"
|
|
item_info["Address"] = fields[4]
|
|
|
|
else:
|
|
item_info["Family"] = "ipv4" if ":" not in fields[3] else "ipv6"
|
|
|
|
if fields[1] == "ALLOW":
|
|
item_info["Strategy"] = "accept"
|
|
else:
|
|
item_info["Strategy"] = "drop"
|
|
|
|
if fields[2] == "IN":
|
|
item_info["Chain"] = "INPUT"
|
|
elif fields[2] == "OUT":
|
|
item_info["Chain"] = "OUTPUT"
|
|
item_info["Address"] = fields[3]
|
|
|
|
return item_info
|
|
|
|
# 2024/3/25 下午 2:29 设置端口转发
|
|
def port_forward(self, info, operation):
|
|
'''
|
|
@name 设置端口转发
|
|
@param port: 端口号
|
|
@param ip: ip地址
|
|
@param operation: 操作
|
|
@return None
|
|
'''
|
|
from firewallModel.app.iptables import Iptables
|
|
self.firewall = Iptables()
|
|
return self.firewall.port_forward(info, operation)
|
|
|
|
# 2024/3/25 下午 2:34 获取所有端口转发列表
|
|
def list_port_forward(self):
|
|
'''
|
|
@name 获取所有端口转发列表
|
|
@return None
|
|
'''
|
|
from firewallModel.app.iptables import Iptables
|
|
self.firewall = Iptables()
|
|
return self.firewall.get_nat_prerouting_rules()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
args = sys.argv
|
|
firewall = Ufw()
|
|
ufw_status = firewall.status()
|
|
if len(args) < 2:
|
|
print("Welcome to the UFW (Uncomplicated Firewall) command-line interface!")
|
|
print("Firewall status is :", ufw_status)
|
|
print("Firewall version: ", firewall.version())
|
|
if ufw_status == "not running":
|
|
print("ufw未启动,请启动ufw后再执行命令!")
|
|
print("启动命令: start")
|
|
print()
|
|
sys.exit(1)
|
|
print()
|
|
print("Available options:")
|
|
print("1. Check Firewall Status: status")
|
|
print("2. Check Firewall Version: version")
|
|
print("3. Start Firewall: start")
|
|
print("4. Stop Firewall: stop")
|
|
print("5. Restart Firewall: restart")
|
|
print("6. Reload Firewall: reload")
|
|
print("7. List All Ports: list_port")
|
|
print("8. List All IP Addresses: list_address")
|
|
print("9. Add Port: add_port <port> <protocol>")
|
|
print("10. Remove Port: remove_port <port> <protocol>")
|
|
print("11. Add Port Rule: add_port_rule <address> <port> <protocol> <strategy> <operation>")
|
|
print("12. Remove Port Rule: remove_port_rule <address> <port> <protocol> <strategy> <operation>")
|
|
print("13. Add IP Rule: add_ip_rule <address> <strategy> <operation>")
|
|
print("14. Remove IP Rule: remove_ip_rule <address> <strategy> <operation>")
|
|
print()
|
|
sys.exit(1)
|
|
if args[1] == "status":
|
|
print(firewall.status())
|
|
elif args[1] == "version":
|
|
print(firewall.version())
|
|
elif args[1] == "start":
|
|
error = firewall.start()
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Firewall started successfully.")
|
|
elif args[1] == "stop":
|
|
error = firewall.stop()
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Firewall stopped successfully.")
|
|
elif args[1] == "restart":
|
|
error = firewall.restart()
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Firewall restarted successfully.")
|
|
elif args[1] == "reload":
|
|
error = firewall.reload()
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Firewall reloaded successfully.")
|
|
elif args[1] == "list_input_port":
|
|
ports = firewall.list_input_port()
|
|
for p in ports:
|
|
print(p)
|
|
elif args[1] == "list_output_port":
|
|
ports = firewall.list_output_port()
|
|
for p in ports:
|
|
print(p)
|
|
elif args[1] == "list_input_address":
|
|
addresses = firewall.list_input_address()
|
|
for a in addresses:
|
|
print(a)
|
|
elif args[1] == "list_output_address":
|
|
addresses = firewall.list_output_address()
|
|
for a in addresses:
|
|
print(a)
|
|
elif args[1] == "add_port":
|
|
port = args[2]
|
|
protocol = args[3]
|
|
error = firewall.input_port(f"{port}/{protocol}", "allow")
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print(f"Port {port}/{protocol} added successfully.")
|
|
elif args[1] == "remove_port":
|
|
port = args[2]
|
|
protocol = args[3]
|
|
error = firewall.input_port(f"{port}/{protocol}", "remove")
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print(f"Port {port}/{protocol} removed successfully.")
|
|
elif args[1] == "add_port_rule":
|
|
address = args[2]
|
|
port = args[3]
|
|
protocol = args[4]
|
|
strategy = args[5]
|
|
operation = args[6]
|
|
error = firewall.rich_rules(
|
|
{"Address": address, "Port": port, "Protocol": protocol, "Strategy": strategy}, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Rich rule added successfully.")
|
|
elif args[1] == "remove_port_rule":
|
|
address = args[2]
|
|
port = args[3]
|
|
protocol = args[4]
|
|
strategy = args[5]
|
|
operation = args[6]
|
|
error = firewall.rich_rules(
|
|
{"Address": address, "Port": port, "Protocol": protocol, "Strategy": strategy}, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Rich rule removed successfully.")
|
|
elif args[1] == "add_ip_rule":
|
|
address = args[2]
|
|
strategy = args[3]
|
|
operation = args[4]
|
|
error = firewall.rich_rules(
|
|
{"Address": address, "Strategy": strategy}, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Rich rule added successfully.")
|
|
elif args[1] == "remove_ip_rule":
|
|
address = args[2]
|
|
strategy = args[3]
|
|
operation = args[4]
|
|
error = firewall.rich_rules(
|
|
{"Address": address, "Strategy": strategy}, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Rich rule removed successfully.")
|
|
elif args[1] == "output_port":
|
|
port = args[2]
|
|
operation = args[3]
|
|
error = firewall.output_port(port, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print(f"Output port {port} {operation} successfully.")
|
|
elif args[1] == "output_rich_rules":
|
|
address = args[2]
|
|
port = args[3]
|
|
protocol = args[4]
|
|
strategy = args[5]
|
|
operation = args[6]
|
|
error = firewall.output_rich_rules(
|
|
{"Address": address, "Port": port, "Protocol": protocol, "Strategy": strategy}, operation)
|
|
if error:
|
|
print(f"Error: {error}")
|
|
else:
|
|
print("Output rich rule added successfully.")
|
|
else:
|
|
print("Invalid args")
|
|
sys.exit(1)
|
|
|