Feature/tunneling

keepercommander/commands/discoveryrotation.py

@@ -67,7 +67,6 @@ def __init__(self):
         self.register_command('config', PAMConfigurationsCommand(), 'Manage PAM Configurations', 'c')
         self.register_command('rotation', PAMRotationCommand(), 'Manage Rotations', 'r')
         self.register_command('action', GatewayActionCommand(), 'Execute action on the Gateway', 'a')
-        self.register_command('tunnel', PAMTunnelCommand(), 'Manage Tunnels', 't')
 
 
 class PAMGatewayCommand(GroupCommand):
@@ -127,6 +126,7 @@ def __init__(self):
         self.register_command('job-cancel', PAMGatewayActionJobCommand(), 'View Job details', 'jc')
 
         # self.register_command('job-list', DRCmdListJobs(), 'List Running jobs')
+        # self.register_command('tunnel', DRTunnelCommand(), 'Tunnel to the server')
 
 
 class PAMCmdListJobs(Command):
@@ -1935,9 +1935,6 @@ def execute(self, params, **kwargs):
 
         gateway_public_key_bytes = retrieve_gateway_public_key(gateway_uid, params, api, utils)
 
-        # TODO remove debug code
-        print("PUBLIC KEY FOUND: ", gateway_public_key_bytes)
-
         record = params.record_cache.get(record_uid)
         if not record:
             print(f"{bcolors.FAIL}Record {record_uid} not found.{bcolors.ENDC}")

keepercommander/commands/tunnel/port_forward/endpoint.py

@@ -94,6 +94,7 @@ def generate_secure_self_signed_cert(private_key_str):   # type: (str) -> Tuple[
     :param private_key_str: PEM-formatted private key as a string.
     :return: Tuple containing the PEM-formatted certificate and private key
     """
+    # This is the code that generates a new private key
     '''
     # Generate an EC private key
     private_key = ec.generate_private_key(
@@ -116,24 +117,6 @@ def generate_secure_self_signed_cert(private_key_str):   # type: (str) -> Tuple[
         password=None,
         backend=default_backend()
     )
-    #
-    # subject = issuer = x509.Name([
-    #     x509.NameAttribute(NameOID.COMMON_NAME, u"localhost"),
-    # ])
-    # cert = (
-    #     x509.CertificateBuilder()
-    #     .subject_name(subject)
-    #     .issuer_name(issuer)
-    #     .public_key(private_key.public_key())
-    #     .serial_number(x509.random_serial_number())
-    #     .not_valid_before(datetime.datetime.utcnow())
-    #     .not_valid_after(
-    #         # Our certificate will be valid for 10 days
-    #         datetime.datetime.utcnow() + datetime.timedelta(days=10)
-    #     )
-    #     .sign(private_key, hashes.SHA256(), default_backend())
-    # )
-    # cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode('utf-8')
 
     # Define subject and issuer
     subject = issuer = x509.Name([
@@ -804,7 +787,7 @@ async def incoming_forward(f_writer):
 
             client_to_remote = asyncio.create_task(out_going_forward(forwarder_reader))
             remote_to_client = asyncio.create_task(incoming_forward(forwarder_writer))
-            
+
             self.client_tasks.extend([client_to_remote, remote_to_client])
             self.forwarder_event.set()
         except Exception as e:

unit-tests/pam-tunnel/test_pam_tunnel.py

@@ -1,14 +1,79 @@
+import datetime
 import socket
 import string
 import unittest
 from unittest import mock
-from keepercommander.commands.tunnel.port_forward.endpoint import (generate_random_bytes, find_open_port)
+
+from cryptography import x509
+from cryptography.hazmat._oid import NameOID
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization, hashes
+from cryptography.hazmat.primitives.asymmetric import ec
+
+from keepercommander.commands.tunnel.port_forward.endpoint import (generate_random_bytes, find_open_port,
+                                                                   verify_tls_certificate)
+
+
+def generate_self_signed_cert(private_key):
+    # Generate a self-signed certificate
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, u"localhost"),
+    ])
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.datetime.utcnow())
+        .not_valid_after(
+            # Our certificate will be valid for 10 days
+            datetime.datetime.utcnow() + datetime.timedelta(days=10)
+        )
+        .sign(private_key, hashes.SHA256(), default_backend())
+    )
+    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode('utf-8')
+
+    return cert_pem
+
+
+def new_private_key():
+    # Generate an EC private key
+    private_key = ec.generate_private_key(
+        ec.SECP256R1(),  # Using P-256 curve
+        backend=default_backend()
+    )
+    # Serialize to PEM format
+    private_key_str = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption()
+    ).decode('utf-8')
+    return private_key, private_key_str
 
 
 class TestVerifyTLSCertificate(unittest.TestCase):
     # TODO: Test that the TLS certificate is verified correctly when we figure it out
+    def setUp(self):
+        self.private_key, self.private_key_str = new_private_key()
+        self.public_cert = generate_self_signed_cert(self.private_key)
+
     def test_verify_tls_certificate(self):
-        pass
+        # Test that the TLS certificate is verified correctly
+        public_key = self.private_key.public_key()
+        trusted = verify_tls_certificate(self.public_cert,
+                                         public_key.public_bytes(encoding=serialization.Encoding.X962,
+                                                                 format=serialization.PublicFormat.UncompressedPoint))
+        self.assertTrue(trusted)
+
+    def test_failed_verify_tls_certificates(self):
+        # Test that the TLS certificate is verified correctly
+        new_private, private_key_str = new_private_key()
+        public_key = new_private.public_key()
+        trusted = verify_tls_certificate(self.public_cert,
+                                         public_key.public_bytes(encoding=serialization.Encoding.X962,
+                                                                 format=serialization.PublicFormat.UncompressedPoint))
+        self.assertFalse(trusted)
 
 
 class TestFindOpenPort(unittest.TestCase):

unit-tests/pam-tunnel/test_private_tunnel.py

@@ -1,32 +1,34 @@
 import asyncio
-import logging
-import unittest
 import hashlib
 import hmac
+import logging
 import ssl
-from asyncio import StreamWriter, IncompleteReadError
-from unittest import mock
-from unittest.mock import call
+import sys
+import unittest
 
 from cryptography.utils import int_to_bytes
 from keeper_secrets_manager_core.utils import bytes_to_base64
-
 from keepercommander import utils
 from keepercommander.commands.tunnel.port_forward.endpoint import (PrivateTunnelEntrance, ControlMessage,
                                                                    CONTROL_MESSAGE_NO_LENGTH, CONNECTION_NO_LENGTH,
                                                                    HMACHandshakeFailedException,
                                                                    ConnectionNotFoundException, generate_random_bytes,
                                                                    TERMINATOR, DATA_LENGTH)
+from test_pam_tunnel import generate_self_signed_cert, new_private_key
+from unittest import mock, skipIf
 
 
+@skipIf(sys.version_info < (3, 11), "requires Python 3.11 or higher")
 class TestPrivateTunnelEntrance(unittest.IsolatedAsyncioTestCase):
     async def asyncSetUp(self):
         self.event = asyncio.Event()
         self.host = 'localhost'
         self.port = 8080
         self.public_tunnel_port = 8081
         self.endpoint_name = 'TestEndpoint'
-        self.cert = 'your_cert_here'
+
+        self.private_key, self.private_key_str = new_private_key()
+        self.cert = generate_self_signed_cert(self.private_key)
         self.logger = mock.MagicMock(spec=logging)
         self.kill_server_event = asyncio.Event()
         self.tunnel_symmetric_key = utils.generate_aes_key()
@@ -113,8 +115,8 @@ async def test_forward_data_to_local_normal(self):
 
         self.pte.connections[1][1].write.assert_called_with(b'some_data')
         self.pte.connections[1][1].drain.assert_called_once()
-        self.assertTrue(self.pte.logger.method_calls[3] == (call.debug('Endpoint TestEndpoint: Forwarding private data '
-                                                            'to local for connection 1 (9)')))
+        self.assertTrue(self.pte.logger.method_calls[3] == (mock.call.debug('Endpoint TestEndpoint: Forwarding private '
+                                                                            'data to local for connection 1 (9)')))
 
     async def test_forward_data_to_local_error(self):
         self.pte.tls_reader = mock.MagicMock(spec=asyncio.StreamReader)
@@ -245,33 +247,40 @@ async def test_start_tls_reader_generic_exception(self):
             self.assertFalse(self.pte.is_connected)
 
     async def test_perform_ssl_handshakes_success(self):
-        self.pte.tls_writer = mock.MagicMock(spec=StreamWriter)
+        self.pte.tls_writer = mock.MagicMock(spec=asyncio.StreamWriter)
         self.pte.logger = mock.MagicMock()
-        with mock.patch('ssl.create_default_context') as mock_ssl_context:
+        with mock.patch('ssl.SSLContext') as mock_ssl_context:
             mock_context = mock.MagicMock()
             mock_ssl_context.return_value = mock_context
 
             await self.pte.perform_ssl_handshakes()
 
-            mock_context.load_verify_locations.assert_called_with(cadata=self.pte.cert)
+            mock_context.load_verify_locations.assert_called_with(cadata=self.pte.server_public_cert)
             self.pte.tls_writer.start_tls.assert_called_with(mock_context, server_hostname='localhost')
             self.pte.logger.debug.assert_called_with('Endpoint TestEndpoint: TLS connection established successfully.')
 
     async def test_perform_ssl_handshakes_start_tls_exception(self):
-        self.pte.tls_writer = mock.MagicMock(spec=StreamWriter)
-        self.pte.tls_writer.start_tls.side_effect = IncompleteReadError(b'', 1)  # Pass bytes as the first argument
+        self.pte.tls_writer = mock.MagicMock(spec=asyncio.StreamWriter)
+        # Pass bytes as the first argument
+        self.pte.tls_writer.start_tls.side_effect = asyncio.IncompleteReadError(b'', 1)
         with mock.patch('ssl.create_default_context') as mock_ssl_context:
             mock_context = mock.MagicMock()
             mock_ssl_context.return_value = mock_context
 
-            with self.assertRaises(IncompleteReadError):
+            with self.assertRaises(asyncio.IncompleteReadError):
                 await self.pte.perform_ssl_handshakes()
 
     async def test_perform_ssl_handshakes_load_verify_locations_exception(self):
-        with mock.patch('ssl.create_default_context') as mock_ssl_context:
-            mock_context = mock.MagicMock()
+        with mock.patch('ssl.SSLContext', new_callable=mock.MagicMock) as MockSSLContext:
+            # No need to specify spec here, as MockSSLContext is already a mock of ssl.SSLContext
+            mock_context = MockSSLContext.return_value
+
+            # Set the side effect for the load_verify_locations method
             mock_context.load_verify_locations.side_effect = FileNotFoundError
-            mock_ssl_context.return_value = mock_context
+
+            self.pte.tls_writer = mock.MagicMock(spec=asyncio.StreamWriter)
+            # Mock asyncio.open_connection
+            self.pte.tls_reader = mock.MagicMock(spec=asyncio.StreamReader)
 
             with self.assertRaises(FileNotFoundError):
                 await self.pte.perform_ssl_handshakes()
@@ -293,7 +302,7 @@ async def test_perform_ssl_handshakes_cert_failure(self):
 
     # Test Server Hostname Mismatch
     async def test_perform_ssl_handshakes_hostname_mismatch(self):
-        self.pte.tls_writer = mock.MagicMock(spec=StreamWriter)
+        self.pte.tls_writer = mock.MagicMock(spec=asyncio.StreamWriter)
         self.pte.tls_writer.start_tls.side_effect = ssl.SSLCertVerificationError("Hostname mismatch")
         with mock.patch('ssl.create_default_context') as mock_ssl_context:
             mock_context = mock.MagicMock()
@@ -353,10 +362,10 @@ async def read_side_effect(*args, **kwargs):
             # Assert that send_control_message was called with ControlMessage.Ping three times
             # and then with ControlMessage.CloseConnection
             expected_calls = [
-                call(ControlMessage.Ping),
-                call(ControlMessage.Ping),
-                call(ControlMessage.Ping),
-                call(ControlMessage.CloseConnection, int.to_bytes(1, CONNECTION_NO_LENGTH, byteorder='big'))
+                mock.call(ControlMessage.Ping),
+                mock.call(ControlMessage.Ping),
+                mock.call(ControlMessage.Ping),
+                mock.call(ControlMessage.CloseConnection, int.to_bytes(1, CONNECTION_NO_LENGTH, byteorder='big'))
             ]
             mock_send_control_message.assert_has_calls(expected_calls)
 

unit-tests/pam-tunnel/test_public_tunnel.py

@@ -1,19 +1,19 @@
 import asyncio
-import unittest
-from unittest import mock
 import hashlib
 import hmac
-
+import sys
+import unittest
 from keeper_secrets_manager_core.utils import bytes_to_base64
-
 from keepercommander import utils
 from keepercommander.commands.tunnel.port_forward.tunnel import ITunnel
 from keepercommander.commands.tunnel.port_forward.endpoint import (ControlMessage, CONTROL_MESSAGE_NO_LENGTH,
                                                                    DATA_LENGTH, CONNECTION_NO_LENGTH, TunnelProtocol,
-                                                                   PlainTextForwarder, generate_random_bytes,
-                                                                   TERMINATOR)
+                                                                   TERMINATOR, PlainTextForwarder,
+                                                                   generate_random_bytes)
+from unittest import mock, skipIf
 
 
+@skipIf(sys.version_info < (3, 11), "requires Python 3.11 or higher")
 class TestPublicTunnel(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):
@@ -199,6 +199,7 @@ async def test_read_connection_with_multiple_data(self):
             self.assertEqual(mock_send_data.call_count, 2)
 
 
+@skipIf(sys.version_info < (3, 11), "requires Python 3.11 or higher")
 class TestPlainTextForwarder(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):