I have a Tcp socket server class that is to run in CPython and MicroPython. It does, no problem there. Only I would like to extend its functionality with an IP filter. From an earlier version that still was a http server, I have a regular expression pattern to match the requesting IP-address. But where can I step in and inspect the remote address to accept or refuse the connection?
Note: the reader object in newConnection has no attribute '_transport' in microPython.
def log(*args, prio = 0):
if prio >= 0:
print(*args)
class TcpServer(SocketBase):
def __init__(self, port=8888, requestHandler = None, ipMask = r"192.168.(\d+).(\d+)|localhost|127.0.0.1"):
'''
:param requestHandler: callable with parameters request, response
'''
super().__init__(host="0.0.0.0", port = port)
self.requestHandler = requestHandler
self.ipRange = re.compile(ipMask)
self.timeout = TimeOut()
async def start(self):
self.server = await asyncio.start_server(self.newConnection, host=self.host, port=self.port)
log("server started at ", (self.host, self.port), prio=1)
while True:
await asyncio.sleep(10)
async def newConnection(self, reader, writer):
log("Connected")
self.connection = (reader, writer)
self.timeout.extend(30)
while not self.timeout.expired():
if not await self.pollOpenConnection(reader, writer):
# not data, then doze a while:
await asyncio.sleep(1./25.)
writer.close()
log("Disconnected")
In your connection handler you can find it from the stream writer:
remote_addr = writer.get_extra_info('peername')[0]