lvkm/pyrad3
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

safe progress

Browse Source
This commit is contained in:
Istvan Ruzman
2020-08-06 18:04:24 +02:00
parent 3254bc66e0
commit fd16436c3c
53 changed files with 2167 additions and 4589 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
example/server_async.py
View File

@@ -12,57 +12,67 @@ from pyrad.server import RemoteHost
try:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except:
pass
logging.basicConfig(level='DEBUG',
format='%(asctime)s [%(levelname)-8s] %(message)s')
logging.basicConfig(
level="DEBUG", format="%(asctime)s [%(levelname)-8s] %(message)s"
)
def print_attributes(packet):
print('Attributes returned by server:')
print("Attributes returned by server:")
for key, value in packet.items():
print(f'{key}: {value}')
print(f"{key}: {value}")
class FakeServer(ServerAsync):
def __init__(self, loop, dictionary):
ServerAsync.__init__(self, loop=loop, dictionary=dictionary,
enable_pkt_verify=True, debug=True)
ServerAsync.__init__(
self,
loop=loop,
dictionary=dictionary,
enable_pkt_verify=True,
debug=True,
)
def handle_auth_packet(self, protocol, packet, addr):
print('Received an authentication request with id ', packet.id)
print('Authenticator ', packet.authenticator.hex())
print('Secret ', packet.secret)
print("Received an authentication request with id ", packet.id)
print("Authenticator ", packet.authenticator.hex())
print("Secret ", packet.secret)
print_attributes(packet)
reply = self.CreateReplyPacket(packet, **{
'Service-Type': 'Framed-User',
'Framed-IP-Address': '192.168.0.1',
'Framed-IPv6-Prefix': 'fc66::/64'
})
reply = self.CreateReplyPacket(
packet,
**{
"Service-Type": "Framed-User",
"Framed-IP-Address": "192.168.0.1",
"Framed-IPv6-Prefix": "fc66::/64",
},
)
reply.code = AccessAccept
protocol.send_response(reply, addr)
def handle_acct_packet(self, protocol, packet, addr):
print('Received an accounting request')
print("Received an accounting request")
print_attributes(packet)
reply = self.CreateReplyPacket(packet)
protocol.send_response(reply, addr)
def handle_coa_packet(self, protocol, packet, addr):
print('Received an coa request')
print("Received an coa request")
print_attributes(packet)
reply = self.CreateReplyPacket(packet)
protocol.send_response(reply, addr)
def handle_disconnect_packet(self, protocol, packet, addr):
print('Received an disconnect request')
print("Received an disconnect request")
print_attributes(packet)
reply = self.CreateReplyPacket(packet)
@@ -77,17 +87,19 @@ def main(path_to_dictionary):
server = FakeServer(loop=loop, dictionary=Dictionary(path_to_dictionary))
# add clients (address, secret, name)
server.hosts['127.0.0.1'] = RemoteHost('127.0.0.1',
b'Kah3choteereethiejeimaeziecumi',
'localhost')
server.hosts["127.0.0.1"] = RemoteHost(
"127.0.0.1", b"Kah3choteereethiejeimaeziecumi", "localhost"
)
try:
# Initialize transports
loop.run_until_complete(
asyncio.ensure_future(
server.initialize_transports(enable_auth=True,
enable_acct=True,
enable_coa=True)))
server.initialize_transports(
enable_auth=True, enable_acct=True, enable_coa=True
)
)
)
try:
# start server
@@ -96,20 +108,22 @@ def main(path_to_dictionary):
pass
# Close transports
loop.run_until_complete(asyncio.ensure_future(
server.deinitialize_transports()))
loop.run_until_complete(
asyncio.ensure_future(server.deinitialize_transports())
)
except Exception as exc:
print('Error: ', exc)
print("Error: ", exc)
traceback.print_exc()
# Close transports
loop.run_until_complete(asyncio.ensure_future(
server.deinitialize_transports()))
loop.run_until_complete(
asyncio.ensure_future(server.deinitialize_transports())
)
loop.close()
if __name__ == '__main__':
dictionary = path.join(path.dirname(path.abspath(__file__)), 'dictionary')
if __name__ == "__main__":
dictionary = path.join(path.dirname(path.abspath(__file__)), "dictionary")
main(dictionary)