Auth funcional

.env.example

@@ -0,0 +1,3 @@
+DOMAIN=example.com.io
+PORT=4354
+PASSWORD=mypassword

FreeTwitch/FreeSwtichClient.py

@@ -0,0 +1,14 @@
+
+class FreeSwitchClient:
+
+    def __init__(self, client):
+        self.client = client
+        self.client.connect()
+
+    def authenticate(password):
+        command = "auth"
+        operation = " ".join([command, password])
+        self.client.send(operation)
+        response = self.client.getText()
+        expected_response = "Content-Type: command/reply\nReply-Text: +OK accepted"
+        assert expected_response in response

FreeTwitch/client.py

@@ -0,0 +1,39 @@
+import socket
+
+class Client:
+    timeout = 5
+    coding = 'utf-8'
+    buffer_size = 1024
+
+    def __init__(self, host, port, string_control='\n\n'):
+        self.host = host
+        self.port = port
+        self.string_control = string_control
+
+    def connect(self, expected_response=''):
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.settimeout(self.timeout)
+        self.socket.connect((self.host, self.port))
+
+        if expected_response:
+            return self.getText(len(expected_response))
+
+    def getText(self, length=0):
+        buffer_text = ''
+        texts = []
+        if length:
+            response = self.socket.recv(length).decode(self.coding)
+            self.socket.recv(len(self.string_control))
+            return response
+        while True:
+            buffer_text += self.socket.recv(self.buffer_size).decode(self.coding)
+            has_end = buffer_text.count(self.string_control)
+            if has_end:
+                return buffer_text.split(self.string_control)[0]        
+
+    def send(self, data):
+        fixed_data = data + self.string_control
+        self.socket.sendall(fixed_data.encode())
+
+    def close(self):
+        self.socket.close()

FreeTwitch/exceptions/freeswitch_connection_error.py

@@ -0,0 +1,9 @@
+class FreeSwitchConnectException(Exception):
+    def __init__(self, message=''):
+        basic_message = 'Error connecting to FreeSwitch Server.'
+        self.message = " ".join([basic_message, message])
+
+class FreeSwitchAuthException(Exception):
+    def __init__(self, message=''):
+        basic_message = 'Error authenticating on FreeSwitch Server.'
+        self.message = " ".join([basic_message, message])

FreeTwitch/freeswitch_client.py

@@ -0,0 +1,22 @@
+from exceptions.freeswitch_connection_error import FreeSwitchConnectException, FreeSwitchAuthException
+
+class FreeswitchClient:
+
+    def __init__(self, client):
+        self.client = client
+        self.__connect()
+
+    def __connect(self):
+        expected_connection_response = 'Content-Type: auth/request'
+        response = self.client.connect(expected_connection_response)
+        if not response == expected_connection_response:
+            raise FreeSwitchConnectException(response)
+
+    def authenticate(self, password):
+        command = "auth"
+        operation = " ".join([command, password])
+        self.client.send(operation)
+        response = self.client.getText()
+        expected_response = "Content-Type: command/reply\nReply-Text: +OK accepted"
+        if expected_response not in response:
+            raise FreeSwitchAuthException(response)

freetwitch.py

@@ -0,0 +1,16 @@
+from FreeTwitch.client import Client
+from FreeTwitch.freeswitch_client import FreeswitchClient
+from dotenv import dotenv_values
+
+config = dotenv_values(".env")
+print(config)
+host = config['HOST']
+port = int(config['PORT'])
+
+socket_client = Client(host, port)
+client = FreeswitchClient(socket_client)
+
+correct_password = config['PASSWORD']
+incorrect_password = 'la misma de su email'
+#client.authenticate(correct_password)
+client.authenticate(incorrect_password)

requirements.txt

@@ -0,0 +1 @@
+python-dotenv==1.0.0