FRP内网穿透如何避免SSH暴力破解（二）—

背景

上篇文章说到，出现了试图反复通过FRP的隧道，建立外网端口到内网服务器TCP链路的机器人，同时试图暴力破解ssh。这些连接造成了流量的浪费和不必要的通信开销。考虑到服务器使用者主要分布在A、B、C地区和国家，我打算对上一篇文章获取的建立连接的ip再过滤一遍，把其他地区的ip加以封禁，确保服务器不被恶意访问骚扰。

思路

在FRP服务端写一个python程序，每个小时查询一次已连接ip的清单，只允许指定区域的ip访问内网的指定端口。本文国家/地区列表我设置为中国大陆、香港、新加坡、马来西亚、美国。

获取地区&检查是否是目标区域

# 设置允许访问地区列表
target_countries = ('China', 'Hong Kong', 'Singapore', 'Malaysia', 'United States')

def get_ip_location(ip_address):
    response = requests.get(f'https://ipapi.co/{ip_address}/json/').json()
    country_name = response.get("country_name")
    # print(country_name)
    return country_name

def check_ip_location(ip_address):
    country_name = get_ip_location(ip_address)
    if country_name in target_countries:
        # print('ok')
        return 'ok'
    else:
        return 'no'

封禁服务区域外的ip

def ban_ip(ip_address):
    # 检查IP是否已经被封禁
    if is_ip_banned(ip_address):
        print(f"IP {ip_address} is already banned.")
        return
    
    try:
        # 封禁IP地址
        subprocess.run(['sudo', 'iptables', '-A', 'INPUT', '-s', ip_address, '-j', 'DROP'], check=True)
        # 记录到文件
        with open('/home/user/ban_ip_no_cn.txt', 'a') as file:
            ban_time = datetime.datetime.now()
            unban_time = ban_time + datetime.timedelta(days=1)
            file.write(f"{ban_time}, {ip_address}, {unban_time}\n")
        print(f"IP {ip_address} has been banned.")
    except Exception as e:
        print(f"Error banning IP {ip_address}: {e}")

封禁时间限制

每天运行一次脚本ban_ip_no_cn.sh，检查是否到了解封时间。

BAN_FILE="/home/user/ban_ip_no_cn.txt"
TEMP_FILE="/tmp/temp_ban_ip_no_cn.txt"

while IFS=, read -r ban_time ip_address unban_time; do
    current_time=$(date +%Y-%m-%d' '%H:%M:%S)
    if [[ "$current_time" > "$unban_time" ]]; then
        sudo iptables -D INPUT -s $ip_address -j DROP
    else
        echo "$line" >> $TEMP_FILE
    fi
done < $BAN_FILE

mv $TEMP_FILE $BAN_FILE

定期清理连接建立记录(ip_ban_delete_old.py)

# 已建立连接ip的清单‘establishment_ip.txt’，每三天释放一次

from datetime import datetime, timedelta
import os

def delete_old_entries(file_path):
    cutoff_date = datetime.now().date() - timedelta(days=1)

    temp_file_path = file_path + ".tmp"

    with open(file_path, 'r') as read_file, open(temp_file_path, 'w') as write_file:
        for line in read_file:
            line_date_str = line.split(' ')[0]  # Extract only the date part
            line_date = datetime.strptime(line_date_str, '%Y-%m-%d').date()

            if line_date >= cutoff_date:
                write_file.write(line)

    os.replace(temp_file_path, file_path)


# Path to the establishment_ip.txt file
file_path = '/home/peter/establishment_ip.txt'
delete_old_entries(file_path)

注意

establishment_ip.txt文件的格式如下，通过ss -anp | grep ":port"（port切换为你的frps开放的port）命令获取。

2024-02-06 07:36:52.541687: Established connection from IP 203.145.18.60 on port 23
2024-02-06 07:36:52.578422: Established connection from IP 203.145.18.60 on port 23
2024-02-06 07:40:01.597133: Established connection from IP 56.101.207.179 on port 24
2024-02-06 07:40:01.597341: Established connection from IP 203.145.18.60 on port 24
2024-02-06 07:40:01.633414: Established connection from IP 203.145.18.60 on port 24
2024-02-06 07:40:36.380221: Established connection from IP 203.145.18.60 on port 24

效果：

ip_ban_no_cn.log输出打印
ban_ip_no_cn.txt的被封禁ip记录

完整Python代码ip_ban_no_cn.py （注意修改路径）

# 每小时运行一次，从已建立连接ip的清单查询，封禁所有不欢迎ip
# 已建立连接ip的清单‘establishment_ip.txt’，每三天释放一次

import re
import requests
import subprocess
import datetime

# 更新国家列表
target_countries = ('China', 'Hong Kong', 'Singapore', 'Malaysia', 'United States')

def get_ip_location(ip_address):
    response = requests.get(f'https://ipapi.co/{ip_address}/json/').json()
    country_name = response.get("country_name")
    # print(country_name)
    return country_name

def check_ip_location(ip_address):
    country_name = get_ip_location(ip_address)
    if country_name in target_countries:
        # print('ok')
        return 'ok'
    else:
        return 'no'

def is_ip_banned(ip_address):
    try:
        with open('/home/{user}/ban_ip_no_cn.txt', 'r') as file:
            for line in file:
                if ip_address in line:
                    return True
    except FileNotFoundError:
        # 如果文件不存在，意味着没有IP被封禁
        return False
    return False

def ban_ip(ip_address):
    # 检查IP是否已经被封禁
    if is_ip_banned(ip_address):
        print(f"IP {ip_address} is already banned.")
        return
    
    try:
        # 封禁IP地址
        subprocess.run(['sudo', 'iptables', '-A', 'INPUT', '-s', ip_address, '-j', 'DROP'], check=True)
        # 记录到文件
        with open('/home/{user}/ban_ip_no_cn.txt', 'a') as file:
            ban_time = datetime.datetime.now()
            unban_time = ban_time + datetime.timedelta(days=1)
            file.write(f"{ban_time}, {ip_address}, {unban_time}\n")
        print(f"IP {ip_address} has been banned.")
    except Exception as e:
        print(f"Error banning IP {ip_address}: {e}")

def main():
    log_file_path = '/home/{user}/establishment_ip.txt'
    ip_pattern = re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b')
    
    current_time = datetime.datetime.now()
    
    try:
        with open(log_file_path, 'r') as file:
            for line in file:
                # 尝试解析每行的时间戳
                parts = line.split(": Established connection from IP ")
                if len(parts) > 1:
                    timestamp_str = parts[0].strip()
                    # print(timestamp_str)
                    try:
                        timestamp = datetime.datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S.%f')
                        # 检查时间是否在最近2小时内
                        # print(timestamp)
                        if (current_time - timestamp) <= datetime.timedelta(hours=2):
                            search_result = ip_pattern.search(line)
                            # print((current_time - timestamp))
                            if search_result:
                                ip_address = search_result.group(0)
                                if check_ip_location(ip_address) == 'no':
                                    ban_ip(ip_address)
                    except ValueError:
                        # 如果时间戳格式不正确，跳过这一行
                        continue
    except FileNotFoundError:
        print(f"File {log_file_path} not found.")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == '__main__':
    main()

Crontab运行的脚本

# Run ban_ip_no_cn.sh every 24 hours and log output
0 0 * * * /home/user/ban_ip_no_cn.sh > /home/user/log_temp/ban_ip_no_cn.log 2>&1

# Run ip_ban_no_cn.py every hour and log output
0 * * * * python3 /home/user/ip_ban_no_cn.py > /home/user/log_temp/ip_ban_no_cn.log 2>&1

# Run ip_ban_delete_old.py every 24 hours and log output
0 0 * * * python3 /home/user/ip_ban_delete_old.py > /home/user/log_temp/ip_ban_delete_old.log 2>&1

总结

代码写完才发现，早就有大神写了个复杂版本。呜呼哀哉，就好像论文idea被抢发了一样：https://github.com/zngw/frptables
ip归属地查询返回的是JSON格式，不光能查国家，还能获取到城市、语言、首都等信息。

get_location() function
As per the API documentation of ipapi, we need to make a GET request on https://ipapi.co/{ip}/{format}/ to get location information for a particular IP address. {ip} is replaced by the IP address and {format} can be replaced with any of these – json, jsonp, xml, csv, yaml.

This function internally calls the get_ip() function to get the IP address and then makes a GET request on the URL with the IP address. This API returns a JSON response that looks like this:

{
    "ip": "117.214.109.137",
    "version": "IPv4",
    "city": "Gaya",
    "region": "Bihar",
    "region_code": "BR",
    "country": "IN",
    "country_name": "India",
    "country_code": "IN",
    "country_code_iso3": "IND",
    "country_capital": "New Delhi",
    "country_tld": ".in",
    "continent_code": "AS",
    "in_eu": false,
    "postal": "823002",
    "latitude": 24.7935,
    "longitude": 85.012,
    "timezone": "Asia/Kolkata",
    "utc_offset": "+0530",
    "country_calling_code": "+91",
    "currency": "INR",
    "currency_name": "Rupee",
    "languages": "en-IN,hi,bn,te,mr,ta,ur,gu,kn,ml,or,pa,as,bh,sat,ks,ne,sd,kok,doi,mni,sit,sa,fr,lus,inc",
    "country_area": 3287590,
    "country_population": 1352617328,
    "asn": "AS9829",
    "org": "National Internet Backbone"
}