Tenho uma classe de servidor de socket TCP que roda em CPython e MicroPython . Ela roda sem problemas. Gostaria apenas de estender sua funcionalidade com um filtro de IP. De uma versão anterior, que ainda era um servidor http, tenho um padrão de expressão regular para corresponder ao endereço IP da solicitação. Mas onde posso intervir e inspecionar o endereço remoto para aceitar ou recusar a conexão?
Observação: o objeto leitor em newConnection não tem o atributo '_transport' no 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")