CVE-2024-21176 是Oracle官方披露的安全漏洞（具体细节未公开），影响MySQL 5.7及更低版本。MySQL 5.7的生命周期已于2023年10月31日结束，官方不再提供免费安全补丁。即使5.7.44是5.7系列的最终版本，也未包含对此漏洞的修复。由于官方未提供针对5.7.44的直接补丁，只能通过更新到mysql8来进行修复。本文记录一次windows安装mysql8.4.4的过程，先在mysql官网上下载mysql8.4.4的windows安装包（https://downloads.mysql.com/archives/community/），这里我选择使用zip格式的压缩包文件。解压文件到D盘，得到目录D:\mysql-8.4.4-winx64，在D:\mysql-8.4.4-winx64中创建data和log目录。mkdir "D:\mysql-8.4.4-winx64\data_8_4" mkdir "D:\mysql-8.4.4-winx64\log" icacls "D:\mysql-8.4.4-winx64\data_8_4" /grant Everyone:F /T icacls "D:\mysql-8.4.4-winx64\log" /grant Everyone:F /T在D:\mysql-8.4.4-winx64中新建my.ini，编辑文件内容。[mysqld] basedir=D:/mysql-8.4.4-winx64 datadir=D:/mysql-8.4.4-winx64/data_8_4 port=3307 server-id=2 bind-address=0.0.0.0 # 正确的认证策略设置（使用逗号分隔） authentication_policy = 'caching_sha2_password,,' max_connections=200 character-set-server=utf8mb4 default-storage-engine=INNODBCMD进入D:\mysql-8.4.4-winx64\bin目录下，执行以下命令.\mysqld --defaults-file=D:\mysql-8.4.4-winx64\my.ini --initialize --console如果一切正常，会得到以下回显2025-06-13T03:15:54.664650Z 0 [System] [MY-015017] [Server] MySQL Server Initialization - start. 2025-06-13T03:15:54.668011Z 0 [System] [MY-013169] [Server] D:\mysql-8.4.4-winx64\bin\mysqld.exe (mysqld 8.4.4) initializing of server in progress as process 4172 2025-06-13T03:15:54.690289Z 1 [System] [MY-013576] [InnoDB] InnoDB initialization has started. 2025-06-13T03:15:55.053355Z 1 [System] [MY-013577] [InnoDB] InnoDB initialization has ended. 2025-06-13T03:15:56.784160Z 6 [Note] [MY-010454] [Server] A temporary password is generated for root@localhost: e_zdxfwjb0A_ 2025-06-13T03:15:58.748588Z 0 [System] [MY-015018] [Server] MySQL Server Initialization - end.如果报错提示缺少DLL文件，去微软官网下载VC++2015的版本库进行安装（https://www.microsoft.com/en-us/download/details.aspx?id=53587）若还是无法显示回显，则需要输出报错日志，根据日志实际情况进行调整。.\mysqld --defaults-file=D:\mysql-8.4.4-winx64\my.ini --initialize --console --log-error=D:\mysql-8.4.4-winx64\log\init.log数据库初始化成功后，创建MySQL84服务并进行启动。.\mysqld --install MySQL84 --defaults-file="D:\mysql-8.4.4-winx64\my.ini" net start MySQL84启动成功的话，命令行会回显MySQL84 服务正在启动 . MySQL84 服务已经启动成功。启动成功后即可使用mysql进行连接操作.\mysql -P3307 -uroot -p Enter password: ************ Welcome to the MySQL monitor. Commands end with ; or \g. Your MySQL connection id is 10 Server version: 8.4.4 Copyright (c) 2000, 2025, Oracle and/or its affiliates. Oracle is a registered trademark of Oracle Corporation and/or its affiliates. Other names may be trademarks of their respective owners. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. mysql>然后就可以在不影响原有的MySQL5.7.44版本的情况下愉快的使用MySQL8.4.4了。
用Python3实现一个简单的P2P文件和消息传输小工具
信令服务器搭建信令服务器搭建需要一个具有公网IP的服务器(VPS)，信令服务器的功能包括以下几点：节点注册 (/register/{peer_id}): 允许一个 P2P 节点（peer）注册其 ID 和它打算用于 P2P 通信的端口。服务器会自动记录请求来源的公网 IP 地址。节点查询 (/query/{peer_id}): 允许一个节点查询另一个已注册节点的信息（公网 IP 和 P2P 端口）。列出所有节点 (/list_peers): 显示当前所有已注册的节点信息（主要用于调试）。节点注销 (/unregister/{peer_id}): 允许节点从服务器注销。from fastapi import FastAPI, Request, HTTPException from pydantic import BaseModel # 用于数据验证和序列化 import uvicorn # ASGI 服务器 import time # 用于记录时间戳 from typing import Dict, Any, Optional # 初始化 FastAPI 应用 app = FastAPI( title="Simple P2P Signaling Server", description="A basic signaling server for P2P peer discovery.", version="0.1.0" ) # 使用 Pydantic 定义请求体模型 class PeerRegisterInfo(BaseModel): p2p_port: int # Peer 宣称其用于 P2P UDP 通信的本地端口 # 使用 Pydantic 定义响应体模型 class PeerDetailsResponse(BaseModel): peer_id: str public_ip: str # 服务器看到的该 Peer 的公网 IP p2p_port: int # Peer 注册的 P2P 端口 registered_at: float # 注册时的时间戳 # 简单的内存数据库来存储 Peer 信息 # 格式: { "peer_id": {"public_ip": "x.x.x.x", "p2p_port": 12345, "registered_at": timestamp} } peers_db: Dict[str, Dict[str, Any]] = {} @app.get("/") async def read_root(): """ 根路径，确认服务器正在运行。 """ return {"message": "P2P Signaling Server is running. Access /docs for API documentation."} @app.post("/register/{peer_id}", status_code=201, response_model=PeerDetailsResponse) async def register_peer(peer_id: str, peer_info: PeerRegisterInfo, request: Request): """ 注册或更新一个 Peer 的信息。 - `peer_id`: Peer 的唯一标识符。 - `peer_info`: 包含 `p2p_port`，即 Peer 用于 P2P 通信的端口。 服务器会自动从请求中获取 Peer 的公网 IP (`request.client.host`)。 """ client_host = request.client.host # 这是 FastAPI 服务器看到的客户端 IP 地址 # 注意: 如果服务器在反向代理后，需要配置代理以传递正确的客户端IP current_time = time.time() peers_db[peer_id] = { "public_ip": client_host, "p2p_port": peer_info.p2p_port, "registered_at": current_time } print(f"[INFO] Peer registered/updated: {peer_id} -> IP: {client_host}, P2P Port: {peer_info.p2p_port}") return PeerDetailsResponse( peer_id=peer_id, public_ip=client_host, p2p_port=peer_info.p2p_port, registered_at=current_time ) @app.get("/query/{peer_id}", response_model=PeerDetailsResponse) async def query_peer(peer_id: str): """ 查询指定 Peer ID 的注册信息。 """ if peer_id not in peers_db: print(f"[WARN] Peer query failed: {peer_id} not found.") raise HTTPException(status_code=404, detail=f"Peer '{peer_id}' not found") peer_data = peers_db[peer_id] print(f"[INFO] Peer query successful for: {peer_id} -> {peer_data}") return PeerDetailsResponse( peer_id=peer_id, public_ip=peer_data["public_ip"], p2p_port=peer_data["p2p_port"], registered_at=peer_data["registered_at"] ) @app.get("/list_peers") async def list_all_peers(): """ 列出所有当前已注册的 Peers 及其信息。 主要用于调试。 """ if not peers_db: return {"message": "No peers currently registered."} return peers_db @app.delete("/unregister/{peer_id}", status_code=200) async def unregister_peer(peer_id: str): """ 从服务器注销一个 Peer。 """ if peer_id in peers_db: del peers_db[peer_id] print(f"[INFO] Peer unregistered: {peer_id}") return {"message": "Peer unregistered successfully", "peer_id": peer_id} else: print(f"[WARN] Attempt to unregister non-existent peer: {peer_id}") raise HTTPException(status_code=404, detail=f"Peer '{peer_id}' not found, cannot unregister") # --- 如何运行此服务器 --- # 1. 将此代码保存为 `signaling_server.py`。 # 2. 安装必要的库: # pip install fastapi "uvicorn[standard]" pydantic # 3. 在终端中运行服务器: # uvicorn signaling_server:app --host 0.0.0.0 --port 8000 --reload # # - `uvicorn signaling_server:app`: 告诉 Uvicorn 运行 `signaling_server.py` 文件中的 `app` FastAPI 实例。 # - `--host 0.0.0.0`: 使服务器监听所有网络接口，这样局域网内或公网（如果端口已转发）的其他机器可以访问。 # - `--port 8000`: 指定服务器监听的端口。 # - `--reload`: (可选，用于开发) 当代码文件发生更改时，服务器会自动重新加载。 # # 你的 P2P 客户端程序随后可以向 `http://<服务器的IP地址>:8000` 发送 HTTP 请求。 # 例如: # - 注册 peer "alice" 在 P2P 端口 12345: # POST http://<server_ip>:8000/register/alice (Body: {"p2p_port": 12345}) # - 查询 peer "alice" 的信息: # GET http://<server_ip>:8000/query/alice if __name__ == "__main__": # 这部分通常不会直接运行，因为 FastAPI 应用推荐使用 Uvicorn 从命令行启动。 # 但为了完整性，这里可以添加一行来提示如何运行。 print("To run this FastAPI application, use the command:") print("uvicorn signaling_server:app --host 0.0.0.0 --port 8000 --reload") # uvicorn.run(app, host="0.0.0.0", port=8000) # 也可以取消注释这行来直接运行，但不推荐用于生产如何使用信令服务器如何使用这个信令服务器 (P2P 客户端的视角):Peer A 启动:Peer A 决定一个用于 P2P UDP 通信的本地端口 (例如，随机选择一个或用户指定一个，如 55555)。Peer A 向信令服务器发送一个 POST 请求到 /register/peerA，请求体为 {"p2p_port": 55555}。信令服务器收到请求，记录下 Peer A 的公网 IP (例如 1.2.3.4，这是服务器看到的 Peer A 的 IP) 和 P2P 端口 55555。Peer B 启动:Peer B 执行类似操作，注册其 P2P 端口 (例如 66666) 到 /register/peerB。信令服务器记录 Peer B 的公网 IP (例如 5.6.7.8) 和 P2P 端口 66666。Peer A 想要连接 Peer B:Peer A 向信令服务器发送一个 GET 请求到 /query/peerB。服务器响应 Peer B 的信息：{"public_ip": "5.6.7.8", "p2p_port": 66666, ...}。Peer B 想要连接 Peer A:Peer B 向信令服务器发送一个 GET 请求到 /query/peerA。服务器响应 Peer A 的信息：{"public_ip": "1.2.3.4", "p2p_port": 55555, ...}。尝试打洞:现在 Peer A 知道了 Peer B 的公网端点 (5.6.7.8:66666)。Peer B 知道了 Peer A 的公网端点 (1.2.3.4:55555)。双方现在可以尝试向对方的公网端点发送 UDP 包以进行 NAT 打洞。重要注意事项:极其简化: 这是一个非常基础的信令服务器。内存存储: 所有 Peer 数据都存储在内存中，服务器重启后数据会丢失。生产环境需要使用数据库（如 Redis, PostgreSQL, MongoDB 等）。无身份验证/授权: 任何人都可以注册或查询任何 Peer ID。生产环境需要安全机制。无错误恢复/重试: 客户端需要自己处理网络请求的失败和重试。无 Peer 超时: 过期的 Peer 信息会一直保留，除非手动注销。可以添加时间戳并定期清理不活跃的 Peer。公网 IP 地址 (request.client.host):request.client.host 获取的是直接连接到 FastAPI 服务器的客户端的 IP 地址。如果你的 FastAPI 服务器直接暴露在公网，这通常就是 Peer 的公网 IP。如果 FastAPI 服务器部署在反向代理（如 Nginx、Traefik）之后，你需要确保反向代理正确设置了 X-Forwarded-For 或类似的头部，并且 FastAPI (或其 ASGI 服务器 Uvicorn) 被配置为信任这些头部来获取真实的客户端 IP。否则，request.client.host 可能会是反向代理的 IP 地址。不处理 NAT 类型: 这个信令服务器不扮演 STUN 服务器的角色，它不帮助客户端检测其 NAT 类型。它只是记录和交换客户端报告的 P2P 端口以及服务器观察到的客户端公网 IP。HTTP 通信: 信令是通过 HTTP 进行的。P2P 客户端需要 HTTP 请求库（如 Python 的 requests 或 httpx）来与此服务器交互。客户端服务客户端需要实现以下基本功能：向信令服务器注册自己的 peer_id 和一个宣称的 P2P 通信端口。向信令服务器查询其他 peer_id 的网络信息。列出所有已注册的节点。从信令服务器注销自己。import requests import json import random import socket import threading import time import queue import os # 用于文件操作 import math # 用于计算块数 # --- 配置和全局变量 (与之前类似，增加文件传输相关) --- SIGNALING_SERVER_URL = "http://信令服务器IP地址:8000" current_peer_id = None current_p2p_port = None p2p_udp_socket = None p2p_target_peer_id = None p2p_target_reported_addr = None p2p_target_actual_addr = None # 打洞成功后对方的 (ip, port) p2p_listener_thread = None p2p_stop_listener = threading.Event() received_messages_q = queue.Queue() # 用于文本消息和控制消息 # 文件传输相关状态 CHUNK_SIZE = 8192 # bytes FILE_TRANSFER_TIMEOUT = 5 # 秒，等待ACK的超时 FILE_TRANSFER_RETRIES = 3 # 重传次数 # 使用字典来管理不同类型事件的Event对象，用于线程同步 ack_events = {} # key: chunk_sq_num, value: threading.Event() file_offer_response_event = threading.Event() # 用于等待对方接受/拒绝文件 file_offer_accepted = False # 标记对方是否接受文件 file_transfer_active = False # 标记当前是否有文件传输在进行 file_transfer_send_thread = None # 文件发送线程 # --- 信令服务器交互函数 (register_peer, query_peer, list_all_peers, unregister_peer) --- # (请从您之前的 p2p_client.py 中复制这些函数过来，确保 unregister_peer 会调用 stop_current_p2p_activity) # <在此处粘贴之前的信令函数> # --- 信令服务器交互函数 (与之前类似，这里省略，假设它们已存在) --- # register_peer, query_peer, list_all_peers, unregister_peer # (请从您之前的 p2p_client.py 中复制这些函数过来) # --- 新增：信令服务器交互函数 (确保它们在这里) --- def register_peer(peer_id: str, p2p_port: int): global current_peer_id, current_p2p_port url = f"{SIGNALING_SERVER_URL}/register/{peer_id}" payload = {"p2p_port": p2p_port} try: response = requests.post(url, json=payload) response.raise_for_status() data = response.json() print(f"[注册成功] 服务器响应: {data}") current_peer_id = peer_id current_p2p_port = p2p_port # 保存我们声明的P2P端口 return data except requests.exceptions.RequestException as e: print(f"[注册失败] 错误: {e}") if hasattr(e, 'response') and e.response is not None: try: print(f"服务器错误详情: {e.response.json()}") except json.JSONDecodeError: print(f"服务器错误详情 (非JSON): {e.response.text}") return None def query_peer(peer_id_to_query: str): if not peer_id_to_query: print("错误: 需要提供要查询的 peer_id。") return None url = f"{SIGNALING_SERVER_URL}/query/{peer_id_to_query}" try: response = requests.get(url) response.raise_for_status() data = response.json() # 我们不在这里打印，让调用者决定如何使用 # print(f"[查询成功] Peer '{peer_id_to_query}' 的信息: {data}") return data except requests.exceptions.RequestException as e: print(f"[查询失败] 错误: {e}") if hasattr(e, 'response') and e.response is not None: try: print(f"服务器错误详情: {e.response.json()}") except json.JSONDecodeError: print(f"服务器错误详情 (非JSON): {e.response.text}") return None def list_all_peers(): url = f"{SIGNALING_SERVER_URL}/list_peers" try: response = requests.get(url) response.raise_for_status() data = response.json() print("[列出所有 Peers] 服务器响应:") if isinstance(data, dict) and data.get("message") == "No peers currently registered.": print(" 目前没有节点注册。") elif isinstance(data, dict): for peer_id, info in data.items(): print(f" - {peer_id}: {info}") else: print(data) return data except requests.exceptions.RequestException as e: print(f"[列出 Peers 失败] 错误: {e}") return None def unregister_peer(peer_id_to_unregister: str): global current_peer_id, current_p2p_port if not peer_id_to_unregister: print("错误: 需要提供要注销的 peer_id。") return None stop_current_p2p_activity() # 注销前停止所有P2P活动 url = f"{SIGNALING_SERVER_URL}/unregister/{peer_id_to_unregister}" try: response = requests.delete(url) response.raise_for_status() data = response.json() print(f"[注销成功] 服务器响应: {data}") if peer_id_to_unregister == current_peer_id: current_peer_id = None current_p2p_port = None return data except requests.exceptions.RequestException as e: print(f"[注销失败] 错误: {e}") if hasattr(e, 'response') and e.response is not None: try: print(f"服务器错误详情: {e.response.json()}") except json.JSONDecodeError: print(f"服务器错误详情 (非JSON): {e.response.text}") return None # --- END 信令函数 --- def initialize_p2p_socket(local_port: int): global p2p_udp_socket if p2p_udp_socket: try: p2p_udp_socket.close() except: pass # 忽略关闭时的错误 p2p_udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: p2p_udp_socket.bind(('0.0.0.0', local_port)) p2p_udp_socket.settimeout(1.0) # 短超时用于打洞和ACK等待 print(f"[P2P] UDP Socket 已在 0.0.0.0:{local_port} 上初始化并绑定。") return True except OSError as e: print(f"[P2P] 错误: 绑定 UDP Socket 到端口 {local_port} 失败: {e}") p2p_udp_socket = None return False def send_p2p_control_message(sock, target_addr, message_type, sender_id, payload_dict=None): """发送P2P控制消息 (JSON格式)""" if not sock: return try: message = {"type": message_type, "sender_id": sender_id} if payload_dict: message.update(payload_dict) sock.sendto(json.dumps(message).encode('utf-8'), target_addr) # print(f"DEBUG: Sent control msg {message_type} to {target_addr}") except Exception as e: print(f"[P2P] 发送控制消息 {message_type} 失败: {e}") def send_p2p_data_chunk(sock, target_addr, sender_id, sq_num, data_bytes): """发送P2P文件数据块 (头部 + 原始数据)""" if not sock: return try: # 头部格式: "DATA:<sender_id>:<sq_num>:" 注意末尾的冒号作为分隔符 header = f"DATA:{sender_id}:{sq_num}:".encode('utf-8') packet = header + data_bytes sock.sendto(packet, target_addr) # print(f"DEBUG: Sent DATA chunk {sq_num} ({len(data_bytes)}B) to {target_addr}") except Exception as e: print(f"[P2P] 发送数据块 {sq_num} 失败: {e}") def udp_listener(): global p2p_udp_socket, p2p_stop_listener, received_messages_q, ack_events global file_offer_response_event, file_offer_accepted, file_transfer_active global p2p_target_actual_addr # 监听线程也可能在打洞阶段更新这个 print("[P2P Listener] 监听线程已启动。") if not p2p_udp_socket: print("[P2P Listener] 错误: Socket未初始化。") return active_file_receive_info = {} # 用于存储正在接收的文件的信息 while not p2p_stop_listener.is_set(): try: # raw_data, addr = p2p_udp_socket.recvfrom(CHUNK_SIZE + 256) # 缓冲区稍大于块大小以容纳头部 raw_data, addr = p2p_udp_socket.recvfrom(65536) # 确保能接收大的块 # 尝试解析为JSON (控制消息) try: message_str = raw_data.decode('utf-8') msg_data = json.loads(message_str) msg_type = msg_data.get("type") sender_id = msg_data.get("sender_id") # print(f"DEBUG Listener: Received JSON msg: {msg_data} from {addr}") if msg_type == "PUNCH_SYN" or msg_type == "PUNCH_ACK_RESPONSE": # 来自打洞过程 if not p2p_target_actual_addr or p2p_target_actual_addr[0] != addr[0]: print(f"[P2P Listener] 收到来自 {addr} 的打洞相关消息: {msg_type}") p2p_target_actual_addr = addr # 更新对方的实际地址 # 如果是SYN，回复ACK_RESPONSE if msg_type == "PUNCH_SYN": send_p2p_control_message(p2p_udp_socket, addr, "PUNCH_ACK_RESPONSE", current_peer_id) # 将打洞消息也放入队列，让主打洞逻辑感知 received_messages_q.put((msg_data, addr)) elif msg_type == "MSG": # 文本消息 received_messages_q.put((msg_data, addr)) elif msg_type == "FILE_OFFER": if file_transfer_active: print(f"[P2P Listener] 收到来自 {sender_id} 的文件提议，但当前正忙于另一传输，已拒绝。") send_p2p_control_message(p2p_udp_socket, addr, "FILE_REJECT", current_peer_id, {"filename": msg_data.get("filename"), "reason": "Receiver busy"}) continue print(f"\n[P2P Listener] 收到来自 {sender_id} ({addr}) 的文件提议: ") print(f" 文件名: {msg_data.get('filename')}") print(f" 大小: {msg_data.get('filesize')} bytes") print(f" 块大小: {msg_data.get('chunk_size')}") print(f" 总块数: {msg_data.get('total_chunks')}") user_choice = input("接受此文件吗? (y/N): ").strip().lower() if user_choice == 'y': file_transfer_active = True active_file_receive_info = { "filename": msg_data.get('filename'), "filesize": msg_data.get('filesize'), "chunk_size": msg_data.get('chunk_size'), "total_chunks": msg_data.get('total_chunks'), "sender_id": sender_id, "sender_addr": addr, "output_file": None, "received_chunks": set(), "next_expected_chunk": 0 } # 创建接收文件夹 receive_dir = "received_files_p2p" if not os.path.exists(receive_dir): os.makedirs(receive_dir) # 构建保存路径，避免路径遍历漏洞 (简单替换) safe_filename = os.path.basename(active_file_receive_info["filename"]) filepath = os.path.join(receive_dir, safe_filename) try: active_file_receive_info["output_file"] = open(filepath, "wb") print(f"[P2P Listener] 接受文件提议，将保存到 {filepath}") send_p2p_control_message(p2p_udp_socket, addr, "FILE_ACCEPT", current_peer_id, {"filename": active_file_receive_info["filename"]}) except IOError as e: print(f"[P2P Listener] 错误: 无法打开文件 {filepath} 进行写入: {e}") send_p2p_control_message(p2p_udp_socket, addr, "FILE_REJECT", current_peer_id, {"filename": active_file_receive_info["filename"], "reason": "Cannot write file"}) file_transfer_active = False active_file_receive_info = {} else: print(f"[P2P Listener] 拒绝文件提议: {msg_data.get('filename')}") send_p2p_control_message(p2p_udp_socket, addr, "FILE_REJECT", current_peer_id, {"filename": msg_data.get('filename')}) elif msg_type == "FILE_ACCEPT": if sender_id == p2p_target_peer_id: # 确保是目标peer的响应 print(f"[P2P Listener] {sender_id} 接受了文件 '{msg_data.get('filename')}'。") file_offer_accepted = True file_offer_response_event.set() # 通知发送方可以开始传输 elif msg_type == "FILE_REJECT": if sender_id == p2p_target_peer_id: print(f"[P2P Listener] {sender_id} 拒绝了文件 '{msg_data.get('filename')}'. 原因: {msg_data.get('reason')}") file_offer_accepted = False file_offer_response_event.set() elif msg_type == "ACK": # ACK:<receiver_id>:<chunk_sq_num> sq_num = msg_data.get("sq_num") if sq_num is not None and sender_id == p2p_target_peer_id: # print(f"DEBUG Listener: Received ACK for chunk {sq_num} from {sender_id}") if sq_num in ack_events: ack_events[sq_num].set() # 通知发送线程对应的ACK已收到 else: # print(f"DEBUG Listener: Received unexpected ACK for chunk {sq_num} or for a non-active transfer.") pass elif msg_type == "FILE_END": if file_transfer_active and active_file_receive_info and \ sender_id == active_file_receive_info["sender_id"] and \ msg_data.get("filename") == active_file_receive_info["filename"]: print(f"[P2P Listener] {sender_id} 完成发送文件 '{active_file_receive_info['filename']}'.") if active_file_receive_info["output_file"]: active_file_receive_info["output_file"].close() # 简单校验：块数是否匹配 if len(active_file_receive_info["received_chunks"]) == active_file_receive_info["total_chunks"]: print(f"[P2P Listener] 文件 '{active_file_receive_info['filename']}' 接收完毕且块数匹配。") send_p2p_control_message(p2p_udp_socket, active_file_receive_info["sender_addr"], "FILE_VERIFY", current_peer_id, {"filename": active_file_receive_info["filename"], "status": "OK"}) else: print(f"[P2P Listener] 警告: 文件 '{active_file_receive_info['filename']}' 接收完毕但块数不匹配! " f"收到 {len(active_file_receive_info['received_chunks'])}/{active_file_receive_info['total_chunks']}") send_p2p_control_message(p2p_udp_socket, active_file_receive_info["sender_addr"], "FILE_VERIFY", current_peer_id, {"filename": active_file_receive_info["filename"], "status": "ERROR_CHUNK_MISMATCH"}) file_transfer_active = False active_file_receive_info = {} elif msg_type == "FILE_VERIFY": if file_transfer_active and sender_id == p2p_target_peer_id: # 确保是发送方在等待这个 print(f"[P2P Listener] 收到来自 {sender_id} 的文件接收确认: '{msg_data.get('filename')}' 状态: {msg_data.get('status')}") # 这里可以设置一个Event来通知发送线程对方已完全接收并校验 # 例如: file_final_verify_event.set() # For now, just print. file_transfer_active = False # 发送方也标记传输结束 # 其他控制消息... except (json.JSONDecodeError, UnicodeDecodeError): # 不是JSON，尝试解析为数据块 header_part = raw_data.split(b':', 3) # DATA:sender:sq_num:actual_data if len(header_part) == 4 and header_part[0] == b'DATA': try: data_sender_id = header_part[1].decode('utf-8') data_sq_num = int(header_part[2].decode('utf-8')) actual_data = header_part[3] # print(f"DEBUG Listener: Received DATA chunk {data_sq_num} from {data_sender_id} ({len(actual_data)}B)") if file_transfer_active and active_file_receive_info and \ data_sender_id == active_file_receive_info["sender_id"] and \ addr[0] == active_file_receive_info["sender_addr"][0]: # 确保来自正确的发送者IP # 简单停止等待：只接受期望的下一个块 if data_sq_num == active_file_receive_info["next_expected_chunk"]: if active_file_receive_info["output_file"]: active_file_receive_info["output_file"].write(actual_data) active_file_receive_info["received_chunks"].add(data_sq_num) active_file_receive_info["next_expected_chunk"] += 1 # print(f" 接收并写入块 {data_sq_num}。发送ACK。") send_p2p_control_message(p2p_udp_socket, active_file_receive_info["sender_addr"], "ACK", current_peer_id, {"sq_num": data_sq_num}) # 打印进度 (简单版) progress = (len(active_file_receive_info["received_chunks"]) / active_file_receive_info["total_chunks"]) * 100 print(f"\r 接收进度: {progress:.2f}% ({len(active_file_receive_info['received_chunks'])}/{active_file_receive_info['total_chunks']})", end="") # 如果是重复的已接收块 (对方可能没收到我们的ACK)，重发ACK elif data_sq_num < active_file_receive_info["next_expected_chunk"] and \ data_sq_num in active_file_receive_info["received_chunks"]: # print(f" 收到重复块 {data_sq_num}。重发ACK。") send_p2p_control_message(p2p_udp_socket, active_file_receive_info["sender_addr"], "ACK", current_peer_id, {"sq_num": data_sq_num}) # else: (忽略乱序的块，在停止等待中) # print(f" 收到乱序块 {data_sq_num} (期望 {active_file_receive_info['next_expected_chunk']})。已忽略。") # else: (不是当前活动传输的数据块，忽略) # print(f" 收到来自 {data_sender_id} 的数据块，但当前无匹配的接收任务。") except Exception as e_parse: print(f"[P2P Listener] 解析DATA块头部失败: {e_parse}, raw_header: {header_part[:3]}") # else: (无法识别的原始数据包，忽略) # print(f"[P2P Listener] 收到来自 {addr} 的无法解析的原始数据: {raw_data[:50]}...") except socket.timeout: continue # 超时是正常的，继续检查 stop_event except OSError as e: if not p2p_stop_listener.is_set(): print(f"[P2P Listener] Socket错误: {e}") break except Exception as e: if not p2p_stop_listener.is_set(): print(f"[P2P Listener] 未知错误: {e}") break print("[P2P Listener] 监听线程已停止。") def attempt_udp_hole_punch(target_id: str, target_ip: str, target_port: int): global p2p_udp_socket, p2p_target_actual_addr, p2p_target_peer_id, p2p_target_reported_addr if not p2p_udp_socket: print("[P2P] 错误: 本地UDP Socket未初始化。") return False print(f"[P2P] 开始尝试与 {target_id} ({target_ip}:{target_port}) 进行UDP打洞...") p2p_target_peer_id = target_id p2p_target_reported_addr = (target_ip, target_port) p2p_target_actual_addr = None # 重置 # 发送几次打洞包 for i in range(3): #减少次数，依赖监听线程收到的PUNCH_SYN和回复的ACK_RESPONSE print(f" 发送打洞包 (PUNCH_SYN) 第 {i+1} 次到 {target_ip}:{target_port}") send_p2p_control_message(p2p_udp_socket, (target_ip, target_port), "PUNCH_SYN", current_peer_id) time.sleep(0.5) print(f"[P2P] 等待来自 {target_id} 的回应 (最多5秒)...") start_time = time.time() responded_correctly = False while time.time() - start_time < 5: try: msg_data, addr = received_messages_q.get_nowait() # 从监听线程队列获取 msg_type = msg_data.get("type") sender_id = msg_data.get("sender_id") if addr[0] == target_ip and sender_id == target_id: # 确保是目标peer的IP和ID if msg_type == "PUNCH_SYN" or msg_type == "PUNCH_ACK_RESPONSE": p2p_target_actual_addr = addr # 核心：使用对方实际发包的地址 print(f"[P2P] 收到来自 {sender_id} ({addr}) 的 {msg_type}。打洞可能成功！") print(f" 对方实际地址已更新为: {p2p_target_actual_addr}") if msg_type == "PUNCH_SYN": # 如果对方也发了SYN，我们回复ACK_RESPONSE send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "PUNCH_ACK_RESPONSE", current_peer_id) responded_correctly = True break except queue.Empty: time.sleep(0.2) continue except Exception as e: print(f" 处理打洞回应时出错: {e}") time.sleep(0.2) continue if responded_correctly and p2p_target_actual_addr: print(f"[P2P] 与 {target_id} 的UDP打洞过程完成。") return True else: print(f"[P2P] 未能从 {target_id} 收到有效回应，打洞可能失败。") p2p_target_actual_addr = None return False def _send_file_in_thread(filepath, filesize, chunk_size, total_chunks, target_addr, target_id_for_ack): global p2p_udp_socket, ack_events, file_transfer_active print(f"[文件发送线程] 开始发送 '{os.path.basename(filepath)}' ({total_chunks} 块)...") try: with open(filepath, "rb") as f: for i in range(total_chunks): if p2p_stop_listener.is_set(): # 检查是否被外部停止 print("[文件发送线程] 发送被中止。") break chunk_data = f.read(chunk_size) if not chunk_data: # 文件提前结束？不太可能如果filesize正确 print(f"[文件发送线程] 错误: 读取文件块 {i} 时数据为空。") break ack_events[i] = threading.Event() # 为当前块创建ACK事件 retries_left = FILE_TRANSFER_RETRIES chunk_sent_successfully = False while retries_left > 0: if p2p_stop_listener.is_set(): break # print(f" 发送块 {i}/{total_chunks-1} (尝试 {FILE_TRANSFER_RETRIES - retries_left + 1})...") send_p2p_data_chunk(p2p_udp_socket, target_addr, current_peer_id, i, chunk_data) # 等待ACK ack_received = ack_events[i].wait(timeout=FILE_TRANSFER_TIMEOUT) if ack_received: # print(f" 收到块 {i} 的ACK。") chunk_sent_successfully = True progress = ((i + 1) / total_chunks) * 100 print(f"\r 发送进度: {progress:.2f}% ({i+1}/{total_chunks})", end="") break else: print(f"\n 块 {i} ACK超时。剩余尝试 {retries_left - 1} 次。") retries_left -= 1 del ack_events[i] # 清理事件 if not chunk_sent_successfully: print(f"\n[文件发送线程] 块 {i} 发送失败，已达最大重试次数。中止传输。") file_transfer_active = False # 标记传输失败 return # 直接从线程退出 if chunk_sent_successfully: # 只有所有块都成功才发送FILE_END print(f"\n[文件发送线程] 所有 {total_chunks} 块已发送并确认。发送FILE_END。") send_p2p_control_message(p2p_udp_socket, target_addr, "FILE_END", current_peer_id, {"filename": os.path.basename(filepath)}) # 此处可以等待对方的 FILE_VERIFY 消息，但为简化，我们只打印 except IOError as e: print(f"[文件发送线程] 文件读写错误: {e}") except Exception as e: print(f"[文件发送线程] 未知错误: {e}") finally: print(f"[文件发送线程] '{os.path.basename(filepath)}' 发送处理完毕。") # file_transfer_active = False # 应该在收到FILE_VERIFY或超时后设置 def initiate_file_transfer(): global p2p_target_actual_addr, p2p_target_peer_id, p2p_udp_socket global file_offer_response_event, file_offer_accepted, file_transfer_active, file_transfer_send_thread if not p2p_target_actual_addr: print("错误: 未建立P2P连接，或对方实际地址未知。请先进行打洞。") return if file_transfer_active: print("错误: 当前已有文件传输正在进行中。") return filepath = input("请输入要发送的完整文件路径: ").strip() if not os.path.exists(filepath) or not os.path.isfile(filepath): print(f"错误: 文件 '{filepath}' 不存在或不是一个文件。") return try: filesize = os.path.getsize(filepath) filename = os.path.basename(filepath) total_chunks = math.ceil(filesize / CHUNK_SIZE) except Exception as e: print(f"获取文件信息失败: {e}") return print(f"准备发送文件: '{filename}', 大小: {filesize} bytes, 分为 {total_chunks} 块 (每块 {CHUNK_SIZE} B)。") # 重置事件和状态 file_offer_response_event.clear() file_offer_accepted = False # 发送FILE_OFFER send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "FILE_OFFER", current_peer_id, {"filename": filename, "filesize": filesize, "chunk_size": CHUNK_SIZE, "total_chunks": total_chunks}) print(f"已向 {p2p_target_peer_id} 发送文件提议，等待响应 (最多15秒)...") # 等待对方通过监听线程设置 file_offer_response_event offer_responded = file_offer_response_event.wait(timeout=15.0) if not offer_responded: print("对方未在规定时间内响应提议。") return if file_offer_accepted: print(f"{p2p_target_peer_id} 已接受文件。开始传输...") file_transfer_active = True # 标记传输开始 # 创建并启动文件发送线程 file_transfer_send_thread = threading.Thread(target=_send_file_in_thread, args=(filepath, filesize, CHUNK_SIZE, total_chunks, p2p_target_actual_addr, p2p_target_peer_id), daemon=True) file_transfer_send_thread.start() # 主线程返回菜单，发送在后台进行 print("文件正在后台发送。您可以在主菜单进行其他操作，或等待发送完成提示。") else: print(f"{p2p_target_peer_id} 拒绝了文件提议或发生错误。") def start_p2p_chat_and_file_menu(): global p2p_target_peer_id, p2p_target_reported_addr, p2p_target_actual_addr global p2p_listener_thread, p2p_stop_listener, received_messages_q, file_transfer_active if not current_peer_id or not current_p2p_port: print("请先注册您的 Peer ID 和 P2P 端口。") return # 如果当前已有P2P会话，先询问是否结束 if p2p_target_peer_id: print(f"当前已连接到 {p2p_target_peer_id} ({p2p_target_actual_addr}).") choice = input("是否要结束当前P2P会话并尝试连接新的Peer? (y/N): ").strip().lower() if choice == 'y': stop_current_p2p_activity() else: # 进入当前会话的菜单 pass # 下面的逻辑会处理 # 如果没有活动P2P目标，或者用户选择连接新的，则进行查询和打洞 if not p2p_target_actual_addr: # p2p_target_actual_addr 是打洞成功的标志 peer_to_connect = input("请输入您想连接的 Peer ID: ").strip() if not peer_to_connect: return if peer_to_connect == current_peer_id: print("不能连接自己。"); return print(f"正在查询 {peer_to_connect} 的信息...") peer_info = query_peer(peer_to_connect) if not peer_info: print(f"未能获取 {peer_to_connect} 的信息。"); return print(f"获取到 {peer_to_connect} 的信息: IP={peer_info['public_ip']}, P2P Port={peer_info['p2p_port']}") stop_current_p2p_activity() # 确保旧的已停止 p2p_stop_listener = threading.Event() # 重置 received_messages_q = queue.Queue() # 重置 if not initialize_p2p_socket(current_p2p_port): return p2p_listener_thread = threading.Thread(target=udp_listener, daemon=True) p2p_listener_thread.start() time.sleep(0.1) if not attempt_udp_hole_punch(peer_to_connect, peer_info['public_ip'], peer_info['p2p_port']): print(f"无法与 {peer_to_connect} 建立P2P连接。") stop_current_p2p_activity() return # --- P2P 会话菜单 --- print(f"\n--- P2P 会话已连接到 {p2p_target_peer_id} ({p2p_target_actual_addr}) ---") while p2p_target_actual_addr and not p2p_stop_listener.is_set(): # 只要连接有效 print("\nP2P 操作:") print("1. 发送文本消息") print("2. 发送文件") print("3. 返回主菜单 (结束当前P2P会话)") # 检查接收到的消息 (只显示文本消息，文件消息由listener处理) try: msg_data, addr = received_messages_q.get_nowait() if p2p_target_actual_addr and addr[0] == p2p_target_actual_addr[0]: msg_type = msg_data.get("type") sender_id = msg_data.get("sender_id") if msg_type == "MSG": text_msg = msg_data.get("text") print(f"\r[{time.strftime('%H:%M:%S')}] {sender_id}: {text_msg}\n> ", end="") # 其他类型的消息（如打洞消息）已在listener或打洞逻辑中处理，或在此忽略 else: # 来自未知源 print(f"\r收到来自未知源 {addr} 的P2P消息: {msg_data}\n> ", end="") except queue.Empty: pass except Exception as e: print(f"处理消息队列时出错: {e}") try: p2p_choice = input("> ").strip() if p2p_choice == '1': message_text = input("输入消息: ").strip() if message_text: send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "MSG", current_peer_id, {"text": message_text}) print(f"[{time.strftime('%H:%M:%S')}] 我: {message_text}") elif p2p_choice == '2': initiate_file_transfer() elif p2p_choice == '3': print("结束当前P2P会话...") send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "MSG", current_peer_id, {"text": "/quit_chat"}) # 通知对方 stop_current_p2p_activity() break # 退出P2P会话循环 else: if p2p_choice: # 如果输入了东西但不是有效选项 print("无效的P2P操作选项。") except KeyboardInterrupt: print("\n检测到 Ctrl+C，结束P2P会话...") if p2p_target_actual_addr: send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "MSG", current_peer_id, {"text": "/quit_chat_ctrl_c"}) stop_current_p2p_activity() break except EOFError: print("\n输入流结束，结束P2P会话...") if p2p_target_actual_addr: send_p2p_control_message(p2p_udp_socket, p2p_target_actual_addr, "MSG", current_peer_id, {"text": "/quit_chat_eof"}) stop_current_p2p_activity() break # 短暂休眠，让监听线程有机会处理消息并打印 # 避免input()完全阻塞消息的显示 (虽然不是完美的解决方案) time.sleep(0.1) def stop_current_p2p_activity(): global p2p_listener_thread, p2p_stop_listener, p2p_udp_socket global p2p_target_peer_id, p2p_target_actual_addr, p2p_target_reported_addr global file_transfer_active, file_transfer_send_thread, ack_events print("[系统] 正在停止所有P2P活动...") file_transfer_active = False # 停止任何进行中的文件传输标记 if p2p_stop_listener: p2p_stop_listener.set() # 发送停止信号 if file_transfer_send_thread and file_transfer_send_thread.is_alive(): print(" 等待文件发送线程结束...") file_transfer_send_thread.join(timeout=2.0) if file_transfer_send_thread.is_alive(): print(" 警告: 文件发送线程未能正常停止。") file_transfer_send_thread = None ack_events.clear() # 清空ACK事件 if p2p_listener_thread and p2p_listener_thread.is_alive(): print(" 等待监听线程结束...") p2p_listener_thread.join(timeout=2.0) if p2p_listener_thread.is_alive(): print(" 警告: 监听线程未能正常停止。") p2p_listener_thread = None if p2p_udp_socket: print(" 关闭UDP Socket。") try: p2p_udp_socket.close() except Exception as e_close: print(f" 关闭socket时出错: {e_close}") p2p_udp_socket = None # 清理队列中的剩余消息 (可选) while not received_messages_q.empty(): try: received_messages_q.get_nowait() except queue.Empty: break print("[系统] P2P活动已清理。") p2p_target_peer_id = None p2p_target_actual_addr = None p2p_target_reported_addr = None def main_menu(): # ... (主菜单逻辑与上一个版本类似，但选项3会调用 start_p2p_chat_and_file_menu) ... # 确保注销和退出时调用 stop_current_p2p_activity() global current_peer_id, current_p2p_port if not current_peer_id: # ... (注册逻辑，与之前相同) ... print("\n--- P2P 客户端 (未注册) ---") my_id = input("请输入你的 Peer ID (例如: alice, bob): ").strip() if not my_id: print("Peer ID 不能为空。") return True try: my_port_str = input(f"请输入你用于P2P的UDP端口 (50000-65000, 回车随机): ").strip() if not my_port_str: my_port = random.randint(50000, 65000) print(f"已随机生成P2P端口: {my_port}") else: my_port = int(my_port_str) if not (1024 < my_port < 65536): print("端口号不合法，将使用随机端口。") my_port = random.randint(50000, 65000) print(f"将使用P2P端口: {my_port}") except ValueError: print("无效的端口输入，将使用随机端口。") my_port = random.randint(50000, 65000) print(f"已随机生成P2P端口: {my_port}") if not register_peer(my_id, my_port): return True print(f"\n--- P2P 客户端 (ID: {current_peer_id}, P2P端口: {current_p2p_port}) ---") print("选择操作:") print("1. [信令] 查询其他 Peer 的信息") print("2. [信令] 列出所有已注册的 Peers") print("3. [P2P] 开始/管理 P2P 会话 (聊天/文件)") print("4. [信令] 注销我的 Peer ID") print("5. [信令] 重新注册 (使用新的ID或端口)") print("6. 退出") choice = input("请输入选项 (1-6): ").strip() if choice == '1': peer_to_query = input("请输入要查询的 Peer ID: ").strip() info = query_peer(peer_to_query) if info: print(f"Peer '{peer_to_query}' 的信息: {info}") elif choice == '2': list_all_peers() elif choice == '3': start_p2p_chat_and_file_menu() # 进入P2P会话管理 elif choice == '4': if current_peer_id: confirm = input(f"确定要注销 Peer ID '{current_peer_id}'吗? (y/N): ").strip().lower() if confirm == 'y': unregister_peer(current_peer_id) else: print("你还没有注册任何 Peer ID。") elif choice == '5': if current_peer_id: print(f"你当前注册为 '{current_peer_id}'。重新注册会使用新的ID和端口。") unregister_peer(current_peer_id) current_peer_id = None current_p2p_port = None elif choice == '6': print("正在退出...") if current_peer_id: unregister_peer(current_peer_id) return False else: print("无效选项，请重试。") return True if __name__ == "__main__": print("欢迎使用 P2P 客户端 Demo (包含P2P文本聊天和文件传输)!") # ... (与之前相同的服务器连接检查) ... print(f"将连接到信令服务器: {SIGNALING_SERVER_URL}") print("请确保信令服务器正在运行。") print("------------------------------------") try: ping_response = requests.get(SIGNALING_SERVER_URL, timeout=3) if ping_response.status_code == 200: print("成功连接到信令服务器。") else: print(f"警告: 信令服务器响应状态码 {ping_response.status_code}。") except requests.exceptions.ConnectionError: print(f"错误: 无法连接到信令服务器 {SIGNALING_SERVER_URL}。请检查服务器。") exit() except requests.exceptions.Timeout: print(f"错误: 连接信令服务器超时。") exit() try: while True: if not main_menu(): break except KeyboardInterrupt: print("\n检测到 Ctrl+C，正在退出程序...") finally: stop_current_p2p_activity() # 确保所有P2P资源被清理 print("客户端已退出。")其中SIGNALING_SERVER_URL全局变量需要修改为自己的信令服务器地址，端口默认为8000。
使用树莓派4B安装ELK Stack来收集和分析服务器日志
介绍ELK Stack：它由 Elasticsearch、Kibana、Beats 和 Logstash（也称为 ELK Stack）等组成。可靠、安全地从任何来源、任何格式获取数据，然后进行搜索、分析和可视化。ELK Stack是由三个开源工具组成的强大日志管理和分析平台，它们分别是Elasticsearch、Logstash和Kibana。这三个关键组件共同构成了一个完整的日志处理流程，从收集到存储，直到可视化展示，每个环节都是紧密相连的。ELK Stack官网前期准备树莓派4B一台(内存至少为4G)、TF内存卡(至少64G)、树莓派电源(5V3A)。系统采用树莓派官方的Raspberry Pi OS系统，刷机过程不多赘述。下载安装包首先，我们需要下载Elasticsearch、Logstash和Kibana的安装包,因树莓派是aarch64架构，可以下载aarch64版的deb包,本教程以7.17.21版本为例，为了更直观的展示，以下操作我均以root用户进行安装操作。wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.21-arm64.deb wget https://artifacts.elastic.co/downloads/kibana/kibana-7.17.21-arm64.deb wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.21-arm64.deb下载完成后得到以下三个文件。root@yuiit:~# ls elasticsearch-7.17.21-arm64.deb kibana-7.17.21-arm64.deb logstash-7.17.21-arm64.deb安装操作用apt-get安装这三个安装包。cd ~ apt install ./elasticsearch-7.17.21-arm64.deb apt install ./kibana-7.17.21-arm64.deb apt install ./logstash-7.17.21-arm64.deb调整配置Elasticsearch配置编辑elasticsearch的配置文件。先修改/etc/elasticsearch/jvm.options配置文件，修改里面的-Xms4g和-Xmx4g项，我使用的树莓派4B为4G内存版，所以我计划将这个参数改为1g。nano /etc/elasticsearch/jvm.options ------ ## 取消这两项的注释，并修改为以下内容 -Xms1g -Xmx1g ------编辑/etc/elasticsearch/elasticsearch.yml配置文件，在配置文件底部添加以下项。nano /etc/elasticsearch/elasticsearch.yml ------ cluster.name: my-pi-cluster node.name: pi-node network.host: 0.0.0.0 discovery.type: single-node xpack.security.enabled: true ------启动elasticsearch。systemctl start elasticsearch.service # 查看启动日志，启动过程需要一些时间，需要耐心等待 tail -fn 50 /var/log/elasticsearch/gc.log # 启动完成后日志会显示程序运行在9200端口，查看9200端口是否正在监听 root@yuiit:~# lsof -i:9200 COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME java 23118 elasticsearch 297u IPv6 165173 0t0 TCP *:9200 (LISTEN)elasticsearch启动完成后需要配置用户名密码，可使用elasticsearch-setup-passwords进行用户名密码的初始化。cd /usr/share/elasticsearch/bin ./elasticsearch-setup-passwords interactive这时会提示如下回显，输入y以进行确认，然后为各个用户设置密码，这里我们统一都设置为[Aa123456.]。Initiating the setup of passwords for reserved users elastic,apm_system,kibana,logstash_system,beats_system,remote_monitoring_user. You will be prompted to enter passwords as the process progresses. Please confirm that you would like to continue [y/N]y Enter password for [elastic]: Aa123456. Reenter password for [elastic]: Aa123456. Enter password for [apm_system]: Aa123456. Reenter password for [apm_system]: Aa123456. Enter password for [kibana_system]: Aa123456. Reenter password for [kibana_system]: Aa123456. Enter password for [logstash_system]: Aa123456. Reenter password for [logstash_system]: Aa123456. Enter password for [beats_system]: Aa123456. Reenter password for [beats_system]: Aa123456. Enter password for [remote_monitoring_user]: Aa123456. Reenter password for [remote_monitoring_user]: Aa123456. Changed password for user [apm_system] Changed password for user [kibana_system] Changed password for user [logstash_system] Changed password for user [beats_system] Changed password for user [remote_monitoring_user] Changed password for user [elastic]设置完成后重启elasticsearch。systemctl restart elasticsearch systemctl enable elasticsearch systemctl status elasticsearch漫长的重启完成后，进行连通性测试。若回显如下，则配置完成。root@yuiit:~# curl -u elastic:Aa123456. localhost:9200 { "name" : "pi-node", "cluster_name" : "my-pi-cluster", "cluster_uuid" : "96gwvTkrT2yoRihKAWCP4g", "version" : { "number" : "7.17.21", "build_flavor" : "default", "build_type" : "deb", "build_hash" : "d38e4b028f4a9784bb74de339ac1b877e2dbea6f", "build_date" : "2024-04-26T04:36:26.745220156Z", "build_snapshot" : false, "lucene_version" : "8.11.3", "minimum_wire_compatibility_version" : "6.8.0", "minimum_index_compatibility_version" : "6.0.0-beta1" }, "tagline" : "You Know, for Search" } Kibana配置修改kibana配置文件。修改配置文件中的内容。nano /etc/kibana/kibana.yml ------ server.host: "0.0.0.0" elasticsearch.hosts: ["http://localhost:9200"] elasticsearch.username: "kibana_system" elasticsearch.password: "Aa123456." i18n.locale: "zh-CN" #简体中文适配 ------启动kibana。systemctl start kibana systemctl enable kibana systemctl status kibana启动完成后可使用浏览器访问树莓派IP的5601端口，可进入kibana的web界面，正常情况下会进入登录界面。用户名：elastic密码：Aa123456.首次进入后选择直接进入系统，不添加集成，直接进入默认界面。配置Logstash首先修改/etc/logstash/jvm.options配置文件。将占用内存限制在512M(比较极限，但不是不能用)。nano /etc/logstash/jvm.options ------ -Xms512m -Xmx512m ------编辑logstash-sample.conf文件，内容如下，并给conf.d目录中也拷贝一份。cd /etc/logstash nano logstash-sample.conf ------ # Sample Logstash configuration for creating a simple # Beats -> Logstash -> Elasticsearch pipeline. input { beats { port => 5044 codec => json_lines } } output { elasticsearch { hosts => ["http://localhost:9200"] index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" user => "elastic" password => "Aa123456." } } ------ cat logstash-sample.conf > conf.d/logstash.conf此时logstash-sample.conf和conf.d/logstash.conf的配置是一致的。使用logstash进行配置文件校验，校验时间较长需要耐心等待，若回显内容包含Configuration OK，则说明配置文件正确。/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/logstash.conf --config.test_and_exit启动logstash。启动时间较长，可查看日志确认是否正常启动。systemctl start logstash systemctl enable logstash只到日志显示如下信息后，才算正常启动。root@yuiit:/etc/logstash# tail -f /var/log/logstash/logstash-plain.log [2025-04-30T10:10:55,115][WARN ][logstash.outputs.elasticsearch][main] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>7} [2025-04-30T10:10:55,366][INFO ][logstash.outputs.elasticsearch][main] Config is not compliant with data streams. `data_stream => auto` resolved to `false` [2025-04-30T10:10:55,375][INFO ][logstash.outputs.elasticsearch][main] Config is not compliant with data streams. `data_stream => auto` resolved to `false` [2025-04-30T10:10:55,614][INFO ][logstash.outputs.elasticsearch][main] Using a default mapping template {:es_version=>7, :ecs_compatibility=>:disabled} [2025-04-30T10:10:56,397][INFO ][logstash.javapipeline ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>500, "pipeline.sources"=>["/etc/logstash/conf.d/logstash.conf"], :thread=>"#<Thread:0xa2468a2 run>"} [2025-04-30T10:11:00,626][INFO ][logstash.javapipeline ][main] Pipeline Java execution initialization time {"seconds"=>4.22} [2025-04-30T10:11:00,706][INFO ][logstash.inputs.beats ][main] Starting input listener {:address=>"0.0.0.0:5044"} [2025-04-30T10:11:00,753][INFO ][logstash.javapipeline ][main] Pipeline started {"pipeline.id"=>"main"} [2025-04-30T10:11:00,975][INFO ][logstash.agent ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]} [2025-04-30T10:11:01,190][INFO ][org.logstash.beats.Server][main][2e5bd4de81fe2e8bda606b1da25ec416498a3651dd5bb91041667e8b03f5be2c] Starting server on port: 5044配置完以上三项内容后，理论上kibana/Logstash已经和Elasticsearch接通成功。检查服务情况进入kibana界面，在左侧菜单的Management找到堆栈监测，点击"Or, set up with self monitoring"，再点击"Turn on monitoring"，转圈完成后可以看到集群概览界面。(我这里Health为黄色警告，显示缺少副本分片，可能是因为分配内存过少？不过不影响正常写入和查询。)Logstash规则配置LogStash可以根据配置文件中的规则，对日志中的字段进行匹配并将匹配后的信息存入Elasticsearch。以下是一个简单的Nginx的规则配置，我尝试做了一些SQL注入之类的识别规则，比较不完善仅为参考之用。nano /etc/logstash/conf.d/logstash.conf ------ # Sample Logstash configuration for creating a simple # Beats -> Logstash -> Elasticsearch pipeline. input { beats { port => 5044 } } filter { # 1. 使用 Grok 解析 Nginx 日志格式 grok { match => { "message" => '%{IP:client_ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:http_method} %{URIPATHPARAM:http_url} HTTP/%{NUMBER:http_version}" %{NUMBER:status_code} %{NUMBER:body_bytes_sent} "%{DATA:http_referer}" "%{DATA:user_agent}"' } remove_field => ["message"] # 可选：移除原始日志以减少存储 } date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] target => "@timestamp" } # SQL注入匹配规则 if "select" in [http_url] or "union" in [http_url] or "insert" in [http_url] or "update" in [http_url] or "delete" in [http_url] or "drop" in [http_url] or "--" in [http_url] or "%27" in [http_url] or "%3B" in [http_url] or "%2F*" in [http_url] or "*%2F" in [http_url] or "benchmark" in [http_url] or "sleep" in [http_url] or "waitfor" in [http_url] or "pg_sleep" in [http_url] or "select" in [http_referer] or "union" in [http_referer] or "insert" in [http_referer] or "update" in [http_referer] or "delete" in [http_referer] or "drop" in [http_referer] or "--" in [http_referer] or "%27" in [http_referer] or "%3B" in [http_referer] or "%2F*" in [http_referer] or "*%2F" in [http_referer] or "benchmark" in [http_referer] or "sleep" in [http_referer] or "waitfor" in [http_referer] or "pg_sleep" in [http_referer] { mutate { add_field => { "threat_type" => "SQL注入" } add_field => { "threat_id" => "1"} } } # 404状态码捕获 if "404" in [status_code] { mutate { add_field => { "threat_type" => "404未找到资源" } add_field => { "threat_id" => "2" } } } # 50X状态码捕获 if "500" in [status_code] or "501" in [status_code] or "502" in [status_code] or "503" in [status_code] or "504" in [status_code] { mutate { add_field => { "threat_type" => "50X服务器内部错误" } add_field => { "threat_id" => "3" } } } # 检查目录遍历攻击模式 if "..." in [http_url] or "../" in [http_url] or ".env" in [http_url] or "/etc/passwd" in [http_url] or "admin.php" in [http_url] or "config.php" in [http_url] or "php://input" in [http_url] or "php://filter" in [http_url] { mutate { add_field => { "threat_type" => "疑似目录遍历攻击" } add_field => { "threat_id" => "4" } } } # 检查敏感文件访问 if ".env" in [http_url] or "/etc/passwd" in [http_url] or "/etc/shadow" in [http_url] or "/root/" in [http_url] or "config.php" in [http_url] or "admin.php" in [http_url] or "mysql.sock" in [http_url] or "wp-config.php" in [http_url] { mutate { add_field => { "threat_type" => "疑似敏感文件访问" } add_field => { "threat_id" => "5" } } } # 恶意扫描工具 User-Agent if "sqlmap" in [user_agent] or "nmap" in [user_agent] or "wget" in [user_agent] or "curl" in [user_agent] or "zgrab" in [user_agent] or "nikto" in [user_agent] or "burpsuite" in [user_agent] { mutate { add_field => { "threat_type" => "疑似恶意扫描" } add_field => { "threat_id" => "6" } } } # Webshell 特征检测 if ".php?" in [http_url] and ( "cmd=" in [http_url] or "eval=" in [http_url] or "system=" in [http_url] or "shell=" in [http_url] or "exec=" in [http_url] or "passthru=" in [http_url] or "popen=" in [http_url] or "fpassthru=" in [http_url] ) { mutate { add_field => { "threat_type" => "疑似Webshell攻击" } add_field => { "threat_id" => "7" } } } # 挖矿池域名访问 if "xmrpool" in [http_url] or "monero" in [http_url] or "minexmr" in [http_url] or "nanopool" in [http_url] { mutate { add_field => { "threat_type" => "疑似挖矿病毒" } add_field => { "threat_id" => "8" } } } # 爬虫访问检测 if "googlebot" in [user_agent] or "baidu" in [user_agent] or "bingbot" in [user_agent] or "yandex" in [user_agent] or "slurp" in [user_agent] or "duckduckbot" in [user_agent] or "archive.org" in [user_agent] or "spider" in [user_agent] or "crawl" in [user_agent] or "bot" in [user_agent] or "crawlbot" in [user_agent] or "facebookexternalhit" in [user_agent] { mutate { add_field => { "threat_type" => "疑似爬虫访问" } add_field => { "threat_id" => "9" } } } } output { elasticsearch { hosts => ["http://localhost:9200"] index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" user => "elastic" password => "Aa123456." } } ------配置完成后重启Logstash生效。安装FileBeat接下来在节点服务器配置FileBeat服务，FileBeat可将服务器内的日志收集好并传给LogStash，由LogStash进行分析后写入Elasticsearch。下载FileBeat程序文件，这里的被监控端我们以Debian12为例。cd ~ wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.17.21-amd64.deb apt install ./filebeat-7.17.21-amd64.deb安装完成后编辑filebeat配置文件/etc/filebeat/filebeat.yml。修改部分参数，假设nginx日志在/var/log/nginx目录下。将output.elasticsearch相关的连接配置信息全部注释，改为配置output.logstash项。nano /etc/filebeat/filebeat.yml ------ - type: filestream id: filebeat-01 paths: - /var/log/nginx/*.log output.logstash: # The Logstash hosts hosts: ["树莓派的IP:5044"] # Optional SSL. By default is off. # List of root certificates for HTTPS server verifications #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"] # Certificate for SSL client authentication #ssl.certificate: "/etc/pki/client/cert.pem" # Client Certificate Key #ssl.key: "/etc/pki/client/cert.key" ## 将以下内容全部注释 #output.elasticsearch: # Array of hosts to connect to. #hosts: ["localhost:9200"] # Protocol - either `http` (default) or `https`. #protocol: "https" # Authentication credentials - either API key or username/password. #api_key: "id:api_key" #username: "elastic" #password: "changeme" ------启动filebeat。systemctl start filebeat systemctl enable filebeat systemctl status filebeat显示如下，启动成功。root@filebeat:/etc/filebeat# systemctl status filebeat.service * filebeat.service - Filebeat sends log files to Logstash or directly to Elasticsearch. Loaded: loaded (/lib/systemd/system/filebeat.service; enabled; preset: enabled) Active: active (running) since Wed 2025-04-30 02:11:11 UTC; 7h ago Docs: https://www.elastic.co/beats/filebeat Main PID: 4273 (filebeat) Tasks: 7 (limit: 38092) Memory: 32.7M CPU: 8.794s CGroup: /system.slice/filebeat.service `-4273 /usr/share/filebeat/bin/filebeat --environment systemd -c /etc/filebeat/filebeat.yml --path.home /usr/share/filebeat --path.config /etc/filebeat --path.data /var/lib> Apr 30 09:52:24 filebeat filebeat[4273]: 2025-04-30T09:52:24.354Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/panel-assess-access.log"> Apr 30 09:52:34 filebeat filebeat[4273]: 2025-04-30T09:52:34.354Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/panel-assess-access.log"> Apr 30 09:52:34 filebeat filebeat[4273]: 2025-04-30T09:52:34.354Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/uptime-assess-access.log> Apr 30 09:52:34 filebeat filebeat[4273]: 2025-04-30T09:52:34.354Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/panel-assess-errors.log"> Apr 30 09:52:34 filebeat filebeat[4273]: 2025-04-30T09:52:34.354Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/uptime-assess-errors.log> Apr 30 09:52:44 filebeat filebeat[4273]: 2025-04-30T09:52:44.358Z INFO [monitoring] log/log.go:184 Non-zero metrics in the last 30s {"monitoring": {"m> Apr 30 09:52:44 filebeat filebeat[4273]: 2025-04-30T09:52:44.358Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/panel-assess-access.log"> Apr 30 09:52:44 filebeat filebeat[4273]: 2025-04-30T09:52:44.358Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/panel-assess-errors.log"> Apr 30 09:52:44 filebeat filebeat[4273]: 2025-04-30T09:52:44.358Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/uptime-assess-errors.log> Apr 30 09:52:44 filebeat filebeat[4273]: 2025-04-30T09:52:44.358Z WARN [file_watcher] filestream/fswatch.go:204 file "/var/log/nginx/uptime-assess-access.log> lines 1-21/21 (END) 创建索引模式进入kibana界面，在左侧菜单中进入Discover，此时这里可能是空的，需要创建索引模式，点击右上方选项，进入"View Discover settings"，在左侧Kibana项中进入索引模式，默认显示的索引模式应该包含 "metrics-" 和 "logs-" ，在右上角创建索引模式，此时就可以看到右侧多出了filebeat推送过来的索引信息。创建完成后，在左侧索引管理中可以看到引用进来的索引表信息。构建面板图表索引配置完成后进入Discover中，选择filebeat-*，就可看到所有传输进来的Nginx日志。在Overview中创建仪表板，依照自己的需求创建仪表板结构并指定图表中的字段，最终简单实现的效果如下。
GrayLog是目前主流的日志管理软件之一，用户可使用该软件查看和分析实时日志，今天以Docker的方式部署一个GrayLog系统试试看。安装环境docker的安装方式不多赘述，之前的教程有Docker-Web虚拟化环境使用。根据官方文档，GrayLog的必要环境为以下几点： Java(>=8) Elasticsearch(6.x或7.x) MongoDB(4.0、4.2或4.4)我们去DockerHub找对应版本并进行安装：docker pull mongo:4.2 docker pull elasticsearch:6.8.19 docker pull graylog/graylog:4.1启动环境启动容器docker run -idt --name mongo mongo:4.2 docker run -idt --name elasticsearch elasticsearch:6.8.19 docker run --link mongo --link elasticsearch -p 9000:9000 -p 12201:12201 -p 1514:1514 -p 5555:5555 -p 5140-5180:5140-5180 -p 5140-5180:5140-5180/udp -e GRAYLOG_HTTP_EXTERNAL_URI="http://127.0.0.1:9000/" -e GRAYLOG_PASSWORD_SECRET=<接口token> -e GRAYLOG_ROOT_PASSWORD_SHA2=<admin密码> --name graylog -itd graylog/graylog:4.1其中admin密码生成和token生成需要用到以下命令# 生成token pwgen -N 1 -s 96 # admin密码生成 echo -n "Enter Password: " && head -1 </dev/stdin | tr -d '\n' | sha256sum | cut -d" " -f1调试打开docker日志调试容器docker logs -f -t --tail 100 graylog访问http://IP:9000，输入用户名密码登录系统对接Nginx编辑Nginx配置文件vim /etc/nginx/nginx.conf # 在http中添加以下内容 log_format main '$remote_addr - $remote_user [$time_local] "$request" ' '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"'; log_format graylog2_json escape=json '{ "timestamp": "$time_iso8601", ' '"remote_addr": "$remote_addr", ' '"remote_user": "$remote_user", ' '"body_bytes_sent": $body_bytes_sent, ' '"request_time": $request_time, ' '"status": $status, ' '"request": "$request", ' '"request_method": "$request_method", ' '"host": "$host",' '"upstream_cache_status": "$upstream_cache_status",' '"upstream_addr": "$upstream_addr",' '"http_x_forwarded_for": "$http_x_forwarded_for",' '"http_referrer": "$http_referer", ' '"http_user_agent": "$http_user_agent" }'; access_log syslog:server=127.0.0.1:5141,facility=local0,tag=nginx,severity=info graylog2_json;用tcpdump查看是否有数据推送tcpdump udp -n -vv port 5141 -i lo -X打开GrayLog->System->InputsSelect Input中选择Raw/Plaintext UDP，Launch new input，勾选Global，port为Nginx中配置的5141，填写完成后Save保存，Network IO出现数字是即为对接成功。对接过程中可能会遇到recvBufferSize不匹配的情况，需要查看graylog日志进行调试，将Receive Buffer Size调整到合适的数字。
Docker-Web环境搭建安装Docker容器yum install -y yum-utils # 设置稳定存储库 yum-config-manager \ --add-repo \ https://download.docker.com/linux/centos/docker-ce.repo # 启用nightly存储库 yum-config-manager --enable docker-ce-nightly # 安装Docker引擎 yum install docker-ce docker-ce-cli containerd.io启动/停止Docker# 对docker进行启停操作 ## 启动 systemctl start docker ## 停止 systemctl stop docker ## 开机启动 systemctl enable docker ## 禁止开机启动 systemctl disable docker ## 查看状态 systemctl status docker安装portainerdocker search portainer # 拉取镜像 docker pull portainer/portainer-ce镜像下载完成[root@hyui ~]# docker images REPOSITORY TAG IMAGE ID CREATED SIZE portainer/portainer-ce latest a0a227bf03dd 5 weeks ago 196MB启动portainerdocker run \ --name portainer \ -d -p 9000:9000 \ -v "/var/run/docker.sock:/var/run/docker.sock" \ portainer/portainer-ce访问http://{IP}:9000，通过Web界面对Docker进行控制。关于firewalld防火墙Docker需要在firewalld之后启动，因为firewalld运行在IPTABLES之上，firewalld在重启时会从IPTABLES中移除DOCKER规则解决方法：Docker容器使用host网络模式启动关闭firewalld防火墙改用iptables关闭CentOS7的firewalld和iptables，改用主机商提供的防火墙(如果有的话)
LNMP安装安装Nginx下载Nginx预编译包https://nginx.org/en/download.html解压压缩包[root@Hyui-VM ~]# mkdir nginx [root@Hyui-VM ~]# mv nginx-1.19.0.tar.gz ./nginx [root@Hyui-VM ~]# cd nginx/ [root@Hyui-VM nginx]# tar -zxvf nginx-1.19.0.tar.gz [root@Hyui-VM nginx]# ls nginx-1.19.0 nginx-1.19.0.tar.gz [root@Hyui-VM nginx]# cd nginx-1.19.0 [root@Hyui-VM nginx-1.19.0]# ls auto CHANGES CHANGES.ru conf configure contrib html LICENSE man README src安装依赖软件yum install gcc gcc-c++ cyrus-sasl-md5 openssl openssl-devel -y编译安装# 将ssl模块也编译入nginx ./configure --prefix=/usr/local/nginx/ --with-http_ssl_module make && make install运行Nginx./sbin/nginx ps -aux | grep nginx root 17033 0.0 0.0 45976 1120 ? Ss 16:30 0:00 nginx: master process ./sbin/nginx nobody 17034 0.0 0.0 46416 2112 ? S 16:30 0:00 nginx: worker process root 17060 0.0 0.0 112812 972 pts/1 S+ 16:33 0:00 grep --color=auto nginx将Nginx放入systemdvim /usr/lib/systemd/system/nginx.service [Unit] Description=Nginx # 描述Nginx信息 Documentation=http://nginx.org/ After=network.target remote-fs.target nss-lookup.target # 表明依赖服务 [Service] Type=forking # 启动时关系的定义 PIDFile=/usr/local/nginx/logs/nginx.pid # PID文件 ExecStart=/usr/local/nginx/sbin/nginx -c /usr/local/nginx/conf/nginx.conf # 启动Nginx使用的命令 ExecReload=/bin/kill -s HUP $MAINPID # 重启Nginx命令 ExecStop=/bin/kill -s QUIT $MAINPID # 停止Nginx命令 PrivateTmp=true # 给服务分配独立的临时空间 [Install] WantedBy=multi-user.target安装PHP下载预编译包https://www.php.net/downloads添加www用户groupadd www useradd -s /sbin/nologin -g www -M www解压预编译包mkdir php mv php-7.2.31.tar.gz php tar -zxvf php-7.2.31.tar.gz安装依赖软件yum install gcc autoconf gcc-c++ libxml2 libxml2-devel openssl openssl-devel bzip2 bzip2-devel libcurl libcurl-devel libjpeg libjpeg-devel libpng libpng-devel freetype freetype-devel gmp gmp-devel readline readline-devel libxslt libxslt-devel systemd-devel openjpeg-devel开始编译mkdir /usr/local/php7.2 ./configure --prefix=/usr/local/php7.2 --with-config-file-path=/usr/local/php7.2/etc --with-zlib-dir --with-freetype-dir --enable-mbstring --with-libxml-dir=/usr --enable-xmlreader --enable-xmlwriter --enable-soap --enable-calendar --with-curl --with-zlib --with-gd --with-pdo-sqlite --with-pdo-mysql --with-mysqli --with-mysql-sock --enable-mysqlnd --disable-rpath --enable-inline-optimization --with-bz2 --with-zlib --enable-sockets --enable-sysvsem --enable-sysvshm --enable-pcntl --enable-mbregex --enable-exif --enable-bcmath --with-mhash --enable-zip --with-pcre-regex --with-jpeg-dir=/usr --with-png-dir=/usr --with-openssl --enable-ftp --with-kerberos --with-gettext --with-xmlrpc --with-xsl --enable-fpm --with-fpm-user=php-fpm --with-fpm-group=php-fpm --with-fpm-systemd --disable-fileinfo make && make install配置文件cd /usr/local/php7.2/etc mv php-fpm.conf.default php-fpm.conf # 编辑配置文件 vim php-fpm.conf # 将pid = run/php-fpm.pid前的“;”注释删除 cd /usr/local/php7.2/etc/php-fpm.d mv www.conf.default www.conf vim www.conf # 修改其中的user和group项user = www，group = www运行php-fpm/usr/local/php7.2/sbin/php-fpm # 检查启动状态 ps -aux | grep php修改Nginx配置文件vim /usr/local/nginx/conf/nginx.conf# 修改user项为www user www;# 去掉以下location的注释 并将fastcgi_param项中的/scripts修改为$document_root location ~ \.php$ { root html; fastcgi_pass 127.0.0.1:9000; fastcgi_index index.php; fastcgi_param SCRIPT_FILENAME $document_root$fastcgi_script_name; include fastcgi_params; } 安装MySQL下载MySQL的rpm包wget http://repo.mysql.com/mysql57-community-release-el7-11.noarch.rpm # 安装MySQL5.7的rpm包 rpm -ivh mysql57-community-release-el7-11.noarch.rpm # 更新源 yum update -y # 安装mysql yum install mysql-server -y启动MySQLsystemctl start mysqld # 开启 systemctl restart mysqld # 重启 systemctl stop mysqld # 关闭登录MySQLcat /var/log/mysqld.log | grep password mysql -u root -p # 会要求输入root密码修改root密码alter user 'root'@'localhost' identified by 'Aa123456.'; #密码最好要有大小写和符号设置有密码外部访问use mysql; # 选择mysql数据库 update user set host='%' where User='root'; # 修改user表中User="root"的行中的host值为"%" flush privileges; # 刷新
防火墙Linux使用的防火墙系统包括两部分: netfilter和iptables，netfilter/iptables框架可以实现数据包过滤、地址转换和数据包管理功能。默认的iptables规则表有：fileter表(过滤规则表)、nat表(地址转换规则表)、mangle(修改数据标记位规则表)、raw(跟踪数据表规则表)。每个规则表中都有多个数据链：INPUT(入站数据过滤)、OUTPUT(出站数据过滤)、FORWARD(转发数据过滤)、PREROUTING(路由前过滤)、POSTROUTING(路由后过滤)。图中防火墙会对数据包进行判断决策：如果是外部主机发送数据包给防火墙主机，则会经过PREROUTING和INPUT链；如果是本机发送数据包给外部主机，则经过OUTPUT和POSTROUTING链；如果防火墙为路由，则负责转发数据，经过PREROUTING、FORWARD和POSTOUTING链。iptables语法格式用法格式：iptables [-t 表名] {-A|-D|-I|-F|-L|-Z|-P} 链名 rule-specification选项-t指定需要维护的防火墙规则表-A追加防火墙规则-D删除防火墙规则-I插入防火墙规则-F清空防火墙规则-L列出防火墙规则-R替换防火墙规则-Z清空防火墙数据表统计信息-P设置链默认规则匹配参数(均可使用!取反)-p匹配协议-s匹配源地址-d匹配目标地址-i匹配入站网卡接口-o匹配出站网卡接口--sport匹配源端口--dport匹配目标端口--src-range匹配源地址范围--dst-range匹配目标地址范围--limit匹配数据表速率--mac-source匹配源MAC地址--sports匹配源端口--dports匹配目标端口--state匹配状态--string匹配应用层字串触发动作ACCEPT允许数据包通过DROP丢弃数据包REJECT拒绝数据包通过LOG将数据包记录syslog日志DNAT目标地址转换ANAT源地址转换MASQUERADE地址欺骗REDIRECT重定向命令使用方法# 查看filter表中所有规则 [root@Hyui-VM ~]# iptables -nL # 查看nat表中所有规则 [root@Hyui-VM ~]# iptables -t nat -nL # 清空filter表中所有规则 [root@Hyui-VM ~]# iptables -F # 在filter表中添加一条入站规则，丢弃223.5.5.5主机发送给防火墙本机的所有数据包 [root@Hyui-VM ~]# iptables -A INPUT -s 223.5.5.5 -j DROP # 在filter表中添加一条入站规则，拒绝223.5.5.5对主机的ping行为 [root@Hyui-VM ~]# iptables -I INPUT -s 223.5.5.5 -p icmp -j REJECT # 查看所有规则并显示编号 [root@Hyui-VM ~]# iptables -nL --line-number # 删除表中的INPUT链第一个规则 [root@Hyui-VM ~]# iptables -D INPUT 1 # 替换表中第二条规则，拒绝192.168.31.254之外的任何主机连接本机 [root@Hyui-VM ~]# iptables -R INPUT 2 ! -s 192.168.31.254 -j REJECT # 修改filter表中INPUT链的默认规则为接收数据包 [root@Hyui-VM ~]# iptables -t filter -P INPUT ACCEPT # 放行SSH和DNS要用到的端口 [root@Hyui-VM ~]# iptables -A INPUT -p tcp --dport 22 -j ACCEPT [root@Hyui-VM ~]# iptables -A INPUT -p udp --dport 53 -j ACCEPT防火墙的备份和还原# 将备份信息重定向至/etc/sysconfig/iptables [root@Hyui-VM ~]# iptables-save > /etc/sysconfig/iptables # 将备份信息重定向至./firewall.bak [root@Hyui-VM ~]# iptables-save > firewall.bak # 还原备份信息 [root@Hyui-VM ~]# iptables-restore < firewall.bakfirewalldfirewall-cmd命令firewall-cmd通过zones管理防火墙规则，一般public为默认zone用法：firewall-cmd [OPTIONS...]使用示例显示所有zone及对应规则信息[root@Hyui-VM ~]# firewall-cmd --list-all-zone在public的zone中添加允许访问ftp的规则[root@Hyui-VM ~]# firewall-cmd --add-service=ftp --zone=public success在public的zone中移除允许访问ftp的规则[root@Hyui-VM ~]# firewall-cmd --remove-service=ftp --zone=public success在public的zone中添加允许访问8080tcp端口的规则[root@Hyui-VM ~]# firewall-cmd --add-port=8080/tcp --zone=public success在public的zone中移除允许访问8080tcp端口的规则[root@Hyui-VM ~]# firewall-cmd --remove-port=8080/tcp --zone=public success将eth0网卡与public绑定，从该接口进入的流量全部与public匹配[root@Hyui-VM ~]# firewall-cmd --add-interface=eth0 --zone=public success将eth0网卡与public解除绑定[root@Hyui-VM ~]# firewall-cmd --remove-interface=eth0 --zone=trusted查看默认zone规则列表和public规则列表[root@Hyui-VM ~]# firewall-cmd --list-all public target: default icmp-block-inversion: no interfaces: sources: services: dhcpv6-client ssh ports: 10000/udp 10000/tcp protocols: masquerade: no forward-ports: source-ports: icmp-blocks: rich rules: [root@Hyui-VM ~]# firewall-cmd --list-all --zone=public public target: default icmp-block-inversion: no interfaces: sources: services: dhcpv6-client ssh ports: 10000/udp 10000/tcp protocols: masquerade: no forward-ports: source-ports: icmp-blocks: rich rules:在public的zone中添加一条永久规则(永久允许访问8080端口)，该规则在重启后依然有效[root@Hyui-VM ~]# firewall-cmd --add-port=8080/tcp --zone=public --permanent success重载防火墙[root@Hyui-VM ~]# firewall-cmd --reload success
系统环境使用CentOS7用yum安装aria2yum update -y yum install aria2在/root目录下创建一个.aria2目录mkdir /root/.aria2并且创建aria2.session和aria2.log和aria2.conf文件在里面cd /root/.aria2 touch aria2.session touch aria2.log touch aria2.conf chmod 744 ./*其中aria2.conf是配置文件 需要进行编辑vim aria2.conf示例配置：## 关于文件保存 # 文件的保存路径，默认:为当前启动位置 dir=/down # 启用磁盘缓存, 0为禁用缓存 #disk-cache=32M # file-allocation=none # 断点续传 continue=true ## 关于下载连接 # 最大同时下载任务数, 默认:5 max-concurrent-downloads=10 # 同一服务器连接数, 默认:1 max-connection-per-server=5 min-split-size=10M # 单个任务最大线程数, 默认:5 split=5 disable-ipv6=false input-file=/root/.aria2/aria2.session save-session=/root/.aria2/aria2.session save-session-interval=60 ## RPC设置 # 启用RPC enable-rpc=true # 允许所有来源 rpc-allow-origin-all=true # 允许非外部访问 rpc-listen-all=true # RPC监听端口, 端口被占用时可以修改, 默认:6800 rpc-listen-port=6800 # 设置的RPC授权令牌 rpc-secret=PASSWORD ## BT下载设置 # 允许进行.torrent的BT种子下载 follow-torrent=true # BT监听端口, 当端口被屏蔽时使用, 默认:6881-6999 listen-port=51413 # 单个种子最大连接数 #bt-max-peers=55 enable-dht=true enable-peer-exchange=true # 客户端伪装 peer-id-prefix=-TR2770- user-agent=Transmission/2.77 seed-ratio=0.1 #一直保留会话 force-save=true bt-seed-unverified=true其中需要改的是RPC授权令牌(将PASSWORD改为自己设定的密码)，这个一会要用得到，改完要记住运行aria2setsid aria2c --conf-path=/root/.aria2/aria2.conf 设置开机启动请自己实现 不多赘述然后安装webui-aria2cd git clone https://github.com/ziahamza/webui-aria2.git cd webui-aria2 chmod +x ./*运行webui-aria2需要nodejs环境yum install nodejs然后修改node-server.js这个文件其中有一段是var http = require("http"), url = require("url"), path = require("path"), fs = require("fs"); port = process.argv[2] || 8888;8888端口是这个程序要占用的端口可以修改为其他端口 例如8999然后运行setsid node ./node-server.js出现running on http://localhost:8888就是运行成功了浏览器进入http://你的IP:8888就可以看到aria2了进入设定->连线设定->密码令牌 输入之前设置好的RPC令牌就可以使用了aria2想要更快更稳的速度可以添加Tracker(可以在github上搜到可供使用的Tracker)添加Tracker可以在aria2.conf文件中的最后一行添加bt-tracker=之后添加Tracker 每个Tracker之间用,隔开文件下载好以后我想把下载好的文件上传到onedrive怎办?可以用onedrivecmd安装onedrivecmd#如果你用的Python pip install onedrivecmd #如果你用的python3 pip3 install onedrivecmd不能用pip的请查找pip安装教程onedrivecmd安装好以后需要登录#个人版登录 onedrivecmd init #企业/Office 365/Education登录 onedrivecmd init_business运行命令以后会给你一段很长的网址复制粘贴到浏览器打开 登陆上账号会显示很长的一段乱码复制那一段乱码粘贴到ssh里面的"Paste code here:"后面回车 就绑定完成了(/root/.onedrive.json文件以及刚才复制的一串乱码千万不能泄露)关于onedrivecmd的常用命令#查看容量 onedrivecmd quota #上传 onedrivecmd put 本机文件 od:网盘目录 例如onedrivecmd put ./test.txt od:/share #下载 onedrivecmd get od:网盘文件 本机目录 例如onedrivecmd get od:/test.txt /root更多命令可以查看https://github.com/cnbeining/onedrivecmd
Endlessh的正确食用方法:防止那些坏家伙连接你的SSH
SSH最近出了个新的玩法，使用Endlessh，对于猜ssh密码的人，能够缓慢的随机的无限的发送SSH banner，可以锁定数小时甚至数天，可以将真正的SSH服务放在另一个端口上，在SSH被暴破时并不会打扰到真正的服务器。那么如何配置这个Endlessh呢？本教程使用ArchLinux系统环境我们可以去github上下载到pacman -S git git clone https://github.com/skeeto/endlessh.git cd endlessh/ 用ls查看目录里的文件发现里面的文件[root@archlinux endlessh]# ls Dockerfile Makefile README.md UNLICENSE endlessh.c util 我们可以通过docker或者make来安装这个程序(本次使用make)#安装make pacman -S make 我们需要安装gccpacman -S gcc 然后makemake 此时目录中有这些文件[root@archlinux endlessh]# ls Dockerfile Makefile README.md UNLICENSE endlessh endlessh.c util 在/etc中创建一个目录/etc/endlessh然后创建config文件(config文件是配置文件)mkdir /etc/endlessh touch /etc/endlessh/config然后编辑config文件vim /etc/endlessh/configgithub中的作者给出了示例配置文件(可以参考)# 监听新ssh连接端口，有防火墙的请在防火墙中打开2222端口 Port 2222 # 随机SSH banner每行之间的毫秒数 Delay 10000 # 每行随机长度的最大值 MaxLineLength 32 # 一次接受的最大连接数 MaxClients 4096 # 日志详细级别(0:少 1:标准 2:多) # 0 = Quiet # 1 = Standard, useful log messages # 2 = Very noisy debugging information LogLevel 0 编辑完成后运行这个程序./endlessh -f config 要保持后台运行可以使用setsidsetsid ./endlessh -f config endlessh默认情况下并不会输出日志消息 可以通过运行./endlessh -v >endlessh.log 2>endlessh.err来将日志和错误日志生成在endlessh.log里和endlessh.err里如果运行后报错返回Address family not supported by protocol请查看grub配置文件中是否禁用了IPV6，如果禁用了请取消禁用
前一段时间发现路由器里面可以自定义DNS，就想如果把纯净的DNS放入路由器里会不会直接访问谷歌和pixiv之类的。教程开始我们制作私人DNS有两种方法：1.dnsmasq2.overture制作环境使用CentOS71.dnsmasq首先我们先更新源yum -y update然后安装dnsmasqyum install dnsmasq -ydnsmasq的配置文件在/etc/dnsmasq.conf中，编辑这个文件vim /etc/dnsmasq.conf里面的东西很多，需要添加的就以下几种#这个是你指定的hosts文件，dnsmasq会使用你指定的hosts(有的时候hosts比较好的可以直接连YouTube) addn-hosts=/etc/hosts #这个是你自定义的上游DNS(第一个是阿里云的DNS，第二个是教育网的DNS) server=223.5.5.5 server=202.38.93.153 #listen-address这个配置不可以有，一定要删掉 #如果看到listen-address=127.0.0.1这个可以注释掉或者删掉dnsmasq的基本指令#开启 sudo service dnsmasq start #关闭 sudo service dnsmasq stop #重启 sudo service dnsmasq restart开启以后查看53端口netstat -tunlp | grep 53如果显示结果为tcp 0 0 0.0.0.0:53 0.0.0.0:* LIST EN 2346/dnsmasq udp 0 0 0.0.0.0:53 0.0.0.0:* 2346/dnsmasqdnsmasq正在运行了dnsmasq有个很大的缺点，虽然操作比较简单，但是dnsmasq是从上游DNS的53端口获取流量，而无法确定上游DNS的53端口的流量是否已经遭到了污染，如果你一套稳定而且很棒的hosts可以用dnsmasq。2.overture和dnsmasq不同的是，overture可以从上游DNS的非标准端口(非53端口)接收流量，不过需要上游DNS的支持。overture不仅可以用在VPS中，也可以用在树莓派甚至Androidoverture可以在GITHUB上找到点此进入根据你的架构下载对应的版本cd ~ wget [要下载文件的网址] 我下载的是overture-linux-amd64.zipmkdir overture && mv overture-linux-amd64.zip overture chmod 774 overture解压zip文件cd overture unzip overture-linux-amd64.zip chmod 774 ./* 会有这些文件[root@hyui overture]# ls config.json domain_primary_sample hosts_sample ip_network_primary_sample overture-linux-amd64.zip domain_alternative_sample ip_network_alternative_sample overture-linux-amd64 修改config.jsonvim config.json 其中需要修改一些地方"PrimaryDNS": [ //你的主DNS信息，使用了DNSPOD { "Name": "DNSPod", "Address": "119.29.29.29:53", "Protocol": "udp", "SOCKS5Address": "", "Timeout": 6, "EDNSClientSubnet": { "Policy": "auto", "ExternalIP": "", "NoCookie": true } } ], "AlternativeDNS": [ //你的备用DNS信息，这个是外国DNS，为了防止对53端口的投毒，可以使用443端口(前提是他们支持443或其他端口) { "Name": "Opendns", "Address": "208.67.220.220:443", "Protocol": "tcp", "SOCKS5Address": "", "Timeout": 6, "EDNSClientSubnet": { "Policy": "disable", "ExternalIP": "", "NoCookie": true } } ],关于EDNSClientSubnet的说明，作者给出的是EDNSClientSubnet: Used to improve DNS accuracy. Please check RFC7871 for details.Policyauto: If client IP is not in the reserved IP network, use client IP. Otherwise, use external IP.manual: Use external IP if this field is not empty, otherwise use client IP if it is not reserved IP.disable: Disable this feature.ExternalIP: If this field is empty, ECS will be disabled when the inbound IP is not an external IP.NoCookie: Disable cookie.EDNSClientSubnet：用于提高DNS准确性。auto：如果客户端IP不在保留的IP网络中，请使用客户端IP。否则，请使用外部IP。manual：如果此字段不为空，则使用外部IP，否则使用客户端IP（如果不是保留IP）。disable：禁用此功能。ExternalIP：如果此字段为空，则当入站IP不是外部IP时，将禁用ECS。NoCookie：禁用cookie。overture同样可以自定义hosts"HostsFile": "/root/overture/hosts" //这一行中的hosts即为自定义的hosts需要自己添加 设置完成后就可以运行了./overture-linux-amd64会显示INFO[0000] Overture v1.5-rc3 INFO[0000] If you need any help, please visit the project repository: https://github.com/shawn1m/overture INFO[0000] Load domain ./domain_primary_sample successful with 1 records INFO[0000] Load domain ./domain_alternative_sample successful with 1 records INFO[0000] Load ./ip_network_primary_sample successful INFO[0000] Load ./ip_network_alternative_sample successful INFO[0000] Minimum TTL is disabled INFO[0000] Cache is disabled INFO[0000] Load hosts file successful INFO[0000] Start overture on :53 说明正在运行了保持overture在后台运行可以用setsid ./overture-linux-amd64 当然我们也可以写一个脚本让它开机自启动mkdir /etc/overture vim /etc/overture/overture.sh #!/bin/bash cd /root/overture/ setsid ./overture-linux-amd64 创建好以后标记为可执行文件chmod +x /etc/overture/overture.sh 将rc.local也标记为可执行文件chmod +x /etc/rc.d/rc.local 打开rc.localvim /etc/rc.d/rc.local 添加脚本进去/etc/overture/overture.sh就可以开机被执行了。后记:后来一直都觉得哪里不太好，觉得是不是哪里可以改进之类的，于是有了这个补充。/root/overture这个目录我们完全可以放在/usr/local/目录中然后进入/usr/local/overture将overture-linux-amd64软链接到/usr/bin这样就可以不用在cd进overture的目录了# 代码实现 ## 移动前可以先停止overture的运行 mv /root/overture /usr/local/ cd /usr/local/overture ## 软链接的实现 ln -s /usr/local/overture/overture-linux-amd64 /usr/bin/overture ## 运行 setsid overture -c /usr/local/overture/config.json 这个时候就不需要那个开机脚本了 直接把运行命令添加到/etc/rc.d/rc.local就够了