วาง Telegram Bot เฝ้า DA Server แบบ blast radius เล็ก

ตอนทำงานกับ DirectAdmin server ที่มีหลาย user อยู่บนเครื่องเดียว ปัญหาที่กวนใจที่สุดไม่ใช่เรื่อง spec หรือ tuning แต่เป็นเรื่อง "จังหวะที่รู้ว่ามีอะไรผิดปกติ" ส่วนใหญ่ผมจะรู้ตอนลูกค้าโทรมาแจ้ง ซึ่งแปลว่าช้าไปแล้วหนึ่งก้าวเสมอ จะ SSH เข้าจากมือถือก็พิมพ์ลำบาก ขาด context มองจอเล็กแล้วเสี่ยงไปแตะของอันตราย จะยก dashboard ขึ้นมาทั้งชุดก็เท่ากับเพิ่ม service ที่ต้องดูแลและเพิ่ม port ที่ต้องเปิด

ผมเลยลองวาง Telegram bot เป็น control surface แบบเบาๆ ดู status ตรวจ log และ block IP จากหน้าแชทได้ บทความนี้คือ setup ขั้น minimum ที่ผมใช้จริง จงใจจำกัดไว้แค่ read-only commands บวก block/unblock IP เท่านั้น ไม่มี kill process ไม่มี restart service ไม่มี auto-action ใดๆ ทั้งสิ้น ตั้งใจให้ bot เป็นแค่ observer บวกกับ mutation ที่ระดับ IP layer เพื่อให้ blast radius เล็กที่สุดเท่าที่จะทำได้

ทำไมเลือก Telegram bot

ก่อนจะลงตัวที่ Telegram ผมเทียบตัวเลือกหลักไว้สี่ทาง แต่ละทางมีจุดที่ติดของมันเอง

ผมลงตัวที่ Telegram ด้วยเหตุผลสี่ข้อ ข้อแรกคือไม่ต้องเปิด port เลย bot ใช้ long-polling วิ่งออกไปหา Telegram API ฝั่ง firewall ขาเข้าไม่ต้องแตะอะไรทั้งนั้น ข้อสองคือมันพ้นปัญหา NAT มือถือไม่ต้องมี public IP ไม่ต้องตั้ง VPN ก็รับ push ได้ ข้อสามคือเป็น conversation สองทางจริง ไม่ใช่ alert ขาเดียวแบบเมล สั่งกลับเข้าไปได้ และข้อสุดท้าย auth ผมจัดการเองได้ด้วย chat_id allowlist ซึ่งกันได้ในระดับเดียวกับ SSH key เลย

หลัก "Observer first, operator later"

นี่คือกฎข้อแรกที่ผมยึดกับ bot ทุกตัว และเป็นเหตุผลที่ทำให้ bot ตัวนี้ปลอดภัยพอจะเปิดให้สั่งจาก Telegram ได้โดยไม่ต้องนอนไม่หลับ Phase แรกของ bot คือ read-only บวก block IP เท่านั้น bot อ่าน log ดู status push alert ได้ ส่วน mutation มีแค่ csf -d กับ csf -dr สำหรับ block และ unblock IP ซึ่ง reversible ด้วยคำสั่งเดียว

จุดที่ผมระวังมากคือ sudoers ต้องไม่มี wildcard ที่อันตราย ไม่มี kill -9 * ไม่มี systemctl restart * เด็ดขาด เพราะ wildcard พวกนี้แปลว่า bot สั่ง kill init หรือ restart sshd ได้ ซึ่งเท่ากับ brick เครื่อง และอีกข้อคือ bot ไม่ auto-act bot แค่ alert บวกแนะนำคำสั่ง คนเป็นคนตัดสินใจกดเองทุกครั้ง ถ้าตั้ง auto-action ไว้ false positive แค่ครั้งเดียวก็ลามไปลูกค้าหลายเจ้าพร้อมกัน

Architecture — bot รันบนเครื่องไหน

คำถามถัดมาคือ bot ควรรันบนเครื่องไหน ผมเทียบสามทาง รันบน web1 ตัวเดียวกับ DA ได้ local access latency เป็นศูนย์ แต่ข้อเสียร้ายแรงคือถ้า web1 ล่ม bot ก็ล่มไปด้วยพอดีตอนที่ต้องใช้มันที่สุด แถมเป็นการเพิ่ม service บนเครื่องที่ public อยู่แล้ว รันบน VPS แยกข้างนอกก็ได้ external eye ดี แต่มี cost เพิ่ม maintenance เพิ่ม และ key ต้องวิ่งออกไปอยู่นอกบ้าน

ผมเลือกทางกลาง คือรัน bot ที่ bak1 ซึ่งเป็นเครื่องเดียวกับ backup host จากบทความก่อนหน้า bot ส่ง SSH ไปหา web1 ผ่าน key ข้อดีที่สำคัญที่สุดคือถ้า web1 ล่ม bot ที่อยู่คนละเครื่องยังบอกได้ทันทีว่า "web1 unreachable" แลกมากับ SSH hop ที่เพิ่มมาแค่ราวร้อยมิลลิวินาที ซึ่งคุ้มมาก

+-------------+        long-poll HTTPS         +------------------+
|  Telegram   | <===========================>  |  bak1 (bot host) |
|   server    |       (port 443 outbound)      |  - bot.py        |
+-------------+                                 |  - systemd unit  |
                                                |  - audit.sqlite  |
                                                +------------------+
                                                         |
                                                         | SSH (key, from=<bak1-ip>)
                                                         v
                                                +------------------+
                                                | web1 (DA prod)   |
                                                |  - user botuser  |
                                                |  - sudoers       |
                                                |    (allowlist)   |
                                                +------------------+

Telegram คุยกับ bak1 ด้วย HTTPS ขาออก port 443 อย่างเดียว ส่วน bak1 ยิง SSH ที่ล็อก from= ไปหา web1 ทั้งสองทิศไม่มี inbound port ที่ต้องเปิดบน firewall เลย

สมัคร bot ผ่าน BotFather

Telegram ให้สมัคร bot ผ่าน bot กลางชื่อ @BotFather เปิด Telegram แล้ว search หา @BotFather กด start จากนั้นพิมพ์คำสั่งตามลำดับ

/newbot
> ตั้งชื่อ display: Example Monitor Bot
> ตั้ง handle (ลงท้าย _bot): example_monitor_bot
< HTTP API token: <bot-token>

token ที่ได้มาให้เก็บเหมือน password ใครได้ token ไปก็สั่ง bot ได้แทนเราทันที ห้าม commit เข้า git ห้าม paste ตอน screen share และห้ามวางไว้ในไฟล์ที่ web user อ่านได้

ตั้งค่าเสริมได้ผ่าน /setdescription สำหรับคำอธิบาย bot, /setabouttext สำหรับหน้า about และ /setcommands สำหรับลง list คำสั่งให้ Telegram ทำ autocomplete ให้

/setcommands

สำหรับ /setcommands ผมใส่ list ตามนี้ จุดที่ต้องระวังคือคำสั่งจริงของ bot ใช้ . เป็น prefix แต่ list ใน BotFather บังคับให้ขึ้นต้นด้วย / เพราะมันบังคับ format ของมันเอง

start - แสดง chat_id
ping - bot alive check
status - สถานะ service หลัก
load - load avg + top CPU
disk - disk usage
queue - mail queue size
whois - ตรวจ IP ใน CSF
block - บล็อก IP ถาวร (admin)
unblock - ปลด IP

เหตุผลที่คำสั่งจริงใช้ prefix . แทน / คือไม่อยากให้ Telegram autocomplete เด้งขึ้นมารกหน้าจอทุกครั้งที่พิมพ์ ส่วน /start เป็นข้อยกเว้นเพราะ Telegram protocol บังคับให้ใช้ /

หา chat_id ของตัวเอง — กำหนด allowlist

bot ตัวนี้ไม่ใช้ OAuth auth ของผมคือ chat_id allowlist ล้วนๆ คือเก็บ list ของ Telegram chat_id ที่อนุญาตให้คุยกับ bot ได้ ใครไม่อยู่ใน list bot จะ drop ทิ้งแบบเงียบไม่ตอบกลับเลย หา chat_id ของตัวเองได้สองวิธี วิธีแรกทัก bot ชื่อ @userinfobot ใน Telegram มันจะตอบ Id กลับมา วิธีที่สองคือหลัง bot ของเรา up แล้ว พิมพ์ /start ใส่ bot ตัวเอง เพราะผม code ให้มันตอบ chat_id ของผู้ส่งกลับมา

สมมุติได้ chat_id ของผมเป็น <chat-id-admin> และ chat_id ของ admin อีกคนเป็น <chat-id-observer> ตัวอย่างการคุยตอนทัก bot ครั้งแรก

[คุณ]
/start

[@example_monitor_bot]
chat_id = <chat-id-admin>
ใส่เลขนี้ใน config.yaml -> allowed_chat_ids
แล้วสั่ง .reload หรือ restart service

เอา chat_id ที่ได้ไปใส่ใน config.yaml

telegram:
  bot_token_env: BOT_TOKEN          # อ่าน token จาก env var ไม่ hardcode
  allowed_chat_ids:
    - <chat-id-admin>                # ผม
    - <chat-id-observer>             # admin อีกคน
  admin_chat_ids:
    - <chat-id-admin>                # admin ทำ mutation ได้
  alert_chat_id: <chat-id-admin>     # ที่ส่ง alert

จุดที่ต้องแยกให้ชัดคือ allowed_chat_ids คือคนที่อ่านได้ ส่วน admin_chat_ids คือ subset ที่ทำ mutation อย่าง block/unblock ได้ คนที่อยู่ใน allowed แต่ไม่อยู่ใน admin จะดู status ได้แต่ block IP ไม่ได้

ตั้ง bot user + systemd hardening บน bak1

bot ไม่ควรรันใต้ root เด็ดขาด สร้าง user แยกชื่อ botuser ที่ไม่มี shell และไม่มี sudo อิสระ

# สร้าง user
useradd -r -s /bin/false -d /opt/srv-bot botuser

# วาง code
mkdir -p /opt/srv-bot/{logs,config}
chown -R botuser:botuser /opt/srv-bot
chmod 750 /opt/srv-bot

จากนั้นเขียน systemd unit ที่ /etc/systemd/system/srv-bot.service พร้อม hardening directive ครบชุด

[Unit]
Description=Server Monitor Telegram Bot
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=botuser
Group=botuser
WorkingDirectory=/opt/srv-bot
EnvironmentFile=/etc/srv-bot/secrets.env   # token อยู่ที่นี่ chmod 600
ExecStart=/opt/srv-bot/.venv/bin/python /opt/srv-bot/bot.py
Restart=on-failure
RestartSec=10
StandardOutput=append:/opt/srv-bot/logs/bot.log
StandardError=append:/opt/srv-bot/logs/bot.log

# Hardening
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ReadWritePaths=/opt/srv-bot/logs /opt/srv-bot/config

[Install]
WantedBy=multi-user.target

directive ชุด hardening ตรงท้ายเป็นหัวใจ NoNewPrivileges กัน process ยกระดับสิทธิ์ตัวเอง ProtectSystem=strict ทำให้ filesystem ส่วนใหญ่เป็น read-only ProtectHome ตัด /home ออกจากสายตา bot และ ReadWritePaths คือ allowlist directory ที่ bot เขียนได้จริงแค่สองที่

token เก็บแยกในไฟล์ secrets ที่ root เป็นเจ้าของ ส่วน bot อ่านผ่าน group

# /etc/srv-bot/secrets.env
BOT_TOKEN=<bot-token>

chmod 700 /etc/srv-bot
chmod 600 /etc/srv-bot/secrets.env
chown root:botuser /etc/srv-bot/secrets.env
# bot อ่านได้ผ่าน systemd EnvironmentFile แต่ user อื่นเปิดดูไม่ได้

SSH key จาก bak1 ไป web1 — least privilege

bot ที่ bak1 ส่ง command ไป web1 ผ่าน SSH ด้วย key user ปลายทางที่ web1 ก็ชื่อ botuser เหมือนกัน แต่ตัวที่อยู่บน web1 นี้มี sudoers จำกัดไว้เฉพาะ command ที่ bot ต้องใช้จริง สร้าง key บน bak1 ในนามของ user botuser ก่อน

sudo -u botuser ssh-keygen -t ed25519 \
    -f /opt/srv-bot/.ssh/web1_key -N ""

แล้วเอา pub key ไปใส่ที่ web1 ใน /home/<botuser>/.ssh/authorized_keys โดยล็อกด้วย from= และปิด forwarding ทุกชนิด

from="<bak1-ip>",no-port-forwarding,no-X11-forwarding,no-agent-forwarding,no-pty ssh-ed25519 AAAA<pub_key> bot-from-bak1

from="<bak1-ip>" ทำให้ key นี้ใช้ได้เฉพาะตอนที่ connection มาจาก bak1 เท่านั้น ถ้า key หลุดออกไปข้างนอก internet ก็ใช้ไม่ได้ ส่วนการปิด forwarding ทุกชนิดบวก no-pty หมายความว่า key นี้ทำได้แค่รัน command ตรงๆ เปิด tunnel หรือ interactive shell ไม่ได้

ฝั่ง sudoers บน web1 เขียนที่ /etc/sudoers.d/botuser-readonly ระบุเฉพาะ command ที่ bot ต้องใช้

# Read-only investigations (no password needed)
botuser ALL=(root) NOPASSWD: /usr/sbin/csf -g *
botuser ALL=(root) NOPASSWD: /usr/sbin/csf -l
botuser ALL=(root) NOPASSWD: /usr/bin/exim -bpc
botuser ALL=(root) NOPASSWD: /usr/bin/systemctl is-active *

# tail/grep — only specific log files (no wildcards on path)
botuser ALL=(root) NOPASSWD: /usr/bin/tail /var/log/lfd.log
botuser ALL=(root) NOPASSWD: /usr/bin/tail /var/log/exim/mainlog

# Block IP — reversible mutation
botuser ALL=(root) NOPASSWD: /usr/sbin/csf -d *
botuser ALL=(root) NOPASSWD: /usr/sbin/csf -dr *
botuser ALL=(root) NOPASSWD: /usr/sbin/csf -td *

Python skeleton + ฟัง command

stack ที่ผมใช้เป็น async ทั้งสาย aiogram 3.x เป็น Telegram bot framework แบบ async-native, asyncssh เป็น SSH client ที่ persist connection ได้ไม่ต้อง re-handshake ทุก command, loguru สำหรับ structured logging, aiosqlite สำหรับ audit log และ PyYAML สำหรับอ่าน config โครงไฟล์ขั้น minimum เป็นแบบนี้

/opt/srv-bot/
|-- bot.py
|-- config/
|   `-- config.yaml
|-- logs/
|   |-- bot.log
|   `-- audit.sqlite
`-- .venv/

ตัวอย่าง bot.py ขั้นต่ำที่รับคำสั่ง .ping พร้อม auth check ให้เห็นโครงก่อน

import os, yaml, asyncio
from aiogram import Bot, Dispatcher, F
from aiogram.filters import Command
from aiogram.types import Message
from loguru import logger

CONFIG = yaml.safe_load(open('config/config.yaml'))
ALLOWED = set(CONFIG['telegram']['allowed_chat_ids'])
ADMINS  = set(CONFIG['telegram']['admin_chat_ids'])

bot = Bot(token=os.environ['BOT_TOKEN'])
dp  = Dispatcher()

# auth gate — ถ้า chat_id ไม่อยู่ใน allowlist drop ทิ้ง
async def is_allowed(m: Message) -> bool:
    if m.chat.id not in ALLOWED:
        logger.warning(f"reject chat_id={m.chat.id} user={m.from_user.username}")
        return False
    return True

# คำสั่งใช้ prefix '.' (ไม่ใช่ '/')
@dp.message(F.text.startswith('.ping'))
async def cmd_ping(m: Message):
    if not await is_allowed(m): return
    await m.reply("pong")

@dp.message(Command('start'))
async def cmd_start(m: Message):
    await m.reply(f"chat_id = {m.chat.id}")  # ตอบ chat_id ให้ user เก็บ

async def main():
    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())

จุดที่ต้องเข้าใจคือ /start ตอบ chat_id ให้ใครก็ได้ที่ทักโดยไม่ผ่าน is_allowed เพราะตั้งใจให้คนใหม่หา chat_id ของตัวเองได้ แต่ทุกคำสั่งอื่นต้องผ่าน auth gate หมด

[คุณ (chat_id <chat-id-admin> อยู่ใน allowlist)]
.ping

[bot]
pong

ถ้า chat_id ของคนที่ทักไม่อยู่ใน allowlist bot จะ silent drop และ log ลง bot.log เป็น "reject chat_id=..." ไม่ตอบกลับเลยสักคำ เพื่อกัน enumeration ไม่ให้ใครเดาว่ามี bot อยู่จริงหรือไม่

คำสั่งพื้นฐาน — read-only

หัวใจของฝั่ง read-only คือ SSH connection pool ที่ persist ไว้ บวกฟังก์ชัน helper ที่รัน command บน web1 โดยส่ง argument เป็น list เสมอ ไม่เคยต่อ string เอง

import asyncssh

# persistent SSH connection pool
_ssh_conn = None
async def get_ssh():
    global _ssh_conn
    if _ssh_conn is None or _ssh_conn.is_closed():
        _ssh_conn = await asyncssh.connect(
            host='<web1-ip>',
            port=22,
            username='botuser',
            client_keys=['/opt/srv-bot/.ssh/web1_key'],
            known_hosts='/opt/srv-bot/.ssh/known_hosts',
        )
    return _ssh_conn

async def web1_run(cmd: list, timeout: float = 30):
    """รัน command บน web1 ผ่าน SSH — argument list (ไม่ใช่ string concat)"""
    ssh = await get_ssh()
    quoted = ' '.join(asyncssh.misc.shlex.quote(a) for a in cmd)
    result = await ssh.run(quoted, check=False, timeout=timeout)
    return result.exit_status, (result.stdout or '') + (result.stderr or '')

SERVICES = ['nginx', 'httpd', 'directadmin', 'exim', 'dovecot',
            'mysqld', 'csf', 'lfd']

@dp.message(F.text.startswith('.status'))
async def cmd_status(m: Message):
    if not await is_allowed(m): return
    lines = []
    for svc in SERVICES:
        rc, out = await web1_run(['sudo', 'systemctl', 'is-active', svc])
        icon = 'OK' if rc == 0 else 'DOWN'
        lines.append(f"{icon} {svc}: {out.strip()}")
    await m.reply("\n".join(lines))

ชุดคำสั่ง basic ที่ผมวางไว้มีเท่านี้ ทุกตัวเป็น read-only ที่วิ่งผ่าน sudo บน web1 ตาม allowlist ที่ตั้งไว้

ตัวอย่าง output จริงที่ bot ตอบกลับในแต่ละคำสั่ง ค่าทั้งหมดเป็น dummy

[คุณ]
.status

[bot]
OK nginx: active
OK httpd: active
OK directadmin: active
OK exim: active
OK dovecot: active
OK mysqld: active
OK csf: active
OK lfd: active

[คุณ]
.load

[bot]
uptime: 14:32:17 up 5 days, load avg 1.84, 1.62, 1.45

top CPU:
  USER     PID  %CPU  COMMAND
  user01  4521  18.3  php-fpm: pool user01
  user02  3847  12.1  php-fpm: pool user02
  mysql   1023   8.7  /usr/sbin/mysqld
  user03  5102   6.4  php-fpm: pool user03
  user01  4598   5.9  php-fpm: pool user01

[คุณ]
.disk

[bot]
Filesystem      Size  Used Avail Use% Mounted
/dev/sda2       180G   72G  104G  41% /
/dev/sda3       720G  482G  208G  70% /home
/dev/sda4        50G   18G   30G  37% /var
/dev/sda5        20G  3.4G   16G  18% /tmp
tmpfs           4.0G  124M  3.9G   3% /run

[คุณ]
.queue

[bot]
mail queue: 27 messages (0 frozen)

[คุณ]
.whois 203.0.113.45

[bot]
IP: 203.0.113.45
csf.deny: yes  (added 2026-04-12 18:42)
  reason: "lfd: blocked due to 10 SSH login failures"
temp_deny: no
temp_allow: no
csf.allow: no

[คุณ]
.tail lfd 5

[bot]
/var/log/lfd.log (last 5)

Apr 12 14:18:02 web1 lfd[2104]: (sshd) Failed login from 203.0.113.12 (...): 5 in last 300s - Blocked
Apr 12 14:21:55 web1 lfd[2104]: *Account Modification* root via ssh from <bak1-ip>
Apr 12 14:24:13 web1 lfd[2104]: (mod_security) ModSecurity triggered by 203.0.113.45 (cnt=4) - Blocked
Apr 12 14:28:41 web1 lfd[2104]: (sshd) Failed login from 203.0.113.77 (...): 6 in last 300s - Blocked
Apr 12 14:30:02 web1 lfd[2104]: Suspicious process running under nobody - pid 9182

Block/Unblock IP — mutation ตัวเดียวที่อนุญาต

mutation ตัวเดียวที่ผมเปิดให้ bot ทำได้คือ block/unblock IP ซึ่งเป็นการแตะที่ระดับ network layer ไม่ยุ่งกับ process หรือ service และ reversible ด้วยคำสั่งเดียว แต่ก่อนรัน command จริงต้อง validate input อย่างเข้มงวด ไม่งั้นถ้า attacker ยึด chat_id ของผมได้ เช่นเครื่องผมโดน hijack เขาอาจส่ง shell injection ทะลุเข้าไปได้

import re

RE_IPV4 = re.compile(r'^(\d{1,3}\.){3}\d{1,3}$')

def validate_ip(ip: str) -> bool:
    if not RE_IPV4.match(ip): return False
    parts = [int(x) for x in ip.split('.')]
    return all(0 <= p <= 255 for p in parts)

def validate_reason(s: str) -> bool:
    # reject shell metachar ทุกชนิด, จำกัดความยาว
    if len(s) > 100: return False
    if re.search(r'[`$<>|&;\n\r\\]', s): return False
    return True

@dp.message(F.text.startswith('.block'))
async def cmd_block(m: Message):
    if not await is_allowed(m): return
    if m.chat.id not in ADMINS:
        await m.reply("admin only")
        return
    parts = m.text.split(maxsplit=2)
    if len(parts) < 3:
        await m.reply("usage: .block <IP> <reason>")
        return
    _, ip, reason = parts
    if not validate_ip(ip):
        await m.reply(f"invalid IP: {ip}")
        return
    if not validate_reason(reason):
        await m.reply("invalid reason (shell metachar / too long)")
        return
    # ส่งเป็น argv list — asyncssh จัดการ quoting ให้
    rc, out = await web1_run(['sudo', 'csf', '-d', ip, reason])
    await audit(m, '.block', ip, rc, out)
    if rc == 0:
        await m.reply(f"blocked {ip}")
    else:
        await m.reply(f"csf -d returned {rc}\n{out[:200]}")

หัวใจของ validation มีสี่ชั้น ชั้นแรกคือส่งเป็น argv list ตลอด ['sudo', 'csf', '-d', ip, reason] ไม่ใช่ f"sudo csf -d {ip}" เพราะ f-string ที่เข้า shell คือ injection ได้ทันที ชั้นที่สองคือ regex check ทุก argument IP ต้องเป็น dotted quad ส่วน reason ห้ามมี backtick, $, ;, |, &, <, > หรือ newline ชั้นที่สามคือ length cap กัน buffer flood และชั้นสุดท้ายคือ admin gate mutation ต้องเป็น chat_id ใน admin_chat_ids เท่านั้น allowed user ทั่วไปทำได้แค่อ่าน

ฝั่ง unblock เขียนคล้ายกัน สั้นกว่าเพราะไม่มี reason

@dp.message(F.text.startswith('.unblock'))
async def cmd_unblock(m: Message):
    if not await is_allowed(m): return
    if m.chat.id not in ADMINS:
        await m.reply("admin only"); return
    parts = m.text.split()
    if len(parts) != 2:
        await m.reply("usage: .unblock <IP>"); return
    _, ip = parts
    if not validate_ip(ip):
        await m.reply(f"invalid IP: {ip}"); return
    rc, out = await web1_run(['sudo', 'csf', '-dr', ip])
    await audit(m, '.unblock', ip, rc, out)
    await m.reply(f"{'OK' if rc==0 else 'FAIL'} csf -dr {ip}\n{out[:300]}")

ตัวอย่างการคุยที่ครอบทุกเคส ตั้งแต่ปฏิเสธไปจนถึงสำเร็จ ช่วยให้เห็นว่า validation แต่ละชั้นทำงานตรงไหน

เคสแรก คนที่ allowed แต่ไม่ใช่ admin ลอง block bot ปฏิเสธทันที

[user ปกติ (chat_id <chat-id-observer> — allowed แต่ไม่ admin)]
.block 203.0.113.45 spam

[bot]
admin only

เคสที่สอง admin ใส่ IP format ผิด bot ปฏิเสธก่อนแตะ shell

[admin]
.block 203.0.113.999 spam

[bot]
invalid IP: 203.0.113.999

เคสที่สาม admin ใส่ reason ที่มี shell metachar bot ปฏิเสธ ชั้นนี้กันทั้ง attacker และกันคนพิมพ์ backtick พลาดเอง

[admin]
.block 203.0.113.45 $(rm -rf /)

[bot]
invalid reason (shell metachar / too long)

เคสที่สี่ block สำเร็จ

[admin]
.block 203.0.113.45 wp brute force /wp-login

[bot]
blocked 203.0.113.45

เคสที่ห้า block ซ้ำ csf รายงานว่า IP นี้มีอยู่แล้ว

[admin]
.block 203.0.113.45 retry

[bot]
csf -d returned 0
*WARNING* 203.0.113.45 already exists in /etc/csf/csf.deny.

เคสสุดท้าย ปลดบล็อกเมื่อตรวจแล้วว่าไม่ใช่ attacker จริง

[admin]
.unblock 203.0.113.45

[bot]
OK csf -dr 203.0.113.45
Removing rule...
DROP   all opt -- in !lo out *  203.0.113.45  -> 0.0.0.0/0
csf: 203.0.113.45 removed from /etc/csf/csf.deny

Audit log — SQLite

ทุก command ที่เป็น mutation ผมเก็บลง SQLite เพื่อให้ trace ย้อนหลังได้ว่าใครสั่งอะไรเมื่อไหร่ ผลลัพธ์เป็นยังไง SQLite ในโหมด WAL รองรับ concurrent write ได้พอสำหรับงานระดับนี้

import aiosqlite

AUDIT_DB = '/opt/srv-bot/logs/audit.sqlite'

async def audit_init():
    async with aiosqlite.connect(AUDIT_DB) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS commands(
                ts        TEXT NOT NULL,
                chat_id   INTEGER NOT NULL,
                username  TEXT,
                cmd       TEXT NOT NULL,
                args      TEXT,
                rc        INTEGER,
                out_short TEXT
            )
        """)
        await db.execute("CREATE INDEX IF NOT EXISTS idx_ts ON commands(ts)")
        await db.execute("PRAGMA journal_mode=WAL")
        await db.commit()

async def audit(m: Message, cmd: str, args: str, rc: int, out: str):
    from datetime import datetime
    async with aiosqlite.connect(AUDIT_DB) as db:
        await db.execute(
            "INSERT INTO commands VALUES(?,?,?,?,?,?,?)",
            (datetime.utcnow().isoformat(), m.chat.id, m.from_user.username,
             cmd, args, rc, out[:500])
        )
        await db.commit()

เวลาอยากดูย้อนหลังก็ query ตรงจาก sqlite3

sqlite3 /opt/srv-bot/logs/audit.sqlite \
  "SELECT ts, username, cmd, args, rc FROM commands ORDER BY ts DESC LIMIT 20"

ไฟล์ครบชุดสำหรับ deploy — copy & paste

รวมทุกอย่างจาก section ก่อนหน้าเป็นไฟล์ที่ใช้งานได้จริง วางที่ /opt/srv-bot/ บน bak1 เริ่มจาก requirements.txt

aiogram==3.13.0
asyncssh==2.18.0
aiosqlite==0.20.0
loguru==0.7.2
PyYAML==6.0.2

config/config.yaml

telegram:
  # อ่าน token จาก env var BOT_TOKEN (ตั้งใน systemd EnvironmentFile)
  allowed_chat_ids:
    - <chat-id-admin>     # admin
    - <chat-id-observer>  # observer (allowed read แต่ block ไม่ได้)
  admin_chat_ids:
    - <chat-id-admin>
  alert_chat_id: <chat-id-admin>

ssh:
  host: <web1-ip>
  port: 22
  user: botuser
  key_file: /opt/srv-bot/.ssh/web1_key
  known_hosts: /opt/srv-bot/.ssh/known_hosts
  cmd_timeout: 30

audit:
  db_path: /opt/srv-bot/logs/audit.sqlite

services:
  - nginx
  - httpd
  - directadmin
  - exim
  - dovecot
  - mysqld
  - csf
  - lfd

bot.py แบบ single file ก๊อปทั้งก้อนไปใช้ได้เลย รวม handler ทุกตัวที่กระจายอยู่ในบทความเข้าด้วยกัน

import os, re, asyncio, yaml
from datetime import datetime
from aiogram import Bot, Dispatcher, F
from aiogram.filters import Command
from aiogram.types import Message
from loguru import logger
import asyncssh
import aiosqlite

# --- config ---
CFG = yaml.safe_load(open('config/config.yaml'))
ALLOWED = set(CFG['telegram']['allowed_chat_ids'])
ADMINS  = set(CFG['telegram']['admin_chat_ids'])
SSH_CFG = CFG['ssh']
SERVICES = CFG['services']
AUDIT_DB = CFG['audit']['db_path']

# --- validation ---
RE_IPV4 = re.compile(r'^(\d{1,3}\.){3}\d{1,3}$')
RE_LOG  = re.compile(r'^(lfd|exim|auth|modsec)$')

LOG_MAP = {
    'lfd':    '/var/log/lfd.log',
    'exim':   '/var/log/exim/mainlog',
    'auth':   '/var/log/auth.log',
    'modsec': '/var/log/nginx/modsec_audit.log',
}

def valid_ip(ip):
    if not RE_IPV4.match(ip): return False
    return all(0 <= int(p) <= 255 for p in ip.split('.'))

def valid_reason(s):
    return len(s) <= 100 and not re.search(r'[`$<>|&;\n\r\\]', s)

# --- audit ---
async def audit_init():
    async with aiosqlite.connect(AUDIT_DB) as db:
        await db.execute("""CREATE TABLE IF NOT EXISTS commands(
            ts TEXT, chat_id INT, username TEXT, cmd TEXT,
            args TEXT, rc INT, out_short TEXT)""")
        await db.execute("CREATE INDEX IF NOT EXISTS idx_ts ON commands(ts)")
        await db.execute("PRAGMA journal_mode=WAL")
        await db.commit()

async def audit(m, cmd, args, rc, out):
    async with aiosqlite.connect(AUDIT_DB) as db:
        await db.execute("INSERT INTO commands VALUES(?,?,?,?,?,?,?)",
            (datetime.utcnow().isoformat(), m.chat.id,
             m.from_user.username, cmd, args, rc, out[:500]))
        await db.commit()

# --- SSH pool ---
_ssh = None
async def get_ssh():
    global _ssh
    if _ssh is None or _ssh.is_closed():
        _ssh = await asyncssh.connect(
            host=SSH_CFG['host'], port=SSH_CFG['port'],
            username=SSH_CFG['user'],
            client_keys=[SSH_CFG['key_file']],
            known_hosts=SSH_CFG['known_hosts'])
    return _ssh

async def web1(*args, timeout=None):
    """รัน command บน web1 — args เป็น list ไม่ใช่ string concat"""
    ssh = await get_ssh()
    t = timeout or SSH_CFG['cmd_timeout']
    quoted = ' '.join(asyncssh.misc.shlex.quote(a) for a in args)
    r = await ssh.run(quoted, check=False, timeout=t)
    return r.exit_status, (r.stdout or '') + (r.stderr or '')

# --- auth gate ---
async def allowed(m):
    if m.chat.id not in ALLOWED:
        logger.warning(f"reject chat_id={m.chat.id} user={m.from_user.username}")
        return False
    return True

async def admin(m):
    if not await allowed(m): return False
    if m.chat.id not in ADMINS:
        await m.reply("admin only"); return False
    return True

# --- handlers ---
bot = Bot(token=os.environ['BOT_TOKEN'])
dp  = Dispatcher()

@dp.message(Command('start'))
async def h_start(m):
    await m.reply(f"chat_id = {m.chat.id}\nadd to allowed_chat_ids ใน config.yaml")

@dp.message(F.text.startswith('.ping'))
async def h_ping(m):
    if not await allowed(m): return
    await m.reply("pong")

@dp.message(F.text.startswith('.status'))
async def h_status(m):
    if not await allowed(m): return
    lines = []
    for s in SERVICES:
        rc, out = await web1('sudo', 'systemctl', 'is-active', s)
        lines.append(f"{'OK' if rc==0 else 'DOWN'} {s}: {out.strip()}")
    await m.reply("\n".join(lines))

@dp.message(F.text.startswith('.load'))
async def h_load(m):
    if not await allowed(m): return
    _, up = await web1('uptime')
    _, top = await web1('bash', '-c',
        "ps axo user:12,pid,pcpu,comm --sort=-pcpu | head -6")
    await m.reply(f"{up}\ntop CPU:\n{top}")

@dp.message(F.text.startswith('.disk'))
async def h_disk(m):
    if not await allowed(m): return
    _, out = await web1('df', '-h', '-x', 'tmpfs', '-x', 'devtmpfs')
    await m.reply(out)

@dp.message(F.text.startswith('.queue'))
async def h_queue(m):
    if not await allowed(m): return
    _, n  = await web1('sudo', 'exim', '-bpc')
    _, fz = await web1('bash', '-c', "sudo exim -bp | grep -c frozen || true")
    await m.reply(f"mail queue: {n.strip()} messages ({fz.strip()} frozen)")

@dp.message(F.text.startswith('.whois'))
async def h_whois(m):
    if not await allowed(m): return
    parts = m.text.split()
    if len(parts) != 2 or not valid_ip(parts[1]):
        await m.reply("usage: .whois <IP>"); return
    _, out = await web1('sudo', 'csf', '-g', parts[1])
    await m.reply(out or "(no info)")

@dp.message(F.text.startswith('.tail'))
async def h_tail(m):
    if not await allowed(m): return
    parts = m.text.split()
    if len(parts) < 2 or not RE_LOG.match(parts[1]):
        await m.reply(f"usage: .tail <{'/'.join(LOG_MAP)}> [N]"); return
    n = '20'
    if len(parts) >= 3 and parts[2].isdigit():
        n = str(min(int(parts[2]), 100))
    _, out = await web1('sudo', 'tail', '-n', n, LOG_MAP[parts[1]])
    await m.reply(f"{LOG_MAP[parts[1]]} (last {n})\n\n{out[:3500]}")

@dp.message(F.text.startswith('.block'))
async def h_block(m):
    if not await admin(m): return
    parts = m.text.split(maxsplit=2)
    if len(parts) < 3:
        await m.reply("usage: .block <IP> <reason>"); return
    _, ip, reason = parts
    if not valid_ip(ip):    await m.reply(f"invalid IP: {ip}"); return
    if not valid_reason(reason):
        await m.reply("invalid reason (shell metachar / too long)"); return
    rc, out = await web1('sudo', 'csf', '-d', ip, reason)
    await audit(m, '.block', f"{ip} {reason}", rc, out)
    if rc == 0 and 'already exists' not in out:
        await m.reply(f"blocked {ip}")
    else:
        await m.reply(f"csf -d returned {rc}\n{out[:300]}")

@dp.message(F.text.startswith('.unblock'))
async def h_unblock(m):
    if not await admin(m): return
    parts = m.text.split()
    if len(parts) != 2 or not valid_ip(parts[1]):
        await m.reply("usage: .unblock <IP>"); return
    rc, out = await web1('sudo', 'csf', '-dr', parts[1])
    await audit(m, '.unblock', parts[1], rc, out)
    await m.reply(f"{'OK' if rc==0 else 'FAIL'} csf -dr {parts[1]}\n{out[:300]}")

# --- main ---
async def main():
    await audit_init()
    logger.add('logs/bot.log', rotation='10 MB', retention='30 days')
    logger.info(f"bot starting — {len(ALLOWED)} allowed, {len(ADMINS)} admin")
    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())

Setup script ที่รันครั้งเดียวในนาม root บน bak1 ช่วยวาง user, venv, key และ env file ให้ครบในรวดเดียว

#!/bin/bash
set -e

# 1. user + dirs
useradd -r -s /bin/false -d /opt/srv-bot botuser
mkdir -p /opt/srv-bot/{logs,config,.ssh} /etc/srv-bot
chown -R botuser:botuser /opt/srv-bot
chmod 750 /opt/srv-bot

# 2. python venv + deps (วาง requirements.txt ก่อน)
apt install python3-venv -y
sudo -u botuser python3 -m venv /opt/srv-bot/.venv
sudo -u botuser /opt/srv-bot/.venv/bin/pip install -r /opt/srv-bot/requirements.txt

# 3. SSH key + known_hosts
sudo -u botuser ssh-keygen -t ed25519 -f /opt/srv-bot/.ssh/web1_key -N ""
sudo -u botuser ssh-keyscan -t ed25519 -p 22 <web1-ip> \
    > /opt/srv-bot/.ssh/known_hosts

# 4. token in env file
echo "BOT_TOKEN=<bot-token>" \
    > /etc/srv-bot/secrets.env
chmod 600 /etc/srv-bot/secrets.env
chown root:botuser /etc/srv-bot/secrets.env
chmod 640 /etc/srv-bot/secrets.env

# 5. systemd unit (วางไฟล์ /etc/systemd/system/srv-bot.service ก่อน — section 6)
systemctl daemon-reload
systemctl enable srv-bot

# 6. ตอนนี้:
#    - cat /opt/srv-bot/.ssh/web1_key.pub  -> เอาไปใส่ใน authorized_keys ที่ web1
#    - แก้ /etc/srv-bot/secrets.env ใส่ token จริง
#    - แก้ config/config.yaml ใส่ chat_ids จริง
#    - แล้ว systemctl start srv-bot

ทดสอบ + start service

พอวางไฟล์ครบ ก็ enable แล้ว start service ตามด้วยดู log สองทาง

# start
systemctl daemon-reload
systemctl enable srv-bot
systemctl start srv-bot

# ดู log
journalctl -u srv-bot -f
# หรือ
tail -f /opt/srv-bot/logs/bot.log

จากนั้นเปิด Telegram ทัก bot ที่ตั้งไว้ พิมพ์ /start bot จะตอบ chat_id ของเรา ถ้า chat_id อยู่ใน allowlist แล้วลอง .ping ได้ "pong" กลับมาก็ถือว่าใช้ได้

ถ้า .ping ไม่ตอบ ให้ไล่เช็คเป็นชั้น

การคุยตอน first run ที่ควรจะได้เห็นเป็นแบบนี้

[คุณ (ทัก @example_monitor_bot ครั้งแรก)]
/start

[bot]
chat_id = <chat-id-admin>
add to allowed_chat_ids ใน config.yaml

[คุณ — หลังแก้ config + systemctl restart srv-bot]
.ping

[bot]
pong

[คุณ]
.status

[bot]
OK nginx: active
OK httpd: active
OK directadmin: active
OK exim: active
OK dovecot: active
OK mysqld: active
OK csf: active
OK lfd: active

ถึงตรงนี้ bot ใช้งานได้แล้ว ทุกคำสั่งพื้นฐานพร้อม ลอง .load, .disk, .queue, .whois <IP> ดู output ได้เลย

สิ่งที่ไม่อยู่ในบทความนี้

โดย design บทความนี้จงใจตัดส่วนต่อไปนี้ออก เพื่อให้ scope อยู่ที่ "basic monitoring ที่ปลอดภัย" ไม่บานปลายเป็นระบบใหญ่ที่ต้องดูแลแยกอีกชั้น

mutation อื่นนอกจาก block/unblock อย่าง kill process, restart service หรือ php-fpm reload ทำได้ แต่ต้องผ่าน wrapper script ที่ check argument allowlist บวก confirmation flow ผมตั้งใจแยกไปเป็น Phase 2 ที่จะเขียนทีหลัง ส่วน auto-action กับ auto-recovery ผมไม่ใส่เลยแม้แต่ block IP เอง คนต้องเป็นคนกดยืนยันทุกครั้ง inline keyboard confirmation แบบปุ่มยืนยันพร้อม TTL ของจริงผมใช้ใน .block แต่ตัดออกจากบทความเพื่อให้ code สั้นพอจะอ่านจบ เช่นเดียวกับ auto-alert engine ที่มี poll loop, sustained-window dedup และ severity tier กับ YAML-defined analyst command สำหรับงาน forensic ลึก เก็บไว้บทความถัดไป

การวาง Telegram bot ให้ดูแล server ได้โดยไม่เพิ่ม attack surface กุญแจอยู่ที่ scope ที่จำกัด มากกว่า code ที่ซับซ้อน bot ที่ทำได้แค่ csf -d กับ csf -dr บวกอ่าน log ตาม allowlist บวกเช็ค service status เป็น control surface ที่ blast radius เล็กมาก เทียบกับการถือ SSH key ผ่านมือถือที่เปิดทุกอย่าง

หลักง่ายๆ ที่ผมใช้คือเริ่มจาก observer บวก IP-layer mutation เท่านั้น แล้วค่อยเปิด process หรือ service mutation ทีหลังเมื่อ wrapper script, audit และ confirmation พร้อม ใครมี pattern หรือ trick เพิ่ม เช่นใช้ CrowdSec แทน CSF, ทำ multi-host support หรือ integrate กับ Prometheus alertmanager ก็ลอง comment แชร์กันได้ จะได้เอาไปปรับใช้บ้าง ตามสะดวก หลักการเหมือนกันหมด ตรงไหนผิดพลาดผมจะกลับมาแก้ในภายหลังครับ