A Telegram Bot for DA Server Monitoring with a Small Blast Radius

When I work with a DirectAdmin server hosting multiple users on one machine, the most frustrating problem is not specs or tuning, but "when I find out something is wrong." I usually find out when a customer calls, which means I am always one step behind. Using SSH from a phone makes typing awkward, lacks context, and risks touching something dangerous on a small screen. Setting up a full dashboard means adding another service to maintain and another port to open.

So I tried setting up a Telegram bot as a lightweight control surface for checking status, inspecting logs, and blocking IPs from a chat. This article covers the minimum setup I actually use, deliberately limited to read-only commands plus IP blocking/unblocking. No killing processes, no restarting services, and no automatic actions at all. I intend the bot to be just an observer with mutations at the IP layer, keeping its blast radius as small as possible.

Why I Chose a Telegram Bot

Before settling on Telegram, I compared four main options. Each had its own drawbacks.

I settled on Telegram for four reasons. First, I do not need to open any ports: the bot uses long-polling to reach the Telegram API, so I do not have to touch the inbound firewall rules. Second, it avoids NAT problems: my phone does not need a public IP or a VPN to receive push notifications. Third, it is a real two-way conversation, unlike one-way email alerts, so I can send commands back. Finally, I can manage auth myself with a chat_id allowlist, which provides protection on the same level as an SSH key.

The "Observer first, operator later" Principle

This is the first rule I follow with every bot, and it is what makes this bot safe enough to accept commands from Telegram without losing sleep. The bot's first phase is read-only plus IP blocking. It can read logs, check status, and push alerts; its only mutations are csf -d and csf -dr for blocking and unblocking IPs, reversible with a single command.

One thing I am very careful about is keeping dangerous wildcards out of sudoers: absolutely no kill -9 * or systemctl restart *. Those wildcards mean the bot can kill init or restart sshd, effectively bricking the machine. Another rule is that the bot does not act automatically: it only alerts and suggests commands, and a person decides to execute them every time. With automatic actions enabled, a single false positive can affect multiple customers at once.

Architecture — Where the Bot Runs

The next question is where the bot should run. I compared three options. Running it on web1 alongside DA gives it local access with zero latency, but the serious drawback is that if web1 goes down, the bot goes down too, right when I need it most. It also adds a service to a machine that is already publicly accessible. Running it on a separate external VPS gives me a useful outside view, but adds cost and maintenance, and means keeping the key outside my own infrastructure.

I chose the middle ground: running the bot on bak1, the same backup host from the previous article. The bot connects to web1 over SSH using a key. The most important benefit is that if web1 goes down, the bot on the other machine can still immediately report "web1 unreachable." The tradeoff is an extra SSH hop of only around a hundred milliseconds, which is well worth it.

+-------------+        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 communicates with bak1 using only outbound HTTPS on port 443, while bak1 connects to web1 over SSH restricted by from=. Neither connection requires opening an inbound port on the firewall.

Register a Bot Through BotFather

Telegram lets me register a bot through its central bot, @BotFather. Open Telegram, search for @BotFather, press start, then enter these commands in order.

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

Treat the token like a password. Anyone who gets it can immediately control the bot in my place. Never commit it to git, paste it during screen sharing, or put it in a file readable by the web user.

Additional settings are available through /setdescription for the bot description, /setabouttext for the about page, and /setcommands to register a command list for Telegram autocomplete.

/setcommands

For /setcommands, I use the following list. One detail to watch is that the bot's actual commands use . as their prefix, while the BotFather list requires / because it enforces its own 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

The actual commands use the . prefix instead of / because I do not want Telegram autocomplete cluttering the screen every time I type. /start is an exception because the Telegram protocol requires /.

Find Your chat_id — Set the Allowlist

This bot does not use OAuth. My auth is entirely a chat_id allowlist: a list of Telegram chat_id values allowed to talk to the bot. If someone is not on the list, the bot silently drops their messages without replying. There are two ways to find your chat_id. The first is to message @userinfobot in Telegram, which replies with an Id. The second is to send /start to your own bot once it is up, because I coded it to reply with the sender's chat_id.

Suppose my chat_id is <chat-id-admin> and the other admin's chat_id is <chat-id-observer>. Here is an example of the first conversation with the bot.

[คุณ]
/start

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

Put the chat_id values into 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

The distinction to keep clear is that allowed_chat_ids lists people who can read, while admin_chat_ids is the subset allowed to perform mutations such as blocking/unblocking. Someone in allowed but not in admin can check status but cannot block IPs.

Set Up the Bot User + systemd Hardening on bak1

The bot must never run as root. Create a separate user named botuser with no shell and no unrestricted sudo access.

# สร้าง 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

Then write a systemd unit at /etc/systemd/system/srv-bot.service with the full set of hardening directives.

[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

The hardening directives at the end are the core of this setup. NoNewPrivileges prevents the process from elevating its own privileges, ProtectSystem=strict makes most of the filesystem read-only, ProtectHome hides /home from the bot, and ReadWritePaths allowlists the only two directories the bot can actually write to.

Store the token separately in a root-owned secrets file, with group access allowing the bot to read it.

# /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 from bak1 to web1 — Least Privilege

The bot on bak1 sends commands to web1 over SSH using a key. The destination user on web1 is also named botuser, but this user has sudoers rules restricted to the commands the bot actually needs. First, create a key on bak1 as botuser.

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

Then place the public key on web1 in /home/<botuser>/.ssh/authorized_keys, restricting it with from= and disabling all 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>" means this key works only when the connection comes from bak1. If the key leaks onto the internet, it cannot be used. Disabling all forwarding and adding no-pty means the key can only run commands directly; it cannot open a tunnel or an interactive shell.

On web1, write the sudoers rules in /etc/sudoers.d/botuser-readonly, listing only the commands the bot needs.

# 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 Handling

My stack is async throughout: aiogram 3.x is an async-native Telegram bot framework, asyncssh is an SSH client that can persist connections without a new handshake for every command, loguru handles structured logging, aiosqlite handles the audit log, and PyYAML reads the config. The minimum file structure looks like this.

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

Here is a minimal bot.py that accepts .ping with an auth check, to show the structure first.

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())

The point to understand is that /start replies with the chat_id to anyone who messages it without going through is_allowed, deliberately allowing new people to find their own chat_id. Every other command must pass the auth gate.

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

[bot]
pong

If the sender's chat_id is not in the allowlist, the bot silently drops the message and logs "reject chat_id=..." to bot.log without replying at all, preventing enumeration so that nobody can determine whether the bot actually exists.

Basic Commands — Read-Only

The core of the read-only side is a persistent SSH connection pool, plus a helper function that runs commands on web1, always passing arguments as a list and never concatenating strings manually.

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))

These are the basic commands I set up. All are read-only and run through sudo on web1 according to the configured allowlist.

Here are examples of the actual output the bot returns for each command. All values are dummy values.

[คุณ]
.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 — The Only Allowed Mutation

The only mutation I allow the bot to perform is blocking/unblocking IPs, which operates at the network layer, leaves processes and services alone, and is reversible with a single command. But input must be strictly validated before running the actual command. Otherwise, if an attacker takes over my chat_id, for example by hijacking my machine, they could get shell injection through.

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 has four core layers. First, always pass an argv list, ['sudo', 'csf', '-d', ip, reason], rather than f"sudo csf -d {ip}", because an f-string passed to a shell immediately allows injection. Second, regex-check every argument: the IP must be a dotted quad, and the reason must not contain a backtick, $, ;, |, &, <, >, or a newline. Third, impose a length cap to prevent buffer flooding. The final layer is the admin gate: mutations require a chat_id in admin_chat_ids; ordinary allowed users can only read.

The unblock handler is similar, but shorter because it has no 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]}")

These conversation examples cover every case, from rejection to success, showing where each validation layer takes effect.

First case: an allowed user who is not an admin tries to block an IP. The bot rejects it immediately.

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

[bot]
admin only

Second case: an admin enters an incorrectly formatted IP. The bot rejects it before reaching the shell.

[admin]
.block 203.0.113.999 spam

[bot]
invalid IP: 203.0.113.999

Third case: an admin enters a reason containing a shell metacharacter. The bot rejects it. This layer protects against both attackers and someone accidentally typing a backtick.

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

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

Fourth case: blocking succeeds.

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

[bot]
blocked 203.0.113.45

Fifth case: a repeated block. csf reports that the IP already exists.

[admin]
.block 203.0.113.45 retry

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

Final case: unblocking after checking that it is not actually an 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

I store every mutation command in SQLite so I can trace who ran what, when, and what the result was. SQLite in WAL mode supports enough concurrent writes for this level of workload.

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()

When I want to review the history, I query it directly with sqlite3.

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

Complete Deployment Files — Copy & Paste

This combines everything from the previous sections into working files. Place them in /opt/srv-bot/ on bak1, starting with 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

Here is bot.py as a single file, ready to copy in full, combining all the handlers spread throughout the article.

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())

A setup script, run once as root on bak1, creates the user, venv, key, and env file in one pass.

#!/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

Test + Start the Service

Once all the files are in place, enable and start the service, then check both logs.

# 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

Then open Telegram and message the bot you configured. Enter /start, and the bot will reply with your chat_id. If the chat_id is already in the allowlist, try .ping; getting "pong" back means it is working.

If .ping gets no reply, check each layer in order.

The first-run conversation should look like this.

[คุณ (ทัก @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

At this point, the bot is working and all the basic commands are ready. Try .load, .disk, .queue, and .whois <IP> to see their output.

What This Article Leaves Out

By design, I have deliberately excluded the following to keep the scope at "safe basic monitoring" and avoid turning it into a large system that needs another layer of maintenance.

Mutations beyond blocking/unblocking, such as killing processes, restarting services, or reloading php-fpm, are possible, but they must go through a wrapper script that checks an argument allowlist plus a confirmation flow. I deliberately set these aside for Phase 2, which I will write about later. I do not include automatic actions or automatic recovery at all, even automatic IP blocking: a person must confirm every action. I use inline keyboard confirmation with a confirmation button and TTL in the real .block implementation, but left it out of this article to keep the code short enough to read through. The same goes for the auto-alert engine with its poll loop, sustained-window dedup, and severity tiers, as well as YAML-defined analyst commands for deeper forensic work; those are for the next article.

The key to setting up a Telegram bot to manage a server without increasing the attack surface is a limited scope, more than complex code. A bot that can only run csf -d and csf -dr, read allowlisted logs, and check service status is a control surface with a very small blast radius compared with carrying an SSH key on a phone that gives access to everything.

My simple rule is to start with an observer plus IP-layer mutations only, then enable process or service mutations later when the wrapper scripts, auditing, and confirmation are ready. If you have other patterns or tricks, such as using CrowdSec instead of CSF, adding multi-host support, or integrating with Prometheus alertmanager, feel free to share them in the comments so I can try them too. The principles are the same throughout. If anything is wrong, I will come back and correct it later.