Merge pull request #1 from wh0am1i/dev

fix: no moudel telnetlib in py3.13

pocsuite3/__init__.py

@@ -1,5 +1,5 @@
 __title__ = 'pocsuite3'
-__version__ = '2.0.9'
+__version__ = '2.0.10'
 __author__ = 'Knownsec 404 Team'
 __author_email__ = '[email protected]'
 __license__ = 'GPLv2'

pocsuite3/modules/listener/bind_tcp.py

@@ -4,7 +4,6 @@
 import pickle
 import base64
 import select
-import telnetlib
 import threading
 from pocsuite3.lib.core.poc import POCBase
 from pocsuite3.lib.utils import random_str
@@ -25,25 +24,18 @@ def read_inputs(s):
     return b''.join(buffer)
 
 
+def read_until(conn, inputs):
+    try:
+        while True:
+            msg = conn.recv(1024).decode('utf-8', errors='ignore')
+            if inputs in msg.lower():
+                break
+    except Exception:
+        pass
+
+
 def read_results(conn, inputs):
-    if isinstance(conn, telnetlib.Telnet):
-        flag = random_str(6).encode()
-        inputs = inputs.strip() + b';' + flag + b'\n'
-        results = b''
-        conn.write(inputs)
-        count = 10
-        while count:
-            count -= 1
-            chunk = conn.read_until(random_str(6).encode(), 0.2)
-            if len(chunk) > 0:
-                results += chunk
-            if results.count(flag) >= 2:
-                # remove the Telnet input echo
-                results = results.split(inputs.strip())[-1]
-                results = os.linesep.encode().join(
-                    results.split(flag)[0].splitlines()[0:-1])
-                return results.strip() + b'\n'
-    elif callable(conn):
+    if callable(conn):
         results = conn(inputs.decode())
         if not isinstance(results, bytes):
             results = results.encode()
@@ -116,15 +108,16 @@ def bind_tcp_shell(host, port, check=True):
 
 
 def bind_telnet_shell(host, port, user, pwd, check=True):
+    # see https://peps.python.org/pep-0594/#telnetlib
     if not check_port(host, port):
         return False
     try:
-        tn = telnetlib.Telnet(host, port)
-        tn.expect([b'Login: ', b'login: '], 10)
-        tn.write(user.encode() + b'\n')
-        tn.expect([b'Password: ', b'password: '], 10)
-        tn.write(pwd.encode() + b'\n')
-        tn.write(b'\n')
+        tn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        tn.connect((host, port))
+        read_until(tn, 'login: ')
+        tn.sendall((user + "\n").encode('utf-8'))
+        read_until(tn, 'password: ')
+        tn.sendall((pwd + "\n").encode('utf-8'))
         if check:
             flag = random_str(6).encode()
             if flag not in read_results(tn, b'echo %s' % flag):

requirements.txt

@@ -16,3 +16,4 @@ PyYAML >= 6.0
 lxml >= 4.6.0
 mmh3 >= 3.0.0
 docker >=  6.1.3
+packaging

setup.py

@@ -21,7 +21,7 @@ def find_packages(where='.'):
 
 setup(
     name='pocsuite3',
-    version='2.0.9',
+    version='2.0.10',
     url='https://pocsuite.org',
     description='Open-sourced remote vulnerability testing framework.',
     long_description=long_description,

tests/test_cmd_diy_options.py

@@ -5,11 +5,15 @@
 # @File    : test_cmd_diy_options.py
 import os
 import unittest
+import warnings
+
+from urllib3.exceptions import InsecureRequestWarning
 
 
 class TestCase(unittest.TestCase):
     def setUp(self):
-        pass
+        warnings.simplefilter("ignore", ResourceWarning)
+        warnings.simplefilter("ignore", InsecureRequestWarning)
 
     def tearDown(self):
         pass
@@ -23,7 +27,7 @@ def test_cmd_run(self):
         eval_path = os.path.join(path, "../pocsuite3/cli.py")
         poc_path = os.path.join(path, "login_demo.py")
         command = (
-            f'python3 {eval_path} -u https://example.com -r {poc_path} --verify -v 2  --password mypass123 '
+            f'python3 {eval_path} -u http://httpbin.org/post -r {poc_path} --verify -v 2  --password mypass123 '
             '--username "asd asd" --testt abctest'
         )
         pipeline = os.popen(command)