How to detect SSTP traffic?

Question

I'm writing a program which should detect a VPN traffic. As far as I read about the subject, it seems the detection of the tunneling protocol is easy as firewall rules, using their dedicated ports:

PPTP: port 1723/TCP

OpenVPN: port 1194

L2TP: port 1701/UDP

My problem is with the SSTP, because it is using port 443, which is widely used.

So I have 2 questions:

1. Am I too naive to think I can detect those VPNs tunneling protocols only by their ports?
2. Does anyone has any idea how to detect SSTP and differ its traffic from any other type / app which uses TLS/SSL (even using DPI)?

I'm attaching piece of Python code which detects the communication in the above ports

import dpkt
import socket

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

# VPN PORTS
import common
import dal

protocols_strs = {"pp2e_gre": "1723/TCP PP2P_GRE_PORT",
                  "openvpn": "1194 OPENVPN_PORT",
                  "ike": "500/UDP IKE_PORT",
                  "l2tp_ipsec": "1701/UDP L2TP_IPSEC_PORT"
                  }

port_protocols = {1723: 'pp2e_gre',
                  1194: 'openvpn',
                  500: 'ike',
                  1701: 'l2tp_ipsec'
                  }

# Dict of sets holding the protocols sessions
protocol_sessions = {"pp2e_gre": [],
                     "openvpn": [],
                     "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_openvpn_ike_handler(five_tuple):
    # Get protocol
    protocol = five_tuple['protocol']

    # Clear old sessions in db
    dal.remove_old_sessions(five_tuple['timestamp'], 'vpn_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_vpn_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)


# -------------------------- VPN ports jump tables
tcp_vpn_ports = {1723: pp2e_gre_openvpn_ike_handler,
                 1194: pp2e_gre_openvpn_ike_handler}

udp_vpn_ports = {500: pp2e_gre_openvpn_ike_handler,
                 1701: l2tp_ipsec_handler,
                 1194: pp2e_gre_openvpn_ike_handler}


# -------------------------- Functions
def process_packet(timestamp, packet):
    """
    Given a packet process it for detecting a VPN 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 VPN ports in session
    detected_ports = set(tcp_vpn_ports).intersection(frame_ports)

    # If TCP VPN 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_vpn_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)

Answer 1

1. "Am I too naive to think I can detect those VPNs tunneling protocols only by their ports?:

"The official OpenVPN port number is 1194, but any port number between 1 and 65535 will work. If you don't provide the 'port' option, 1194 will be used.

So if your code is looking for 1194 traffic, as per the dictionary entries, you will capture default Open VPN flows only.

2. The SSTP message is encrypted with the SSL channel of the HTTPS protocol. So I don't see how you would identify this traffic as it is encrypted. (Source)