前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >一个简单Python脚本,实现轻量应用服务器防火墙自动添加客户端公网IP访问

一个简单Python脚本,实现轻量应用服务器防火墙自动添加客户端公网IP访问

作者头像
邓鹏
修改2024-04-26 16:26:27
2380
修改2024-04-26 16:26:27

当需要设置轻量级服务器的防火墙策略时,一般用户都是需要手动去设置开放指定公网IP访问轻量服务器,这样也是比较麻烦的。就此作者想了下,如何避免每次手动去修改公网IP,这样也是非常麻烦,如果腾讯云轻量服务器控制台能增加一个安全选项,实现浏览器获取当前用户公网IP并一键设置允许常用端口访问。这样大大的增加用户安全体验下。

如下实现了自动获取当前客户端网络环境的公网IP并允许当前公网IP访问轻量服务器指定端口:

用户首先需要获取腾讯云的访问密钥(SecretId和Key)。然后,用户可以将这些密钥代入Python脚本,并指定服务器所在的区域以及云服务器的ID信息。

这个Python脚本的作用是解决用户远程访问腾讯云服务器的安全问题。运行此Python脚本后,它会自动获取当前用户的公网IP地址,并设置腾讯云服务器只允许当前公网IP地址访问服务器的22和8888端口,从而确保服务器的22和8888端口不对所有人开放。此脚本对于移动办公和运维人员来说非常方便。

运行Python脚本后,默认情况下,回车键将会自动开启所有用户对80和443端口的访问权限,并且只允许当前公网IP访问服务器的22端口和8888端口。

代码语言:txt
复制
import json
from tabulate import tabulate
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.lighthouse.v20200324 import lighthouse_client, models
import requests
import time

#腾讯云SecretId信息和秘钥
SecretId = "AKIDqilQ1DSjsy7BHcHqHZjtESf"
key = "PdibItIYlrsqmibKxW3"
#轻量服务器地域
TencentZone = "guangzhou"
#轻量服务器ID,如下示例
Serverid = "lhins-6y22h1kaip"
#为了防止误运行,设置了密码,底部可修改。默认 admin 密码 123456

cred = credential.Credential(SecretId, key)
def add_local(ip,open_port="22",status="ACCEPT"):
    try:
        httpProfile = HttpProfile()
        httpProfile.endpoint = "lighthouse.tencentcloudapi.com"

        # 实例化一个client选项,可选的,没有特殊需求可以跳过
        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象,clientProfile是可选的
        client = lighthouse_client.LighthouseClient(cred, TencentZone, clientProfile)

        # 实例化一个请求对象,每个接口都会对应一个request对象
        req = models.ModifyFirewallRulesRequest()
        params = {
            "InstanceId": Serverid,
            "FirewallRules": [
                {
                    "Protocol": "TCP",
                    "Port": open_port,
                    "CidrBlock": ip,
                    "Action": status,
                    "FirewallRuleDescription": "服务器远程访问"
                }
            ]
        }
        req.from_json_string(json.dumps(params))

        # 返回的resp是一个ModifyFirewallRulesResponse的实例,与请求对象对应
        resp = client.ModifyFirewallRules(req)
        # 输出json格式的字符串回包
        print(resp.to_json_string())
    except TencentCloudSDKException as err:
        print(err)
def add_https(ip="0.0.0.0/0",port ="443,80"):
    try:
        # 实例化一个http选项,可选的,没有特殊需求可以跳过
        httpProfile = HttpProfile()
        httpProfile.endpoint = "lighthouse.tencentcloudapi.com"

        # 实例化一个client选项,可选的,没有特殊需求可以跳过
        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象,clientProfile是可选的
        client = lighthouse_client.LighthouseClient(cred, TencentZone, clientProfile)

        # 实例化一个请求对象,每个接口都会对应一个request对象
        req = models.CreateFirewallRulesRequest()
        params = {
            "InstanceId": Serverid,
            "FirewallRules": [
                {
                    "Protocol": "TCP",
                    "Port": port,
                    "CidrBlock": ip,
                    "Action": "ACCEPT",
                    "FirewallRuleDescription": "网站访问"
                }
            ]
        }
        req.from_json_string(json.dumps(params))

        # 返回的resp是一个CreateFirewallRulesResponse的实例,与请求对象对应
        resp = client.CreateFirewallRules(req)
        # 输出json格式的字符串回包
        print(resp.to_json_string())

    except TencentCloudSDKException as err:
        print(err)

def add_new(ip,new_port):
    try:
        # 实例化一个http选项,可选的,没有特殊需求可以跳过
        httpProfile = HttpProfile()
        httpProfile.endpoint = "lighthouse.tencentcloudapi.com"

        # 实例化一个client选项,可选的,没有特殊需求可以跳过
        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象,clientProfile是可选的
        client = lighthouse_client.LighthouseClient(cred, TencentZone, clientProfile)

        # 实例化一个请求对象,每个接口都会对应一个request对象
        req = models.CreateFirewallRulesRequest()
        params = {
            "InstanceId": Serverid,
            "FirewallRules": [
                {
                    "Protocol": "TCP",
                    "Port": new_port,
                    "CidrBlock": ip,
                    "Action": "ACCEPT"
                }
            ]
        }
        req.from_json_string(json.dumps(params))

        # 返回的resp是一个CreateFirewallRulesResponse的实例,与请求对象对应
        resp = client.CreateFirewallRules(req)
        # 输出json格式的字符串回包
        print(resp.to_json_string())

    except TencentCloudSDKException as err:
        print(err)
def add_list():
    try:
        # 实例化一个http选项,可选的,没有特殊需求可以跳过
        httpProfile = HttpProfile()
        httpProfile.endpoint = "lighthouse.tencentcloudapi.com"

        # 实例化一个client选项,可选的,没有特殊需求可以跳过
        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象,clientProfile是可选的
        client = lighthouse_client.LighthouseClient(cred, TencentZone, clientProfile)

        # 实例化一个请求对象,每个接口都会对应一个request对象
        req = models.DescribeFirewallRulesRequest()
        params = {
            "InstanceId": Serverid
        }
        req.from_json_string(json.dumps(params))

        # 返回的resp是一个DescribeFirewallRulesResponse的实例,与请求对象对应
        resp = client.DescribeFirewallRules(req)
        # # 输出json格式的字符串回包
        # print(resp.to_json_string())
        # 将对象转换为JSON字符串
        # 将对象转换为JSON字符串
        resp_str = resp.to_json_string()

        # 解析JSON字符串为字典
        resp_dict = json.loads(resp_str)

        # 只取需要的字段
        selected_fields = [{"CidrBlock": rule["CidrBlock"], "Protocol": rule["Protocol"], "Port": rule.get("Port"),
                            "Action": rule["Action"]} for rule in resp_dict["FirewallRuleSet"]]
        # 格式化输出为表格
        table = tabulate(selected_fields, headers="keys", tablefmt="orgtbl")
        # 输出结果
        print(table)

    except TencentCloudSDKException as err:
        print(err)
def local_ip():
    url = "https://www.ipuu.net/ipuu/user/getIP"
    response = requests.get(url)
    data = response.json()
    return data["data"]
def caidan():
    print("""---------------------------------------
         欢迎进入网站维护程序
---------------------------------------
    1 允许管理登录测试网站服务器SSH
    2 服务器开放端口
    3 显示服务器端口
    4 退出程序
---------------------------------------""")
print(f"本地公网IP:{local_ip()}")
def run_start():
    while True:
        caidan()
        try:
            num = int(input("请输入你选择的功能序号:"))
            if  num == 1 :
                print(f"本地公网IP:{local_ip()}")
                ip = input("请输入你要添加访问的IP地址,回车默认本地公网IP地址:") or local_ip()
                open_port = input("请输入要开放的TCP端口,默认22:") or "22"
                add_local(ip,open_port)
                add_https()
                add_https(ip=ip,port="8888")
                # add_https(ip=ip, port="80")

            elif num == 2:
                print(f"本地公网IP:{local_ip()}")
                ip = input("请输入你要添加访问的IP地址,回车默认本地公网IP地址:") or local_ip()
                open_port = input("请输入要开放的TCP端口:")
                add_new(ip, open_port)
            elif num == 3:
                add_list()

            elif num == 4:
                exit()
            else:
                print("功能选项不存在,退出程序,若需再次调式,请重新运行")
                exit()
        except ValueError:
            print("输入错误,请重新运行程序")
def user_log(user, passwd):
    if user == "admin" and passwd == "123456":
        print("登录成功")
        return True
    else:
        print("登录失败,请重新登录")
        return False
def login():
    print("运行本程序前,需要进行登录")
    while True:
        user = input("请输入登录用户名:")
        passwd = input("请输入登录密码:")
        if user_log(user, passwd):
            run_start()
            break
if __name__ == '__main__':
    login()

交互式运行显示如下:

本文系转载,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文系转载前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
云服务器
云服务器(Cloud Virtual Machine,CVM)提供安全可靠的弹性计算服务。 您可以实时扩展或缩减计算资源,适应变化的业务需求,并只需按实际使用的资源计费。使用 CVM 可以极大降低您的软硬件采购成本,简化 IT 运维工作。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档