京峰教育-Linux技术论坛-大模型初始化Linux系统

admin 发表于 2026-10-14 09:37:48

在当今数字化信息时代，网络安全与系统性能的重要性日益凸显。Linux 操作系统以其开源、稳定、高效等诸多优势，成为众多服务器和网络设备的首选操作系统。然而，新部署的 Linux 系统往往默认开启了许多不必要的端口、服务和配置信息，这不仅会消耗大量系统资源，还会增加系统遭受攻击的风险，影响系统在高并发场景下的性能表现。 为了让 Linux 系统在保障安全的同时，能高效稳定地运行，我们可以借助自动化脚本与先进的豆包大模型，构建一套智能化的系统优化方案。下面为你详细介绍该方案的具体操作流程： 我们要编写一个自动化脚本。这个脚本的主要任务是与豆包大模型的 API 接口建立连接。在连接之前，脚本会收集当前 Linux 系统的详细信息，比如系统的版本、已安装的软件包、正在运行的服务、开放的端口等。这些信息就像是系统的 “健康档案”，能让豆包大模型全面了解系统的现状。 接着，脚本将收集到的系统信息通过 API 接口发送给豆包大模型。豆包大模型拥有强大的数据分析和决策能力，它就像一位经验丰富的系统专家，能够对这些信息进行深入分析。大模型会综合考虑系统的安全性、性能优化以及高并发场景下的运行需求，判断出哪些端口、服务和配置信息是不必要的，并给出具体的优化建议。 豆包大模型会将这些优化建议通过 API 接口返回给脚本。这些建议可能包括关闭某些不必要的服务、禁用一些默认开启但当前系统不需要的端口，以及修改一些不合理的配置参数等。 脚本会根据豆包大模型返回的建议，自动对 Linux 系统进行优化和初始化操作。对于不必要的服务，脚本会使用相应的命令将其停止并禁止开机自启，例如对于 CentOS 系统，可以使用systemctl stop和systemctl disable命令。对于不需要开放的端口，脚本会通过防火墙规则进行关闭，像使用iptables或firewalld来限制端口的访问。同时，脚本还会自动修改系统配置文件，移除或调整那些可能影响系统性能和安全的配置信息。 脚本会对优化后的系统进行简单的检查，确保各项操作都已正确执行，系统没有出现异常。并且，为了保证系统的稳定性和性能，脚本还会提供操作记录和日志，方便后续的查看和问题排查。 通过这样一套自动化的优化方案，我们可以充分利用豆包大模型的智能分析能力，快速、准确地找出 Linux 系统中存在的潜在问题，并及时进行优化和初始化。这不仅能有效提升系统的安全性，降低被攻击的风险，还能显著提高系统在高并发场景下的性能表现，为用户提供更加稳定、高效的服务。同时，这种自动化的方式大大减少了人工操作的繁琐和失误，提高了系统管理的效率和准确性。 此处将使用 Python 3 编写，因此需要确保你的服务器上已经安装了 Python 3。可以使用以下命令检查 Python 版本： python3 --version <img src="https://bbs.jfedu.net/jf_imgs/image1836.png" alt="" style="max-width:100%;"> 安装Python requests 库用于发送 HTTP 请求，是脚本中与模型 API 进行交互的关键库。可以使用以下命令安装： mkdir -p ~/.pip && echo -e "\nindex-url = https://pypi.tuna.tsinghua.edu.cn/simple\n\ntrusted-host = pypi.tuna.tsinghua.edu.cn" > ~/.pip/pip.conf <img src="https://bbs.jfedu.net/jf_imgs/image1837.png" alt="" style="max-width:100%;"> pip install requestspip install psutil <img src="https://bbs.jfedu.net/jf_imgs/image1838.png" alt="" style="max-width:100%;"> 编写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 = # 定义无效的服务名 INVALID_SERVICES = ['UNIT', 'LOAD', 'ACTIVE', 'SUB'] # 定义拒绝手动停止的服务 REFUSE_MANUAL_STOP_SERVICES = ['auditd'] # 定义要关闭的特定服务 SERVICES_TO_CLOSE = ['dhcpd', 'nfs', 'bind', 'vsftpd', 'named'] # 定义特定服务对应的端口 SERVICE_PORT_MAPPING = { 'dhcpd': , 'nfs': , 'bind': , 'vsftpd': , 'named': } 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(, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return True except FileNotFoundError: print(f"\033, [] 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.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.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 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"]["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 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: 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 只会在首次创建应用时显示一次，请妥善保存。 <img src="https://bbs.jfedu.net/jf_imgs/image1840.png" alt="" style="max-width:100%;"> <img src="https://bbs.jfedu.net/jf_imgs/image1841.png" alt="" style="max-width:100%;"> 通过如上Python脚本+doubao大模型，即可实现自动调整Linux内核配置文件参数，脚本测试如图所示：python3 doubao_init_system.sh <img src="https://bbs.jfedu.net/jf_imgs/image1848.png" alt="" style="max-width:100%;"> <img src="https://bbs.jfedu.net/jf_imgs/image1849.png" alt="" style="max-width:100%;">

页: [1]

京峰教育-Linux技术论坛's Archiver