如何检测 SSTP 流量？

Posted 2023-02-22

【中文标题】如何检测 SSTP 流量？

【英文标题】：How to detect SSTP traffic?

【发布时间】：2017-05-22 21:05:14

【问题描述】：

我正在编写一个应该检测 *** 流量的程序。 就我所读到的主题而言，隧道协议的检测似乎很容易，就像防火墙规则一样，使用它们的专用端口：

PPTP：端口 1723/TCP

Open***：端口 1194

L2TP：端口 1701/UDP

我的问题在于 SSTP，因为它使用的是广泛使用的端口 443。

所以我有两个问题：

我是不是太天真了，以为我只能通过它们的端口来检测那些 *** 隧道协议？ 是否有人知道如何检测 SSTP 并将其流量与使用 TLS/SSL（甚至使用 DPI）的任何其他类型/应用程序区分开来？

我正在附加一段 Python 代码，用于检测上述端口中的通信

import dpkt
import socket

# -------------------------- Globals

# *** PORTS
import common
import dal

protocols_strs = "pp2e_gre": "1723/TCP PP2P_GRE_PORT",
                  "open***": "1194 OPEN***_PORT",
                  "ike": "500/UDP IKE_PORT",
                  "l2tp_ipsec": "1701/UDP L2TP_IPSEC_PORT"
                  

port_protocols = 1723: 'pp2e_gre',
                  1194: 'open***',
                  500: 'ike',
                  1701: 'l2tp_ipsec'
                  

# Dict of sets holding the protocols sessions
protocol_sessions = "pp2e_gre": [],
                     "open***": [],
                     "ike": [],
                     "l2tp_ipsec": []


# -------------------------- Functions
def is_bidirectional(five_tuple, protocol):
    """
    Given a tuple and protocol check if the connection is bidirectional in the protocol
    :param five_tuple:
    :return: True of the connection is bidirectional False otherwise
    """
    src_ip = five_tuple['src_ip']
    dest_ip = five_tuple['dest_ip']

    # Filter the sessions the five tuple's ips spoke in
    ike_sessions = filter(lambda session: (session['src_ip'] == src_ip and session['dest_ip'] == dest_ip)
                                          or
                                          (session['dest_ip'] == src_ip and session['src_ip'] == dest_ip),
                          protocol_sessions[protocol])
    # Return true if 2 session (1 for each direction) were found
    return len(ike_sessions) == 2


def print_alert(timestamp, protocol, five_tuple):
    """
    Print alert description to std
    :param timestamp:
    :param protocol:
    :param five_tuple:
    :return:
    """
    print timestamp, ":\t detected port %s communication (%s:%s ---> %s:%s)" % \
                     (protocol, five_tuple['src_ip'], five_tuple['src_port'], five_tuple['dest_ip'],
                      five_tuple['dest_port'])


def pp2e_gre_open***_ike_handler(five_tuple):
    # Get protocol
    protocol = five_tuple['protocol']

    # Clear old sessions in db
    dal.remove_old_sessions(five_tuple['timestamp'], '***_sessions')

    # Clear old sessions in cache
    protocol_sessions[protocol] = common.clear_old_sessions(five_tuple, protocol_sessions[protocol])

    # If session already exists - return
    if common.check_if_session_exists(five_tuple, protocol_sessions[protocol]):
        session_to_update = common.get_session(five_tuple, protocol_sessions[protocol])
        session_to_update['timestamp'] = five_tuple['timestamp']
        return

    # Update DB
    dal.upsert_***_session(five_tuple)

    # Add to cache
    protocol_sessions[protocol].append(five_tuple)

    # Print alert
    print_alert(five_tuple['timestamp'], protocols_strs[protocol], five_tuple)


def l2tp_ipsec_handler(five_tuple):
    if five_tuple in protocol_sessions['l2tp_ipsec']:
        return

    # If bi-directional IKE protocol performed earlier - alert
    if not is_bidirectional(five_tuple, 'ike'):
        return

    protocol_sessions['l2tp_ipsec'].append(five_tuple)
    print_alert(five_tuple['timestamp'], protocols_strs['l2tp_ipsec'], five_tuple)


# -------------------------- *** ports jump tables
tcp_***_ports = 1723: pp2e_gre_open***_ike_handler,
                 1194: pp2e_gre_open***_ike_handler

udp_***_ports = 500: pp2e_gre_open***_ike_handler,
                 1701: l2tp_ipsec_handler,
                 1194: pp2e_gre_open***_ike_handler


# -------------------------- Functions
def process_packet(timestamp, packet):
    """
    Given a packet process it for detecting a *** communication

    :param packet:
    :param timestamp:
    :return:
    """
    # Parse the input
    eth_frame = dpkt.ethernet.Ethernet(packet)

    # Check if IP
    if eth_frame.type != dpkt.ethernet.ETH_TYPE_IP:
        return

    # If not IP return
    ip_frame = eth_frame.data

    # if TCP or UDP
    if ip_frame.p not in (dpkt.ip.IP_PROTO_TCP, dpkt.ip.IP_PROTO_UDP):
        return

    # Extract L3 frame
    frame = ip_frame.data

    # Extract ports
    frame_ports = (frame.sport, frame.dport)

    # get *** ports in session
    detected_ports = set(tcp_***_ports).intersection(frame_ports)

    # If TCP *** port was not detected return
    if not detected_ports:
        return

    # Get detected port
    port = detected_ports.pop()

    # Translate port to str
    protocol_str = port_protocols[port]

    # Choose handler by port
    handler = tcp_***_ports[port]

    # Extract 5-tuple parameters from frames
    five_tuple = 'src_ip': socket.inet_ntoa(ip_frame.src),
                  'dest_ip': socket.inet_ntoa(ip_frame.dst),
                  'src_port': frame.sport,
                  'dest_port': frame.dport,
                  'protocol': protocol_str,
                  'timestamp': timestamp

    # Invoke the chosen handler
    handler(five_tuple)

【参考方案1】：

“我是不是太天真了，以为只能通过端口检测到那些 *** 隧道协议？：

“officialOpen*** 端口号是 1194，但 1 到 65535 之间的任何端口号都可以使用。如果您不提供“端口”选项，将使用 1194。

因此，如果您的代码正在查找 1194 流量，则根据字典条目，您将仅捕获默认的 Open *** 流量。

SSTP 消息使用 HTTPS 协议的 SSL 通道加密。所以我看不出你会如何识别这个流量，因为它是加密的。 (Source)