windows已建立威胁IP排查

在应急响应的时候，需要筛选出服务器建立连接的进程、PID，此代码可满足该需求实现共计2步

第一步获取服务器建立连接的ip

import re


# 从文件读取 netstat 输出
def read_netstat_file(file_path):
    try:
        with open(file_path, 'r') as file:
            return file.read()
    except Exception as e:
        print(f"Error reading file: {e}")
        return None


# 提取已建立连接和未建立连接的外部 IP 地址
def extract_ips_by_state(netstat_output):
    established_ips = set()  # 存储已建立连接的外部 IP 地址
    listening_ips = set()  # 存储未建立连接的外部 IP 地址

    # 正则表达式匹配 ESTABLISHED 状态的连接，提取外部 IP 地址（不包括端口）
    established_pattern = re.compile(r'^\s*TCP\s+\S+\s+(\S+)\s+ESTABLISHED', re.MULTILINE)
    # 正则表达式匹配 LISTENING 状态的连接，提取外部 IP 地址（不包括端口）
    listening_pattern = re.compile(r'^\s*TCP\s+\S+\s+(\S+)\s+LISTENING', re.MULTILINE)

    # 查找已建立连接的外部 IP 地址（不包含端口号）
    for match in established_pattern.findall(netstat_output):
        established_ips.add(match.split(':')[0])  # 去掉端口号部分

    # 查找未建立连接的外部 IP 地址（不包含端口号）
    for match in listening_pattern.findall(netstat_output):
        listening_ips.add(match.split(':')[0])  # 去掉端口号部分

    return established_ips, listening_ips


# 将 IP 地址写入文件
def write_ips_to_file(established_ips, listening_ips, output_file):
    try:
        with open(output_file, 'w') as file:
            file.write("已建立连接的外部 IP 地址:\n")
            for ip in established_ips:
                file.write(f"{ip}\n")

            file.write("\n未建立连接的外部 IP 地址:\n")
            for ip in listening_ips:
                file.write(f"{ip}\n")

        print(f"IP 地址已成功写入 {output_file}")
    except Exception as e:
        print(f"Error writing to file: {e}")


def main():
    input_file = 'all.txt'  # 你的 netstat 输出文件路径
    output_file = 'ips.txt'  # 要写入的输出文件路径
    netstat_output = read_netstat_file(input_file)

    if netstat_output:
        established_ips, listening_ips = extract_ips_by_state(netstat_output)
        write_ips_to_file(established_ips, listening_ips, output_file)


if __name__ == '__main__':
    main()

第二部查看服务器ip地址

import requests
import time


# 获取 IP 地理位置信息的函数
def get_ip_info(ip):
    url = f'https://qifu-api.baidubce.com/ip/geo/v1/district?ip={ip}'

    # 设置请求头（可以根据需要修改）
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Content-Type': 'application/json',
    }

    try:
        response = requests.get(url, headers=headers)  # 将请求头传入
        response.raise_for_status()  # 检查请求是否成功
        data = response.json()

        if data["code"] == "Success":
            info = data["data"]
            prov = info.get("prov", "未知")
            city = info.get("city", "未知")
            return ip, prov, city
        else:
            return ip, "未知", "未知"
    except requests.exceptions.RequestException as e:
        print(f"请求错误: {e}")
        return ip, "错误", "错误"
    except ValueError as e:
        print(f"解析错误: {e}")
        return ip, "错误", "错误"
# 批量查询 IP 地址信息
def batch_query_ips(file_path):
    with open(file_path, 'r') as file:
        ips = file.read().splitlines()

    for ip in ips:
        ip, prov, city = get_ip_info(ip)  # 解包为三个值
        print(f"IP: {ip}, Province: {prov}, City: {city}")
        time.sleep(1)  # 每次请求后延迟 5 秒


# 替换 'ips.txt' 为你的文件路径
batch_query_ips('ips.txt')