File:  [ELWIX - Embedded LightWeight unIX -] / embedaddon / miniupnpc / testupnpigd.py
Revision 1.1.1.2 (vendor branch): download - view: text, annotated - select for diffs - revision graph
Wed Sep 27 11:21:37 2023 UTC (7 months, 3 weeks ago) by misho
Branches: miniupnpc, elwix, MAIN
CVS tags: v2_2_5p0, HEAD
Version 2.2.5p0

#! /usr/bin/env python
# $Id: testupnpigd.py,v 1.1.1.2 2023/09/27 11:21:37 misho Exp $
# MiniUPnP project
# Author : Thomas Bernard
# This Sample code is public domain.
# website : https://miniupnp.tuxfamily.org/

# import the python miniupnpc module
import miniupnpc
import socket

try:
    from http.server import BaseHTTPRequestHandler, HTTPServer
except ImportError:
    from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer

# function definition
def list_redirections():
	i = 0
	while True:
		p = u.getgenericportmapping(i)
		if p==None:
			break
		print(i, p)
		i = i + 1

#define the handler class for HTTP connections
class handler_class(BaseHTTPRequestHandler):
	def do_GET(self):
		self.send_response(200)
		self.end_headers()
		self.wfile.write(b"OK MON GARS")

# create the object
u = miniupnpc.UPnP()
#print 'inital(default) values :'
#print ' discoverdelay', u.discoverdelay
#print ' lanaddr', u.lanaddr
#print ' multicastif', u.multicastif
#print ' minissdpdsocket', u.minissdpdsocket
u.discoverdelay = 200;

try:
	print('Discovering... delay=%ums' % u.discoverdelay)
	ndevices = u.discover()
	print(ndevices, 'device(s) detected')

	# select an igd
	u.selectigd()
	# display information about the IGD and the internet connection
	print('local ip address :', u.lanaddr)
	externalipaddress = u.externalipaddress()
	print('external ip address :', externalipaddress)
	print(u.statusinfo(), u.connectiontype())

	#instanciate a HTTPd object. The port is assigned by the system.
	httpd = HTTPServer((u.lanaddr, 0), handler_class)
	eport = httpd.server_port

	# find a free port for the redirection
	r = u.getspecificportmapping(eport, 'TCP')
	while r != None and eport < 65536:
		eport = eport + 1
		r = u.getspecificportmapping(eport, 'TCP')

	print('trying to redirect %s port %u TCP => %s port %u TCP' % (externalipaddress, eport, u.lanaddr, httpd.server_port))

	b = u.addportmapping(eport, 'TCP', u.lanaddr, httpd.server_port,
	                    'UPnP IGD Tester port %u' % eport, '')
	if b:
		print('Success. Now waiting for some HTTP request on http://%s:%u' % (externalipaddress ,eport))
		try:
			httpd.handle_request()
			httpd.server_close()
		except KeyboardInterrupt as details:
			print("CTRL-C exception!", details)
		b = u.deleteportmapping(eport, 'TCP')
		if b:
			print('Successfully deleted port mapping')
		else:
			print('Failed to remove port mapping')
	else:
		print('Failed')

	httpd.server_close()

except Exception as e:
	print('Exception :', e)

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>