mirror of
https://gitlab.com/fabinfra/fabaccess/fabfire_adapter.git
synced 2025-03-12 23:01:44 +01:00
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
import asyncio
|
|
import socket
|
|
import ssl
|
|
|
|
import capnp
|
|
connection_capnp = capnp.load('schema/connection.capnp')
|
|
authenticationsystem_capnp = capnp.load('schema/authenticationsystem.capnp')
|
|
|
|
async def myreader(client, reader):
|
|
while True:
|
|
data = await reader.read(4096)
|
|
client.write(data)
|
|
|
|
|
|
async def mywriter(client, writer):
|
|
while True:
|
|
data = await client.read(4096)
|
|
writer.write(data.tobytes())
|
|
await writer.drain()
|
|
|
|
async def connect(host, port, user, pw):
|
|
# Setup SSL context
|
|
ctx = ssl.create_default_context(
|
|
ssl.Purpose.SERVER_AUTH
|
|
)
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
# Handle both IPv4 and IPv6 cases
|
|
try:
|
|
reader, writer = await asyncio.open_connection(
|
|
host, port, ssl=ctx, family=socket.AF_INET6
|
|
)
|
|
except Exception:
|
|
reader, writer = await asyncio.open_connection(
|
|
host, port, ssl=ctx, family=socket.AF_INET
|
|
)
|
|
|
|
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
|
|
client = capnp.TwoPartyClient()
|
|
cap = client.bootstrap().cast_as(connection_capnp.Bootstrap)
|
|
|
|
# Assemble reader and writer tasks, run in the background
|
|
coroutines = [myreader(client, reader), mywriter(client, writer)]
|
|
asyncio.gather(*coroutines, return_exceptions=True)
|
|
|
|
auth = cap.authenticationSystem().authenticationSystem
|
|
req = auth.start_request()
|
|
req.request.mechanism = "PLAIN"
|
|
req.request.initialResponse.initial = "\0" + user + "\0" + pw
|
|
rep_prom = req.send()
|
|
response = await rep_prom.a_wait()
|
|
if response.response.outcome.result == "successful":
|
|
return cap
|
|
else:
|
|
print("Auth failed")
|
|
return None
|