全栈自动化与纵深防御：构建基于 Cloudflare、Docker 与 Shlink 的生产级短链系统

# 安装 UFW
sudo apt update
sudo apt install ufw

# 默认规则：拒绝所有入站
sudo ufw default deny incoming
sudo ufw default allow outgoing

# 允许 SSH (务必先配置，否则会被锁在外面)
sudo ufw allow 22/tcp

# 下载并应用 Cloudflare IP 范围
# IPv4
curl https://www.cloudflare.com/ips-v4 | while read ip; do
  sudo ufw allow from $ip to any port 80,443 proto tcp
done

# IPv6
curl https://www.cloudflare.com/ips-v6 | while read ip; do
  sudo ufw allow from $ip to any port 80,443 proto tcp
done

# 启用防火墙
sudo ufw enable

# 查看状态
sudo ufw status verbose

sudo mkdir -p /etc/ssl/cloudflare
sudo nano /etc/ssl/cloudflare/cert.pem
# 粘贴证书内容

sudo nano /etc/ssl/cloudflare/key.pem
# 粘贴私钥内容

sudo chmod 600 /etc/ssl/cloudflare/key.pem

# 安装 Certbot
sudo apt install certbot python3-certbot-nginx

# 获取证书
sudo certbot --nginx -d your.domain

# 证书自动续期
sudo certbot renew --dry-run

# 监听特定端口
sudo tcpdump -i any -n -A -s 0 'port 53842'

# 只抓 HTTP POST 请求
sudo tcpdump -i any -A -s 0 'tcp port 8080 and (tcp[((tcp[12:1] & 0xf0) >> 2):4] = 0x504f5354)'

# 抓取包含特定字符串的包
sudo tcpdump -i any -A -s 0 'tcp port 8080' | grep -A 10 -B 5 "X-Api-Key"

# 保存到文件供 Wireshark 分析
sudo tcpdump -i any -s 0 -w /tmp/capture.pcap 'port 8080'

sqlite3 gokapi.db "SELECT * FROM ApiKeys;"

# 备份数据库
cp gokapi.db gokapi.db.backup

# 修改过期时间为 2099 年
sqlite3 gokapi.db <<EOF
UPDATE ApiKeys 
SET Expiry = 4102444800 
WHERE key = 'abc123...';
.quit
EOF

# 验证
sqlite3 gokapi.db "SELECT key, datetime(Expiry, 'unixepoch') FROM ApiKeys;"

# 检查容器时间
docker exec gokapi date

# 同步宿主机时间到容器
docker run -it --rm \
  -v /etc/localtime:/etc/localtime:ro \
  -v /etc/timezone:/etc/timezone:ro \
  gokapi:latest date

version: '3.8'

services:
  shlink:
    image: shlinkio/shlink:stable
    container_name: shlink
    restart: always  # 容器崩溃自动重启
    
    environment:
      # 核心配置
      - DEFAULT_DOMAIN=your.domain
      - IS_HTTPS_ENABLED=true
      - GEOLITE_LICENSE_KEY=${GEOLITE_KEY}  # 从环境变量读取
      
      # 数据库配置 (使用内置 SQLite)
      - DB_DRIVER=sqlite
      
      # 性能优化
      - INITIAL_API_KEYS=${INITIAL_API_KEY}  # 首次启动自动生成 API Key
      
      # 安全配置
      - DISABLE_TRACKING=false  # 启用访问统计
      - ANONYMIZE_REMOTE_ADDR=true  # 匿名化 IP 地址（GDPR 合规）
      
    ports:
      - "127.0.0.1:8080:8080"  # 只监听本地回环，不暴露到公网
    
    volumes:
      - ./data:/etc/shlink/data  # 数据持久化
    
    healthcheck:
      test: ["CMD", "wget", "--quiet", "--tries=1", "--spider", "http://localhost:8080/rest/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s
    
    networks:
      - shlink-network
    
    # 资源限制（防止内存泄漏）
    deploy:
      resources:
        limits:
          cpus: '0.5'
          memory: 512M
        reservations:
          memory: 256M

networks:
  shlink-network:
    driver: bridge

GEOLITE_KEY=your_maxmind_license_key
INITIAL_API_KEY=your_generated_api_key_here

# Bind Mount (绑定挂载) - 我们使用的方式
volumes:
  - ./data:/etc/shlink/data

# Named Volume (命名卷)
volumes:
  - shlink-data:/etc/shlink/data

volumes:
  shlink-data:
    driver: local

# 启动服务
docker compose up -d

# 创建一个短链接
curl -X POST http://localhost:8080/rest/v3/short-urls \
  -H "X-Api-Key: your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"longUrl": "https://example.com"}'

# 检查数据文件
ls -lh /opt/shlink/data/
# 应该能看到 database.sqlite

# 销毁容器
docker compose down

# 数据文件依然存在
ls -lh /opt/shlink/data/
# database.sqlite 仍然在

# 重新启动，数据恢复
docker compose up -d
# 之前创建的短链接仍然有效

docker exec -it shlink shlink api-key:generate

environment:
  - INITIAL_API_KEYS=my_predefined_key_12345

curl -X GET http://localhost:8080/rest/v3/short-urls \
  -H "X-Api-Key: your_api_key"

{
  "shortUrls": {
    "data": [],
    "pagination": {...}
  }
}

{
  "title": "Invalid API key",
  "type": "https://shlink.io/api/error/invalid-api-key",
  "detail": "Provided API key does not exist or is invalid",
  "status": 401
}

# 启动服务
cd /opt/shlink
docker compose up -d

# 查看日志
docker compose logs -f

# 查看实时日志（最近 50 行）
docker compose logs --tail=50 -f shlink

# 重启服务
docker compose restart

# 停止并删除容器（数据保留）
docker compose down

# 停止并删除容器及数据（危险操作）
docker compose down -v

# 更新镜像
docker compose pull
docker compose up -d

# 查看容器状态
docker compose ps

# 查看资源使用
docker stats shlink

cd /opt/shlink_bot
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import logging
import asyncio
import aiohttp
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes
)
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 配置日志
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

# 配置常量
TELEGRAM_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
SHLINK_API_URL = os.getenv('SHLINK_API_URL', 'http://localhost:8080/rest/v3')
SHLINK_API_KEY = os.getenv('SHLINK_API_KEY')
ALLOWED_USER_IDS = [int(uid) for uid in os.getenv('ALLOWED_USER_IDS', '').split(',') if uid]

# 验证配置
if not all([TELEGRAM_TOKEN, SHLINK_API_KEY]):
    logger.error("缺少必要的环境变量！")
    exit(1)


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """处理 /start 命令"""
    user = update.effective_user
    await update.message.reply_text(
        f'你好 {user.mention_html()}！\n\n'
        '发送任何链接给我，我会为你创建短链接。\n\n'
        '命令列表：\n'
        '/start - 查看帮助\n'
        '/stats - 查看统计信息',
        parse_mode='HTML'
    )


async def create_short_url(long_url: str) -> dict:
    """调用 Shlink API 创建短链接"""
    headers = {
        "X-Api-Key": SHLINK_API_KEY,
        "Content-Type": "application/json"
    }
    
    payload = {
        "longUrl": long_url,
        "validateUrl": False,  # 不验证 URL 有效性
        "findIfExists": True,  # 如果已存在则返回现有短链
    }
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{SHLINK_API_URL}/short-urls",
                json=payload,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return {
                        'success': True,
                        'short_url': data.get('shortUrl'),
                        'long_url': data.get('longUrl')
                    }
                else:
                    error_text = await response.text()
                    logger.error(f"Shlink API 错误: {response.status} - {error_text}")
                    return {
                        'success': False,
                        'error': f'API 返回错误: {response.status}'
                    }
    except asyncio.TimeoutError:
        logger.error("Shlink API 请求超时")
        return {'success': False, 'error': '请求超时，请稍后重试'}
    except Exception as e:
        logger.error(f"创建短链接时发生异常: {e}")
        return {'success': False, 'error': str(e)}


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """处理用户消息"""
    # 权限检查
    if ALLOWED_USER_IDS and update.effective_user.id not in ALLOWED_USER_IDS:
        await update.message.reply_text("❌ 你没有使用此 Bot 的权限。")
        return
    
    text = update.message.text
    
    # 简单的 URL 验证
    if not (text.startswith('http://') or text.startswith('https://')):
        await update.message.reply_text(
            "⚠️ 请发送有效的 URL（以 http:// 或 https:// 开头）"
        )
        return
    
    # 发送处理中消息
    processing_msg = await update.message.reply_text("⏳ 正在生成短链接...")
    
    # 创建短链接
    result = await create_short_url(text)
    
    if result['success']:
        await processing_msg.edit_text(
            f"✅ 短链接创建成功！\n\n"
            f"🔗 短链接: {result['short_url']}\n"
            f"📎 原始链接: {result['long_url'][:50]}..."
        )
    else:
        await processing_msg.edit_text(
            f"❌ 创建失败\n\n"
            f"错误信息: {result.get('error', '未知错误')}"
        )


async def get_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """获取统计信息"""
    if ALLOWED_USER_IDS and update.effective_user.id not in ALLOWED_USER_IDS:
        await update.message.reply_text("❌ 你没有使用此 Bot 的权限。")
        return
    
    headers = {"X-Api-Key": SHLINK_API_KEY}
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{SHLINK_API_URL}/short-urls",
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    total = data.get('shortUrls', {}).get('pagination', {}).get('totalItems', 0)
                    await update.message.reply_text(
                        f"📊 统计信息\n\n"
                        f"短链接总数: {total}"
                    )
                else:
                    await update.message.reply_text("❌ 获取统计信息失败")
    except Exception as e:
        logger.error(f"获取统计信息时发生异常: {e}")
        await update.message.reply_text(f"❌ 错误: {str(e)}")


async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """全局错误处理"""
    logger.error(f"Update {update} caused error {context.error}")
    
    if update and update.effective_message:
        await update.effective_message.reply_text(
            "⚠️ 处理你的请求时发生了错误，请稍后重试。"
        )


def main():
    """主函数"""
    # 创建 Application
    application = Application.builder().token(TELEGRAM_TOKEN).build()
    
    # 注册处理器
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("stats", get_stats))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # 注册错误处理器
    application.add_error_handler(error_handler)
    
    # 启动 Bot
    logger.info("Bot 启动中...")
    application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == '__main__':
    main()

# Telegram Bot Token (从 @BotFather 获取)
TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11

# Shlink 配置
SHLINK_API_URL=http://localhost:8080/rest/v3
SHLINK_API_KEY=your_shlink_api_key_here

# 访问控制（可选，留空则允许所有用户）
ALLOWED_USER_IDS=123456789,987654321

# 同步版本（阻塞）
def create_short_url_sync(url):
    response = requests.post(...)  # 这里会阻塞整个程序
    return response.json()

# 异步版本（非阻塞）
async def create_short_url_async(url):
    async with aiohttp.ClientSession() as session:
        async with session.post(...) as response:  # 可以处理其他请求
            return await response.json()

# 激活虚拟环境
source venv/bin/activate

# 直接运行测试
python bot.py

# 查看详细日志
python bot.py 2>&1 | tee bot.log

# 后台运行
nohup python bot.py > bot.log 2>&1 &

# 创建专用用户（无家目录，无登录 Shell）
sudo useradd -r -s /bin/false -M -d /opt/shlink_bot tgbot_user

# 参数说明：
# -r: 创建系统账户
# -s /bin/false: 禁止登录
# -M: 不创建家目录
# -d: 指定工作目录（但不实际创建）

# 创建工作目录
sudo mkdir -p /opt/shlink_bot

# 转移所有权
sudo chown -R tgbot_user:tgbot_user /opt/shlink_bot

# 设置权限（用户可读写执行，其他人无权限）
sudo chmod 750 /opt/shlink_bot

ls -ld /opt/shlink_bot
# 输出: drwxr-x--- 2 tgbot_user tgbot_user 4096 Jan  2 10:00 /opt/shlink_bot

# 测试是否可以写入
sudo -u tgbot_user touch /opt/shlink_bot/test.txt
sudo -u tgbot_user rm /opt/shlink_bot/test.txt

[Unit]
Description=Telegram Shlink Bot
Documentation=https://github.com/your-repo
After=network-online.target docker.service
Wants=network-online.target
Requires=docker.service

[Service]
Type=simple
User=tgbot_user
Group=tgbot_user
WorkingDirectory=/opt/shlink_bot

# 环境变量（从文件加载）
EnvironmentFile=/opt/shlink_bot/.env

# 启动命令
ExecStart=/opt/shlink_bot/venv/bin/python /opt/shlink_bot/bot.py

# 重启策略
Restart=always
RestartSec=10

# 安全加固
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/shlink_bot

# 资源限制
LimitNOFILE=4096
MemoryLimit=512M
CPUQuota=50%

# 日志配置
StandardOutput=journal
StandardError=journal
SyslogIdentifier=shlink-bot

[Install]
WantedBy=multi-user.target

# 重新加载 systemd 配置
sudo systemctl daemon-reload

# 启动服务
sudo systemctl start shlink-bot

# 设置开机自启
sudo systemctl enable shlink-bot

# 查看状态
sudo systemctl status shlink-bot

# 查看日志（实时）
sudo journalctl -u shlink-bot -f

# 查看最近 100 行日志
sudo journalctl -u shlink-bot -n 100

# 重启服务
sudo systemctl restart shlink-bot

# 停止服务
sudo systemctl stop shlink-bot

PermitRootLogin no
PasswordAuthentication no  # 只允许密钥登录
PubkeyAuthentication yes

sudo systemctl restart sshd

# 安装
sudo apt install fail2ban

# 配置
sudo cp /etc/fail2ban/jail.conf /etc/fail2ban/jail.local
sudo nano /etc/fail2ban/jail.local

# 修改：
[sshd]
enabled = true
port = ssh
filter = sshd
logpath = /var/log/auth.log
maxretry = 3
bantime = 3600

# 启动
sudo systemctl enable fail2ban
sudo systemctl start fail2ban

sudo apt install htop
htop -p $(pgrep -f bot.py)

# 实时监控容器资源
docker stats shlink

# 输出格式化
docker stats --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"

#!/bin/bash

# 检查 Bot 进程
if ! pgrep -f "bot.py" > /dev/null; then
    echo "ERROR: Bot process not running"
    systemctl restart shlink-bot
    exit 1
fi

# 检查 Shlink 容器
if ! docker ps | grep -q shlink; then
    echo "ERROR: Shlink container not running"
    cd /opt/shlink && docker compose up -d
    exit 1
fi

# 检查 API 可用性
if ! curl -s -f http://localhost:8080/rest/health > /dev/null; then
    echo "ERROR: Shlink API not responding"
    exit 1
fi

echo "OK: All services healthy"
exit 0

# 每 5 分钟检查一次
*/5 * * * * /opt/shlink_bot/health_check.sh >> /var/log/health_check.log 2>&1

#!/bin/bash

BACKUP_DIR="/opt/backups"
DATE=$(date +%Y%m%d_%H%M%S)

# 备份 Shlink 数据
tar -czf "${BACKUP_DIR}/shlink_data_${DATE}.tar.gz" /opt/shlink/data

# 备份 Bot 配置
tar -czf "${BACKUP_DIR}/bot_config_${DATE}.tar.gz" /opt/shlink_bot/.env /opt/shlink_bot/bot.py

# 保留最近 7 天的备份
find "${BACKUP_DIR}" -name "*.tar.gz" -mtime +7 -delete

echo "Backup completed: ${DATE}"

0 2 * * * /opt/backups/backup.sh >> /var/log/backup.log 2>&1

# 1. 检查服务状态
sudo systemctl status shlink-bot

# 2. 查看最近日志
sudo journalctl -u shlink-bot -n 50

# 3. 检查进程
ps aux | grep bot.py

# 4. 测试网络连接
curl https://api.telegram.org/bot<TOKEN>/getMe

# 验证 API Key
curl -X GET http://localhost:8080/rest/v3/short-urls \
  -H "X-Api-Key: your_api_key"

# 检查 Shlink 日志
docker compose -f /opt/shlink/docker-compose.yml logs --tail=50

# 重新生成 API Key
docker exec -it shlink shlink api-key:generate

# 更新 .env 文件
nano /opt/shlink_bot/.env

# 重启 Bot
sudo systemctl restart shlink-bot

# 查看详细错误
docker compose up --no-start
docker compose logs

# 检查端口占用
sudo netstat -tulpn | grep 8080

# 检查磁盘空间
df -h

services:
  redis:
    image: redis:7-alpine
    restart: always
    
  shlink:
    environment:
      - CACHE_NAMESPACE=shlink
      - REDIS_SERVERS=redis:6379
    depends_on:
      - redis

# 定期清理访问记录（保留 30 天）
docker exec shlink shlink visit:delete --since=30d

# 查看失败的 API 请求
docker compose logs shlink | grep "401\|403"

# 检查 Cloudflare 防火墙日志
# 访问 Cloudflare Dashboard → Security → Events

# 临时封禁 IP（在 Cloudflare WAF）
# Expression: ip.src eq 1.2.3.4
# Action: Block

# 1. 立即停止服务
sudo systemctl stop shlink-bot
docker compose -f /opt/shlink/docker-compose.yml down

# 2. 备份数据
cp -r /opt/shlink/data /opt/backups/emergency_backup_$(date +%s)

# 3. 检查日志
sudo journalctl -u shlink-bot --since "1 hour ago" > /tmp/incident.log

# 4. 更换 API Key
docker exec shlink shlink api-key:generate
# 吊销旧 Key
docker exec shlink shlink api-key:disable old_key_id

# 5. 重新启动
docker compose -f /opt/shlink/docker-compose.yml up -d
sudo systemctl start shlink-bot