Ubuntu巡察术讲义：脚本自成妙用法

教程目的与适用人群

本教程旨在帮助刚开始接触服务器的小白用户及Linux爱好者学习如何编写系统检测脚本，通过本教程，你将掌握如何实时监控系统状态，及时发现系统异常，保障服务器或工作站的稳定运行。

开始

本文将详述 邮件提醒 与 Telegram Bot提醒 之法，特奉上完整 Python 脚本。脚本内设详尽注释，烦请依注释指引，妥善替换相关配置信息。愿此助君快速掌握自动提醒之术。

import re
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from datetime import datetime, timedelta
import os

# ===== 配置区域 =====
LANG = 'zh'  # 'zh' 或 'en'

LANG_PACK = {
    'zh': {
        'subject': "服务器安全警报",
        'alert_detected': "🚨 发现安全警报：",
        'no_alert': "未检测到异常。",
        'email_fail': "邮件发送失败：",
        'email_success': "邮件发送成功。",
        'telegram_fail': "Telegram发送失败：",
        'telegram_success': "Telegram消息发送成功。",
        'attachment_name': "警报日志片段.txt"
    },
    'en': {
        'subject': "Server Security Alert",
        'alert_detected': "🚨 Security Alert Detected:",
        'no_alert': "No alerts detected.",
        'email_fail': "Failed to send email:",
        'email_success': "Email sent successfully.",
        'telegram_fail': "Telegram send failed:",
        'telegram_success': "Telegram message sent.",
        'attachment_name': "alert_log_snippet.txt"
    }
}

# 日志文件及规则配置（你可以添加更多规则）
LOG_CONFIG = {
    '/var/log/auth.log': [
        {
            'name': 'SSH Failed Login',
            'pattern': r'(\w{3} \d{1,2} \d{2}:\d{2}:\d{2}).*Failed password for .* from (\d+\.\d+\.\d+\.\d+)',
            'threshold': 3,
            'time_window': 5
        },
        # 你可以在这里继续添加其他检测规则，比如端口扫描等
    ],
    # '/var/log/other.log': [...]
}

# 邮件配置
SMTP_SERVER = 'smtp.example.com'
SMTP_PORT = 587
SMTP_USER = '[email protected]'
SMTP_PASS = 'your_email_password'
EMAIL_TO = '[email protected]'
EMAIL_FROM = SMTP_USER

# Telegram配置
TELEGRAM_BOT_TOKEN = 'your_bot_token'
TELEGRAM_CHAT_ID = 'your_chat_id'

# ===== 功能实现 =====

def send_email(subject, body, attachment_path=None):
    lang = LANG_PACK[LANG]

    msg = MIMEMultipart()
    msg['Subject'] = subject
    msg['From'] = EMAIL_FROM
    msg['To'] = EMAIL_TO

    msg.attach(MIMEText(body, 'plain', 'utf-8'))

    if attachment_path and os.path.exists(attachment_path):
        with open(attachment_path, 'rb') as f:
            part = MIMEBase('application', 'octet-stream')
            part.set_payload(f.read())
        encoders.encode_base64(part)
        part.add_header('Content-Disposition',
                        f'attachment; filename="{os.path.basename(attachment_path)}"')
        msg.attach(part)

    try:
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASS)
            server.sendmail(EMAIL_FROM, EMAIL_TO, msg.as_string())
        print(lang['email_success'])
    except Exception as e:
        print(f"{lang['email_fail']} {e}")

def send_telegram(message):
    url = f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage'
    data = {'chat_id': TELEGRAM_CHAT_ID, 'text': message}
    lang = LANG_PACK[LANG]
    try:
        response = requests.post(url, data=data)
        if response.status_code == 200:
            print(lang['telegram_success'])
        else:
            print(f"{lang['telegram_fail']} {response.text}")
    except Exception as e:
        print(f"{lang['telegram_fail']} {e}")

def parse_time(log_time_str):
    try:
        log_time = datetime.strptime(f"{datetime.now().year} {log_time_str}", "%Y %b %d %H:%M:%S")
        if log_time > datetime.now():
            log_time = log_time.replace(year=log_time.year - 1)
        return log_time
    except:
        return None

def detect_attacks():
    alerts = []
    lang = LANG_PACK[LANG]
    alert_log_lines = []

    for log_file, rules in LOG_CONFIG.items():
        try:
            with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                log_lines = f.readlines()
        except Exception as e:
            print(f"Failed to read {log_file}: {e}")
            continue

        for rule in rules:
            name = rule['name']
            pattern = re.compile(rule['pattern'])
            threshold = rule['threshold']
            time_window = rule['time_window']

            now = datetime.now()
            time_limit = now - timedelta(minutes=time_window)
            ip_counts = {}

            for line in log_lines:
                match = pattern.search(line)
                if match:
                    if len(match.groups()) == 2:
                        time_str, ip = match.group(1), match.group(2)
                        log_time = parse_time(time_str)
                    else:
                        continue

                    if log_time and log_time >= time_limit:
                        ip_counts[ip] = ip_counts.get(ip, 0) + 1
                        alert_log_lines.append(line.strip())

            for ip, count in ip_counts.items():
                if count >= threshold:
                    alerts.append(f"[{name}] {ip} {count} times in last {time_window} minutes [{log_file}]")

    return alerts, alert_log_lines

def save_alert_log(lines):
    lang = LANG_PACK[LANG]
    filename = lang['attachment_name']
    with open(filename, 'w', encoding='utf-8') as f:
        f.write("\n".join(lines))
    return filename

def main():
    lang = LANG_PACK[LANG]
    alert_list, alert_logs = detect_attacks()
    if alert_list:
        message = lang['alert_detected'] + "\n" + "\n".join(alert_list)
        print(message)
        attach_file = save_alert_log(alert_logs)
        send_email(lang['subject'], message, attachment_path=attach_file)
        send_telegram(message)
        if os.path.exists(attach_file):
            os.remove(attach_file)
    else:
        print(lang['no_alert'])

if __name__ == '__main__':
    main()
    

顺带提醒：很多邮箱需要开SMTP服务或授权码，而不是直接用登录密码，具体看你邮箱的安全设置。

以 Gmail 为例：

SMTP 配置参数

• SMTP 服务器地址（SMTP_SERVER）：smtp.gmail.com
• 端口号（SMTP_PORT）：587（TLS）或者465（SSL）
• 邮箱账号（SMTP_USER）：你的完整 Gmail 地址，例如 [email protected]
• 邮箱密码（SMTP_PASS）：应用专用密码（App Password），不是你的登录密码

获取 Gmail 应用专用密码步骤

• 登录你的 Gmail 账号
• 进入 Google 账号安全设置
• 确保已经开启 两步验证（2-Step Verification）
• 在 “应用密码”（App Passwords）里创建一个新的密码，选择 “邮件” + 你的设备名称
• 生成的 16 位密码即为 SMTP_PASS，在脚本中使用它

SMTP_SERVER = 'smtp.gmail.com'
SMTP_PORT = 587
SMTP_USER = '[email protected]'
SMTP_PASS = 'your_16_character_app_password'

安装依赖

连接你的服务器，通过以下命令安装所需依赖

pip3 --version

sudo apt update
sudo apt install python3-pip
pip3 install psutil requests

python3 --version

测试代码

python3 filename.py

如果输出No alerts detected，但是卡住了，可能的原因是：

• 程序执行到了 send_email() 函数；
• 并卡在了with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:

原因可能是以下几点之一：

• 无法连接到 SMTP 服务器（最常见）
  • 你的服务器可能无法访问目标 SMTP 服务器，即使能联网（例如能 ping）也可能是端口或防火墙问题。
• SMTP 端口被防火墙、VPS 或目标服务封锁
  • 端口 587（STARTTLS）、465（SSL）和 25（老 SMTP）经常被 VPS 提供商限制（尤其是 Vultr、DigitalOcean）。
  • 你可以用如下方式测试是否能连接 SMTP 服务器：
    • telnet smtp.gmail.com 587, 如果使用telnet smtp.gmail.com 587卡住了，说明VPS封锁了**外发 SMTP（端口 25、465、587）**流量。

解决方案总结：

以下是这封 工单的 中英文对照版本，你可以在提交时使用英文版本，同时了解每一句的中文含义，便于你根据需要增减内容。（请根据你的 VPS 厂商选择对应版本）

Request to unblock SMTP port 587

Hello Support,

I would like to request the unblocking of SMTP port 587 on my VPS. I need this to send alert emails from my server using a secure mail provider. The emails are only for system monitoring and notification purposes — no bulk mailing or spam involved.

My use case is to send periodic system health reports and attack alerts via email using Gmail’s SMTP service. If you need further clarification or logs, I would be happy to provide them.

Please let me know if any further information is required.

Thank you!

你好，支持团队，

我想申请解除我的 VPS 上 SMTP 端口 587 的封锁。我需要使用这个端口通过一个安全的邮件服务商发送系统告警邮件。这些邮件仅用于系统监控与提醒，不涉及群发、营销或垃圾邮件。

我的用途是通过 Gmail 的 SMTP 服务定期发送系统运行状态报告和攻击提醒邮件。如果你们需要更多信息或日志，我可以提供。

如需进一步确认，请随时联系我。

谢谢！

设置自动运行

配置定时运行（每 12 小时发送一次）

• 编辑定时任务：crontab -e
• 添加这一行： 0 */12 * * * /usr/bin/python3 /root/filename.py >> /root/filename.log 2>&1

所示方法虽为简易入门，欲臻完善与专业，尚需君自行借助 AI 辅助 以深究拓展。