|
|
在当今数字化信息时代，网络安全与系统性能的重要性日益凸显。Linux 操作系统以其开源、稳定、高效等诸多优势，成为众多服务器和网络设备的首选操作系统。然而，新部署的 Linux 系统往往默认开启了许多不必要的端口、服务和配置信息，这不仅会消耗大量系统资源，还会增加系统遭受攻击的风险，影响系统在高并发场景下的性能表现。
为了让 Linux 系统在保障安全的同时，能高效稳定地运行，我们可以借助自动化脚本与先进的豆包大模型，构建一套智能化的系统优化方案。下面为你详细介绍该方案的具体操作流程：
我们要编写一个自动化脚本。这个脚本的主要任务是与豆包大模型的 API 接口建立连接。在连接之前，脚本会收集当前 Linux 系统的详细信息，比如系统的版本、已安装的软件包、正在运行的服务、开放的端口等。这些信息就像是系统的 “健康档案”，能让豆包大模型全面了解系统的现状。
接着，脚本将收集到的系统信息通过 API 接口发送给豆包大模型。豆包大模型拥有强大的数据分析和决策能力，它就像一位经验丰富的系统专家，能够对这些信息进行深入分析。大模型会综合考虑系统的安全性、性能优化以及高并发场景下的运行需求，判断出哪些端口、服务和配置信息是不必要的，并给出具体的优化建议。
豆包大模型会将这些优化建议通过 API 接口返回给脚本。这些建议可能包括关闭某些不必要的服务、禁用一些默认开启但当前系统不需要的端口，以及修改一些不合理的配置参数等。
脚本会根据豆包大模型返回的建议，自动对 Linux 系统进行优化和初始化操作。对于不必要的服务，脚本会使用相应的命令将其停止并禁止开机自启，例如对于 CentOS 系统，可以使用systemctl stop和systemctl disable命令。对于不需要开放的端口，脚本会通过防火墙规则进行关闭，像使用iptables或firewalld来限制端口的访问。同时，脚本还会自动修改系统配置文件，移除或调整那些可能影响系统性能和安全的配置信息。
脚本会对优化后的系统进行简单的检查，确保各项操作都已正确执行，系统没有出现异常。并且，为了保证系统的稳定性和性能，脚本还会提供操作记录和日志，方便后续的查看和问题排查。
通过这样一套自动化的优化方案，我们可以充分利用豆包大模型的智能分析能力，快速、准确地找出 Linux 系统中存在的潜在问题，并及时进行优化和初始化。这不仅能有效提升系统的安全性，降低被攻击的风险，还能显著提高系统在高并发场景下的性能表现，为用户提供更加稳定、高效的服务。同时，这种自动化的方式大大减少了人工操作的繁琐和失误，提高了系统管理的效率和准确性。
此处将使用 Python 3 编写，因此需要确保你的服务器上已经安装了 Python 3。可以使用以下命令检查 Python 版本：
python3 --version
安装Python requests 库用于发送 HTTP 请求，是脚本中与模型 API 进行交互的关键库。可以使用以下命令安装：
mkdir -p ~/.pip && echo -e "[global]\nindex-url = https://pypi.tuna.tsinghua.edu.cn/simple\n[install]\ntrusted-host = pypi.tuna.tsinghua.edu.cn" > ~/.pip/pip.conf
pip install requestspip install psutil
编写Python脚本代码，如下所示：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import os
import json
import time
import threading
import sys
import subprocess
import re
import signal
# ===================== 七牛大模型配置 =====================
API_KEY = "sk-aab9187dbe9990b5bb92ee6c2945a5619d0f06b00831135d083a0a0a0e50005e"
API_URL = 'https://api.qnaigc.com/v1/chat/completions'
MODEL_NAME = "qwen-turbo"
# =========================================================
# 全局变量，用于控制进度提示线程
stop_progress = False
# 定义需要保留的服务和端口
ALLOWED_SERVICES = ['ssh', 'nginx', 'dbus', 'systemd-journald', 'polkit', 'NetworkManager', 'systemd-hostnamed', 'systemd-logind', 'systemd-udevd']
ALLOWED_PORTS = [22, 80, 443]
# 定义无效的服务名
INVALID_SERVICES = ['UNIT', 'LOAD', 'ACTIVE', 'SUB']
# 定义拒绝手动停止的服务
REFUSE_MANUAL_STOP_SERVICES = ['auditd']
# 定义要关闭的特定服务
SERVICES_TO_CLOSE = ['dhcpd', 'nfs', 'bind', 'vsftpd', 'named']
# 定义特定服务对应的端口
SERVICE_PORT_MAPPING = {
'dhcpd': [67, 68],
'nfs': [2049],
'bind': [53],
'vsftpd': [20, 21],
'named': [53]
}
def handle_ctrl_c(signum, frame):
print("\n\n🛑 检测到 Ctrl+C，程序正常退出！")
sys.exit(0)
signal.signal(signal.SIGINT, handle_ctrl_c)
# 新增：检查系统命令是否存在
def check_command_exists(cmd):
"""检查系统命令是否存在，避免执行不存在的命令报错"""
try:
subprocess.run([cmd, '--version'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return True
except FileNotFoundError:
print(f"\033[31m警告：系统缺少 {cmd} 命令，相关功能可能无法正常执行！\033[0m")
return False
except Exception:
return True
def get_current_services_and_ports():
"""获取当前正在运行的服务和监听的端口"""
try:
services = get_running_services()
ports = get_listening_ports()
return services, ports
except Exception as e:
print(f"获取服务和端口信息失败: {e}")
return [], []
def get_running_services():
"""获取当前正在运行的服务"""
if not check_command_exists('systemctl'):
return []
try:
result = subprocess.run(
['systemctl', 'list-units', '--type=service', '--state=running', '--no-pager', '--plain'],
capture_output=True, text=True, timeout=10
)
output = result.stdout
services = []
for line in output.splitlines():
parts = line.split()
if len(parts) > 0:
service = parts[0].replace('.service', '')
if service not in INVALID_SERVICES and service.strip():
services.append(service)
return list(set(services))
except subprocess.TimeoutExpired:
print("获取运行中服务超时")
return []
except Exception as e:
print(f"获取运行中服务失败: {e}")
return []
def get_listening_ports():
"""获取当前正在监听的端口"""
if not check_command_exists('ss'):
return []
try:
result = subprocess.run(
['ss', '-tuln', '--no-header'],
capture_output=True, text=True, timeout=10
)
output = result.stdout
ports = []
for line in output.splitlines():
parts = line.split()
if len(parts) > 4:
address_port = parts[4].split(':')
if len(address_port) > 1:
try:
port = int(address_port[-1])
ports.append(port)
except ValueError:
pass
return list(set(ports))
except subprocess.TimeoutExpired:
print("获取监听端口超时")
return []
except Exception as e:
print(f"获取监听端口失败: {e}")
return []
def progress_indicator():
"""进度转圈提示"""
global stop_progress
symbols = ['-', '\\', '|', '/']
index = 0
while not stop_progress:
symbol = symbols[index % len(symbols)]
sys.stdout.write(f"\r正在请求大模型分析无用服务与端口，请耐心等待... {symbol}")
sys.stdout.flush()
time.sleep(0.2)
index += 1
sys.stdout.write("\r请求完成，已接收AI优化建议\n")
def get_optimized_suggestions(services, ports):
"""调用AI获取关闭不必要服务和端口的建议"""
headers = {
'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'
}
prompt = f"""当前系统正在运行的服务有：{', '.join(services)}，正在监听的端口有：{', '.join(map(str, ports))}。
要求：
1. 仅返回可以关闭的不必要服务名和端口号，英文逗号分隔；
2. 必须排除保留服务：{', '.join(ALLOWED_SERVICES)} 和保留端口：{', '.join(map(str, ALLOWED_PORTS))}；
3. 只返回纯列表，不要任何解释、标题、注释；
4. 服务名不带.service后缀，端口仅输出数字。
示例输出：dhcpd,nfs,67,68"""
data = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": "Linux安全运维专家，严格遵守输出格式，只返回逗号分割的服务与端口列表。"},
{"role": "user", "content": prompt}
]
}
try:
print("正在向大模型请求无用服务/端口清理建议...")
global stop_progress
stop_progress = False
progress_thread = threading.Thread(target=progress_indicator)
progress_thread.start()
response = requests.post(API_URL, headers=headers, json=data, timeout=120)
stop_progress = True
progress_thread.join()
response.raise_for_status()
response_data = response.json()
content = response_data["choices"][0]["message"]["content"]
# 剥离代码块标记
block_reg = re.compile(r'```(?:txt|shell)?(.*?)```', re.DOTALL)
match = block_reg.search(content)
if match:
return match.group(1).strip()
return content.strip()
except requests.RequestException as e:
stop_progress = True
progress_thread.join()
print(f"\033[31mAPI请求出错: {e}\033[0m")
return None
except Exception as e:
stop_progress = True
progress_thread.join()
print(f"\033[31m解析响应异常: {e}\033[0m")
return None
def parse_suggestions(suggestions):
"""解析模型返回的建议"""
services_to_stop = []
ports_to_close = []
if not suggestions:
return services_to_stop, ports_to_close
items = suggestions.split(',')
for item in items:
item = item.strip().replace(' ', '')
if not item:
continue
if re.match(r'^\d+$', item):
try:
port = int(item)
if port not in ALLOWED_PORTS and 1 <= port <= 65535:
ports_to_close.append(port)
except ValueError:
pass
elif item not in INVALID_SERVICES and item not in ALLOWED_SERVICES and item not in REFUSE_MANUAL_STOP_SERVICES:
services_to_stop.append(item)
return list(set(services_to_stop)), list(set(ports_to_close))
def stop_unnecessary_services(services_to_stop):
"""停止不必要的服务"""
if not check_command_exists('systemctl'):
return [], []
stopped_services = []
skipped_services = []
devnull = open(os.devnull, 'w')
print(f"\n准备停止以下服务: {', '.join(services_to_stop)}")
for service in services_to_stop:
if service in ALLOWED_SERVICES:
print(f"跳过保留服务: {service}")
continue
if service in REFUSE_MANUAL_STOP_SERVICES:
print(f"拒绝停止受保护服务: {service}")
skipped_services.append(service)
continue
try:
check = subprocess.run(
['systemctl', 'is-active', f'{service}.service'],
capture_output=True, text=True, timeout=5
)
if check.returncode != 0:
continue
subprocess.run(
['systemctl', 'stop', f'{service}.service'],
stdout=devnull, stderr=devnull, timeout=10
)
subprocess.run(
['systemctl', 'disable', f'{service}.service'],
stdout=devnull, stderr=devnull, timeout=10
)
stopped_services.append(service)
print(f"✅ 成功停止并禁用服务: {service}")
except subprocess.TimeoutExpired:
print(f"⏱️ 停止服务 {service} 超时")
except Exception as e:
print(f"❌ 停止服务 {service} 失败: {e}")
devnull.close()
return stopped_services, skipped_services
def check_firewalld_status():
"""检查 firewalld 是否运行"""
if not check_command_exists('firewall-cmd'):
return False
try:
result = subprocess.run(
['systemctl', 'is-active', 'firewalld'],
capture_output=True, text=True, timeout=5
)
return result.stdout.strip() == 'active'
except Exception as e:
print(f"检查firewalld状态失败: {e}")
return False
def close_unnecessary_ports(ports_to_close, services_to_stop):
"""防火墙关闭多余端口"""
if not check_command_exists('firewall-cmd'):
print("缺少firewall-cmd命令，无法操作防火墙端口")
return []
closed_ports = []
if 'firewalld' in services_to_stop:
print("firewalld即将停止，跳过端口操作")
return closed_ports
if not check_firewalld_status():
devnull = open(os.devnull, 'w')
print("firewalld未启动，进行启动与开机自启配置")
subprocess.run(['systemctl', 'start', 'firewalld'], stdout=devnull, stderr=devnull, timeout=10)
subprocess.run(['systemctl', 'enable', 'firewalld'], stdout=devnull, stderr=devnull, timeout=10)
devnull.close()
devnull = open(os.devnull, 'w')
print(f"\n准备防火墙关闭端口: {', '.join(map(str, ports_to_close))}")
for port in ports_to_close:
if port in ALLOWED_PORTS:
print(f"跳过保留端口: {port}")
continue
try:
subprocess.run(['firewall-cmd', '--permanent', '--remove-port', f'{port}/tcp'], stdout=devnull, stderr=devnull)
subprocess.run(['firewall-cmd', '--permanent', '--remove-port', f'{port}/udp'], stdout=devnull, stderr=devnull)
subprocess.run(['firewall-cmd', '--reload'], stdout=devnull, stderr=devnull)
closed_ports.append(port)
print(f"✅ 防火墙移除端口: {port}")
except Exception as e:
print(f"❌ 关闭端口 {port} 失败: {e}")
devnull.close()
return closed_ports
def close_specific_services_and_ports():
"""强制清理预设危险服务"""
running_services = get_running_services()
stopped_services = []
closed_ports = []
devnull = open(os.devnull, 'w')
print(f"\n处理预设待关闭服务列表: {', '.join(SERVICES_TO_CLOSE)}")
for service in SERVICES_TO_CLOSE:
if service not in running_services:
print(f"服务 {service} 未运行，跳过")
continue
try:
subprocess.run(['systemctl', 'stop', f'{service}.service'], stdout=devnull, stderr=devnull, timeout=10)
subprocess.run(['systemctl', 'disable', f'{service}.service'], stdout=devnull, stderr=devnull, timeout=10)
stopped_services.append(service)
print(f"✅ 成功停止预设服务: {service}")
except Exception as e:
print(f"❌ 停止服务 {service} 失败: {e}")
if service in SERVICE_PORT_MAPPING:
for port in SERVICE_PORT_MAPPING[service]:
if port in ALLOWED_PORTS:
continue
try:
subprocess.run(['firewall-cmd', '--permanent', '--remove-port', f'{port}/tcp'], stdout=devnull, stderr=devnull)
subprocess.run(['firewall-cmd', '--permanent', '--remove-port', f'{port}/udp'], stdout=devnull, stderr=devnull)
subprocess.run(['firewall-cmd', '--reload'], stdout=devnull, stderr=devnull)
closed_ports.append(port)
print(f"✅ 防火墙移除预设端口: {port}")
except Exception as e:
print(f"❌ 端口 {port} 操作失败: {e}")
devnull.close()
return stopped_services, closed_ports
def summarize_results(
api_suggestions,
api_stopped_services,
api_skipped_services,
api_closed_ports,
specific_stopped_services,
specific_closed_ports
):
"""输出执行汇总信息"""
filtered_suggestions = []
if api_suggestions:
services_to_stop, ports_to_close = parse_suggestions(api_suggestions)
filtered_suggestions.extend(services_to_stop)
filtered_suggestions.extend(map(str, ports_to_close))
print("\n" + "="*60)
print("📋 系统服务与端口清理 执行结果汇总")
print("="*60)
print("1. AI分析建议（过滤后）：")
if filtered_suggestions:
print(f" → {', '.join(filtered_suggestions)}")
else:
print(" → 无有效清理建议")
print("\n2. 根据AI建议执行结果：")
if api_stopped_services:
print(f" ✅ 已停止服务：{', '.join(api_stopped_services)}")
else:
print(" - 未停止任何服务")
if api_skipped_services:
print(f" ⚠️ 受保护跳过服务：{', '.join(api_skipped_services)}")
if api_closed_ports:
print(f" ✅ 防火墙关闭端口：{', '.join(map(str, api_closed_ports))}")
else:
print(" - 未关闭任何端口")
print("\n3. 预设强制清理列表执行结果：")
if specific_stopped_services:
print(f" ✅ 已停止服务：{', '.join(specific_stopped_services)}")
else:
print(" - 无匹配运行服务")
if specific_closed_ports:
print(f" ✅ 防火墙关闭端口：{', '.join(map(str, specific_closed_ports))}")
else:
print(" - 无端口变更")
print("="*60)
def main():
print("================ Linux系统无用服务&端口安全清理工具 ================")
check_command_exists('systemctl')
check_command_exists('ss')
check_command_exists('firewall-cmd')
try:
services, ports = get_current_services_and_ports()
print(f"\n当前运行服务: {', '.join(services)}")
print(f"当前监听端口: {', '.join(map(str, ports))}")
suggestions = get_optimized_suggestions(services, ports)
api_stopped_services = []
api_skipped_services = []
api_closed_ports = []
specific_stopped_services = []
specific_closed_ports = []
if suggestions:
services_to_stop, ports_to_close = parse_suggestions(suggestions)
print(f"\nAI解析待处理服务：{', '.join(services_to_stop)}")
print(f"AI解析待处理端口：{', '.join(map(str, ports_to_close))}")
api_stopped_services, api_skipped_services = stop_unnecessary_services(services_to_stop)
api_closed_ports = close_unnecessary_ports(ports_to_close, services_to_stop)
specific_stopped_services, specific_closed_ports = close_specific_services_and_ports()
summarize_results(suggestions, api_stopped_services, api_skipped_services, api_closed_ports,
specific_stopped_services, specific_closed_ports)
except Exception as e:
print(f"\033[31m脚本执行异常: {e}\033[0m")
sys.exit(1)
print("\n🏁 全部任务执行完成！")
if __name__ == "__main__":
main()
使用豆包大模型需要登录火山引擎平台去创建应用、获取API秘钥，操作的方法和步骤如下：
#注册并登录火山引擎平台
访问火山引擎官网，点击右上角 “注册” 按钮，按照提示完成注册流程。注册方式可以选择手机号、邮箱等，完成注册后登录平台。
#开通大模型服务
登录后，在控制台页面，通过搜索框搜索 “大模型”，或者在产品与服务列表中找到 “大模型” 服务并进入。
若你是首次使用该服务，系统会提示你开通，按照引导完成开通操作。开通过程中可能需要进行实名认证等步骤，请根据系统提示完成。
#创建应用
进入大模型服务控制台后，通常在左侧导航栏中找到 “应用管理” 选项并点击进入。
#在应用管理页面，点击 “创建应用” 按钮。
填写应用信息：
应用名称：为你的应用取一个有意义的名称，方便识别和管理，例如 “Nginx 配置建议应用”。
#应用描述：简要描述该应用的用途，如 “用于根据 Nginx 并发连接数获取 worker_processes 建议值”。
选择模型：根据你的需求选择合适的大模型，例如可以选择豆包模型。
填写完成后，点击 “确定” 按钮，系统会创建一个新的应用。
#获取 API Key 和 Secret Key
应用创建成功后，在应用管理列表中找到你刚创建的应用，点击该应用名称进入应用详情页。
在应用详情页中，你可以看到 “API Key” 和 “Secret Key”，这两个密钥用于调用大模型 API 时进行身份验证。注意，Secret Key 只会在首次创建应用时显示一次，请妥善保存。
通过如上Python脚本+doubao大模型，即可实现自动调整Linux内核配置文件参数，脚本测试如图所示：python3 doubao_init_system.sh
 |
|
发表于