解决动态ip访问宝塔面板的白名单设置问题

服务器使用宝塔面板，安全策略只开放了80 443 以及宝塔面板端口，客户端要持续向服务器提交数据，也就必须访问3306 mysql端口，宝塔面板提供白名单，可以将客户端ip写入白名单，这样就可以访问所有端口，非常方便！
但是，因为客户端网络经常不定时切换ip，新切换的ip并不在白名单内，无法访问3306端口，导致数据更新失败。
解决思路：让服务器定时检测客户端ip，发现变更自动写入白名单。
具体方法：首先在dynu.com注册个账号，下载安装客户端，这样就可以随时用固定的域名检测客户端ip是多少，然后在服务器端设置定时任务，每隔3分钟ping一下域名，将返回的ip写入白名单，因为宝塔白名单有唯一性，所以不用担心写入无数个同样的ip。
宝塔面板>9.6.0 ubuntu
python检测代码：
```python
#coding: utf-8
#free_firewall
import time, hashlib, sys, os, json, socket
import glob, logging

LOG_DIR = "/var/log/btwhitelist"
MAX_LOG_DAYS = 2
logging.basicConfig(
    filename='/var/log/btwhitelist/debug.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class bt_api:
    __BT_KEY = 'xxxxxxxxxxxxxxxxxxxxxx'
    __BT_PANEL = 'https://x.x.x.x:44415'

def __init__(self, bt_panel=None, bt_key=None):
        if bt_panel: 
            if not bt_panel.startswith(('http://', 'https://')):
                bt_panel = 'https://' + bt_panel
            self.__BT_PANEL = bt_panel
        if bt_key: 
            self.__BT_KEY = bt_key

def get_domain_ip(self, domain):
        try:
            return socket.gethostbyname(domain)
        except socket.gaierror:
            return None

def get_firewall_rules(self):
        # 修改为使用新的IP规则列表接口
        url = self.__BT_PANEL + '/firewall/com/ip_rules_list'
        p_data = self.__get_key_data()
        p_data['chain'] = "ALL"
        p_data['p'] = 1  # 添加分页参数
        p_data['row'] = 30  # 获取足够多的规则
        result = self.__http_post_cookie(url, p_data)
        return json.loads(result) if result else {"data": []}

def add_firewall_rule(self, ip):
        # 修改为使用新的设置IP规则接口
        url = self.__BT_PANEL + '/firewall/com/set_ip_rule'
        p_data = self.__get_key_data()
        # 构建符合新接口要求的参数
        p_data['types'] = 'accept'
        p_data['brief'] = 'API_ADD_HOME'
        p_data['address'] = ip
        p_data['chain'] = 'INPUT'  
        p_data['family'] = 'ipv4'  
        p_data['zone'] = 'public'  
        p_data['operation'] = 'add'
        p_data['strategy'] = 'accept'
        result = self.__http_post_cookie(url, p_data)
        try:
            return json.loads(result)
        except:
            return {"status": False, "msg": result}

def ensure_ip_access(self, domain):
        ip = self.get_domain_ip(domain)
        if not ip:
            return {"status": False, "msg": f"域名解析失败: {domain}"}
        
        rules = self.get_firewall_rules()
        # 检查IP是否已在白名单中
        existing_rules = [rule for rule in rules.get("data", []) 
                         if rule.get('address') == ip 
                         and rule.get('type') == 'accept'
                         and rule.get('port') == '0:65535'
                         and rule.get('protocol') == 'all']
        
        if existing_rules:
            return {"status": True, "msg": f"IP {ip} 已在白名单中（全端口开放）"}
        else:
            result = self.add_firewall_rule(ip)
            print("添加规则响应:", result)  # 调试输出
            if result.get("status") or "设置成功" in str(result):
                return {"status": True, "msg": f"已添加IP {ip} 到防火墙白名单"}
            else:
                return {"status": False, "msg": f"添加失败: {result.get('msg', '未知错误')}"}

def __get_md5(self, s):
        return hashlib.md5(s.encode('utf-8')).hexdigest()

def __get_key_data(self):
        now_time = int(time.time())
        token = self.__get_md5(str(now_time) + self.__get_md5(self.__BT_KEY))
        return {"request_token": token, "request_time": now_time}

def __http_post_cookie(self, url, p_data, timeout=10):
        cookie_file = f'./{self.__get_md5(self.__BT_PANEL)}.cookie'
        if sys.version_info[0] == 2:
            return "{}"  # Python2支持省略
        else:
            import urllib.request, ssl, http.cookiejar
            
            # 创建忽略SSL验证的上下文
            context = ssl._create_unverified_context()
            
            cookie_obj = http.cookiejar.MozillaCookieJar(cookie_file)
            if os.path.exists(cookie_file):
                cookie_obj.load(cookie_file, ignore_discard=True, ignore_expires=True)
            
            # 创建使用自定义上下文的HTTPS处理器
            https_handler = urllib.request.HTTPSHandler(context=context)
            
            # 创建包含cookie处理器和HTTPS处理器的opener
            handler = urllib.request.HTTPCookieProcessor(cookie_obj)
            opener = urllib.request.build_opener(handler, https_handler)
            
            # 准备请求数据
            data = urllib.parse.urlencode(p_data).encode('utf-8')
            req = urllib.request.Request(url, data)
            
            try:
                response = opener.open(req, timeout=timeout)
                result = response.read()
                if type(result) == bytes: 
                    result = result.decode('utf-8')
                
                # 保存更新后的cookie
                cookie_obj.save(ignore_discard=True, ignore_expires=True)
                return result
            except Exception as e:
                error_msg = f"API请求失败: {str(e)}"
                print(error_msg)  # 输出详细错误信息
                return json.dumps({"status": False, "msg": error_msg})

# 清理旧日志
def clean_old_logs():
    now = time.time()
    for log_file in glob.glob(f"{LOG_DIR}/*.log"):
        if os.stat(log_file).st_mtime < now - MAX_LOG_DAYS * 86400:
            try:
                os.remove(log_file)
                print(f"Deleted old log: {log_file}")
            except Exception as e:
                print(f"Error deleting {log_file}: {str(e)}")

# 执行主逻辑
if __name__ == '__main__':
    clean_old_logs()
    logging.info("脚本启动")
    domain = "*.*.com"
    api = bt_api()
    
    # 调试步骤1: 先获取当前防火墙规则
    print("获取防火墙规则...")
    rules = api.get_firewall_rules()
    print(f"当前规则数: {len(rules.get('data', []))}")
    print("=== 获取白名单 ===")
    addresses = [item["Address"] for item in rules["data"]]
    # 打印结果
    for addr in addresses:
        print(addr)
        
    # 调试步骤2: 获取域名IP
    ip = api.get_domain_ip(domain)
    logging.info(f"域名 {domain} 解析到IP: {ip}")
    print(f"域名 {domain} 解析到IP: {ip}")
    
    # 调试步骤3: 执行主要功能
    # result = api.ensure_ip_access(domain)

result = api.add_firewall_rule(ip)
    print("最终结果:", json.dumps(result, ensure_ascii=False, indent=2))
    logging.info(json.dumps(result, ensure_ascii=False, indent=2))
```
然后在宝塔面板计划任务中，增加
/usr/bin/python3 /root/btwhitelist.py
# 搞定！

解决动态ip访问宝塔面板的白名单设置问题

评论区

评论一下~

暂无评论，要不来一发？