192.168.1.1:7777 => 1.2.3.4:10000 => 8.8.8.8:123

8.8.8.8:123 => 1.2.3.4:10000 => 192.168.1.1:7777

#!/usr/bin/python3
#coding=utf-8

import socket
import sys

def server(addr):
    print("[NAT CHECK launch as server on %s]" % str(addr))

    # listen UDP service
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(addr)

    # [1. check "Open Internet" and "Symmetric NAT"]
    # recevie client request and return export ip
    data, cconn = sock.recvfrom(1024)
    print("server get client info: %s" % str(cconn))
    data = "%s:%d" % (cconn[0], cconn[1])
    sock.sendto(data.encode("utf-8"), cconn)

    # receive assist data about client another export ip
    data, aconn = sock.recvfrom(1024)
    print("server get client info (from assist): %s" % data.decode("utf-8"))
    sock.sendto(data, cconn)

    # [2. check "Full-Cone NAT", "Restricted NAT" and "Restricted-Port NAT"]
    # recevie client request
    data, cconn = sock.recvfrom(1024)
    print("server get client info: %s" % str(cconn))
    # receive assist data about client another export ip
    data, aconn = sock.recvfrom(1024)
    print("server get client info (from assist): %s" % data.decode("utf-8"))

    # send data to client through (assist get) export ip
    print("send packet for testing Full-Cone NAT")
    array = data.decode("utf-8").split(":")
    caconn = (array[0], int(array[1]))
    sock.sendto("TEST FOR FULL-CONE NAT".encode("utf-8"), caconn)

    # send data to client through (server get) export ip and with different port
    sock.recvfrom(1024) # NEXT flag
    print("send packet for testing Restricted NAT")
    cdconn = (cconn[0], cconn[1] - 1)
    sock.sendto("TEST FOR Restricted NAT".encode("utf-8"), cdconn)

    # send data to client through (server get) export ip
    sock.recvfrom(1024) # NEXT flag
    print("send packet for testing Restricted-Port NAT")
    sock.sendto("TEST FOR Restricted-Port NAT".encode("utf-8"), cconn)
# server()

def assist(addr, serv):
    print("[NAT CHECK launch as assist on %s && server=%s]" %
                                                    (str(addr), str(serv)))

    # listen UDP service
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(addr)

    # [1. check "Open Internet" and "Symmetric NAT"]
    # recevie client request and forward to server
    data, conn = sock.recvfrom(1024)
    print("assist get client info: %s" % str(conn))
    data = "%s:%d" % (conn[0], conn[1])
    sock.sendto(data.encode("utf-8"), serv)

    # [2. check "Full-Cone NAT", "Restricted NAT" and "Restricted-Port NAT"]
    # recevie client request and forward to server
    data, conn = sock.recvfrom(1024)
    print("assist get client info: %s" % str(conn))
    data = "%s:%d" % (conn[0], conn[1])
    sock.sendto(data.encode("utf-8"), serv)
# assist()

def client(serv, ast):
    print("[NAT CHECK launch as client to server=%s && assist=%s]" %
                                                    (str(serv), str(ast)))

    # [1. check "Open Internet" and "Symmetric NAT"]
    print("send data to server and assist")
    # get local address
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.connect(serv)
    localaddr = sock.getsockname()

    # send data to server and assist with same socket
    # and register so that the server can obtain the export ip
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.sendto("register".encode("utf-8"), serv)
    sock.sendto("register".encode("utf-8"), ast)

    # receive export ip from server
    data, conn = sock.recvfrom(1024)
    exportaddr = data.decode("utf-8")
    print("get export ip: %s, localaddr: %s" % (exportaddr, str(localaddr)))

    # check it is "Open Internet"
    if exportaddr.split(":")[0] == localaddr[0]:
        print("[Open Internet]")
        return
    # end if

    # receive another export ip (assist) from server
    data, conn = sock.recvfrom(1024)
    anotheraddr = data.decode("utf-8")
    print("get export ip(assist): %s, export ip(server): %s" % (anotheraddr, exportaddr))

    # check it is "Symmetric NAT"
    if exportaddr != anotheraddr:
        print("[Symmetric NAT]")
        return
    # end if

    # [2. check "Full-Cone NAT", "Restricted NAT" and "Restricted-Port NAT"]
    # send data to server and assist with different socket
    # receive the data sent back by the server through the export ip(assist) mapping
    ssock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    ssock.sendto("register".encode("utf-8"), serv)
    asock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    asock.sendto("register".encode("utf-8"), ast)

    asock.settimeout(5)
    try:
        data, conn = asock.recvfrom(1024)
        print("[Full-Cone NAT]")
        return
    except:
        pass

    # receive the data sent back by the server with different port
    ssock.sendto("NEXT".encode("utf-8"), serv)
    ssock.settimeout(5)
    try:
        data, conn = ssock.recvfrom(1024)
        print("[Restricted NAT]")
        return
    except:
        pass

    # receive the data sent back by the server
    ssock.sendto("NEXT".encode("utf-8"), serv)
    ssock.settimeout(5)
    try:
        data, conn = ssock.recvfrom(1024)
        print("[Restricted-Port NAT]")
    except:
        print("[Unknown, something error]")
# client()

def usage():
    print("Usage:")
    print("  python3 nat_check.py server [ip:port]")
    print("  python3 nat_check.py assist [ip:port] [server]")
    print("  python3 nat_check.py client [server] [assist]")
# end usage()

if __name__ == "__main__":
    if len(sys.argv) < 3:
        usage()
        exit(0)
    # end if

    role = sys.argv[1]
    array = sys.argv[2].split(":")
    address1 = (array[0], int(array[1]))
    if role == "assist" or role == "client":
        if len(sys.argv) > 3:
            array = sys.argv[3].split(":")
            address2 = (array[0], int(array[1]))
        else:
            usage()
            exit(0)
    # end if

    # server/client launch
    if role == "server":
        server(address1)
    elif role == "assist":
        assist(address1, address2)
    elif role == "client":
        client(address1, address2)
    else:
        usage()
# end main()

#!/usr/bin/python3
#coding=utf-8

import socket

if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 
    sock.bind(("0.0.0.0", 1024))

    # 1.receive message and get one export ip:port (PC1)
    data, conn1 = sock.recvfrom(1024)
    addr1 = "%s:%d" % (conn1[0], conn1[1])
    print("1.get PC1 export ip:port = %s" % addr1)

    # 2.receive message and get another export ip:port (PC2)
    data, conn2 = sock.recvfrom(1024)
    addr2 = "%s:%d" % (conn2[0], conn2[1])
    print("2.get PC2 export ip:port = %s" % addr2)

    # 3.send export address of PC1 to PC2
    sock.sendto(addr1.encode("utf-8"), conn2)
    print("3.send export address of PC1(%s) to PC2(%s)" % (addr1, addr2))

    # 4.send export address of PC2 to PC1
    sock.sendto(addr2.encode("utf-8"), conn1)
    print("4.send export address of PC2(%s) to PC1(%s)" % (addr2, addr1))

    print("done")
    sock.close()
# end main()

#!/usr/bin/python3
#coding=utf-8

import random
import socket
import string
import time

if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 
    #serv = ("10.0.1.1", 1024)
    serv = ("192.168.50.55", 1024)
    print("server =>", serv)

    # 1/2.send message to server, server can get our export ip:port
    sock.sendto("REGISTER".encode("utf-8"), serv)
    print("1/2.send REGISTER message to server")

    # 3/4.receive the export address of the peer from the server
    data, conn = sock.recvfrom(1024)
    array = data.decode("utf-8").split(":")
    addr = (array[0], int(array[1]))
    print("3/4.receive the export address of the peer, %s" % str(addr))

    # 5/6.send KNOCK message to export address of peer
    wait = random.randint(2, 5)
    print("5/6.send KNOCK message to export address of peer (wait %d s)" % wait)
    # in order to stagger the two clients
    # so that the router can better create the mapping
    time.sleep(wait)
    sock.sendto("KNOCK".encode("utf-8"), addr)

    name = "".join(random.sample(string.ascii_letters, 8))
    print("my name is %s, start to communicate" % name)
    # 7.communicate each other
    count = 0
    while True:
        sock.settimeout(5)
        try:
            data, conn = sock.recvfrom(1024)
            print("%s => %s" % (str(conn), data.decode("utf-8")))
        except Exception as e:
            print(e)

        msg = "%s: %d" % (name, count)
        count += 1
        sock.sendto(msg.encode("utf-8"), conn)

        time.sleep(1)
    # end while()

    sock.close()
# end main()