Fix PAM tunnel diagnose proxy support

keepercommander/commands/tunnel_and_connections.py

@@ -11,14 +11,13 @@
 
 import argparse
 import datetime
-import http.client
 import json
 import logging
 import os
 import platform
+import requests
 import signal
 import socket
-import ssl
 import struct
 import subprocess
 import sys
@@ -1732,48 +1731,60 @@ def _parse_stun(cls, data: bytes) -> dict:
 
     # ── individual Python-side tests ──────────────────────────────────────────
     @classmethod
-    def _test_https(cls, hostname: str, port: int = 443) -> Tuple[bool, str, int]:
+    def _test_https(cls, hostname: str, port: int = 443, proxies=None, verify=True) -> Tuple[bool, str, int]:
         """Returns (passed, detail, ms)."""
         t0 = time.monotonic()
-        conn = None
+        resp = None
         try:
-            ctx = ssl.create_default_context()
-            conn = http.client.HTTPSConnection(hostname, port=port, context=ctx, timeout=10)
-            conn.request('GET', '/', headers={'User-Agent': 'keeper-pam-diagnose/1.0'})
-            resp = conn.getresponse()
+            resp = requests.get(
+                f'https://{hostname}:{port}/',
+                headers={'User-Agent': 'keeper-pam-diagnose/1.0'},
+                proxies=proxies,
+                verify=verify,
+                timeout=10,
+                stream=True,
+            )
             ms = int((time.monotonic() - t0) * 1000)
-            return 100 <= resp.status < 400, f'HTTP {resp.status}  (reachable)', ms
+            return 100 <= resp.status_code < 400, f'HTTP {resp.status_code}  (reachable)', ms
         except Exception as exc:
             return False, str(exc)[:60], int((time.monotonic() - t0) * 1000)
         finally:
-            if conn:
-                try: conn.close()
-                except Exception: pass
+            if resp is not None:
+                try:
+                    resp.close()
+                except Exception:
+                    pass
 
     @classmethod
-    def _test_websocket(cls, hostname: str, port: int = 443) -> Tuple[bool, str, int]:
+    def _test_websocket(cls, hostname: str, port: int = 443, proxies=None, verify=True) -> Tuple[bool, str, int]:
         """HTTP Upgrade probe — any 4xx means the server is reachable."""
         t0 = time.monotonic()
-        conn = None
+        resp = None
         try:
-            ctx = ssl.create_default_context()
-            conn = http.client.HTTPSConnection(hostname, port=port, context=ctx, timeout=10)
-            conn.request('GET', '/', headers={
-                'Upgrade': 'websocket',
-                'Connection': 'Upgrade',
-                'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
-                'Sec-WebSocket-Version': '13',
-                'User-Agent': 'keeper-pam-diagnose/1.0',
-            })
-            resp = conn.getresponse()
+            resp = requests.get(
+                f'https://{hostname}:{port}/',
+                headers={
+                    'Upgrade': 'websocket',
+                    'Connection': 'Upgrade',
+                    'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
+                    'Sec-WebSocket-Version': '13',
+                    'User-Agent': 'keeper-pam-diagnose/1.0',
+                },
+                proxies=proxies,
+                verify=verify,
+                timeout=10,
+                stream=True,
+            )
             ms = int((time.monotonic() - t0) * 1000)
-            return 100 <= resp.status < 400, f'HTTP {resp.status}', ms
+            return 100 <= resp.status_code < 400, f'HTTP {resp.status_code}', ms
         except Exception as exc:
             return False, str(exc)[:60], int((time.monotonic() - t0) * 1000)
         finally:
-            if conn:
-                try: conn.close()
-                except Exception: pass
+            if resp is not None:
+                try:
+                    resp.close()
+                except Exception:
+                    pass
 
     @classmethod
     def _test_tcp_stun(cls, hostname: str) -> Tuple[bool, str, int, Optional[str]]:
@@ -1963,10 +1974,12 @@ def _record(name: str, passed: bool, detail: str, ms: int):
         except Exception as exc:
             _record(f'DNS  {server_host}', False, str(exc)[:60], int((time.monotonic() - t0) * 1000))
 
-        passed, detail, ms = self._test_https(server_host)
+        passed, detail, ms = self._test_https(
+            server_host, proxies=params.rest_context.proxies, verify=params.ssl_verify)
         _record(f'HTTPS  {server_host}:443', passed, detail, ms)
 
-        passed, detail, ms = self._test_websocket(connect_host)
+        passed, detail, ms = self._test_websocket(
+            connect_host, proxies=params.rest_context.proxies, verify=params.ssl_verify)
         _record(f'WebSocket  {connect_host}:443', passed, detail, ms)
 
         print()

unit-tests/pam/test_pam_tunnel.py

@@ -2,6 +2,7 @@
 from unittest import mock
 
 from keepercommander.error import CommandError
+from keepercommander.commands.tunnel_and_connections import PAMTunnelDiagnoseCommand
 
 import datetime
 import socket
@@ -154,3 +155,83 @@ def test_uniqueness(self):
         random_bytes1 = generate_random_bytes()
         random_bytes2 = generate_random_bytes()
         self.assertNotEqual(random_bytes1, random_bytes2)
+
+
+class TestPAMTunnelDiagnose(unittest.TestCase):
+    def test_execute_passes_session_proxy_to_https_probes(self):
+        proxies = {'http': 'http://proxy.example:8080', 'https': 'http://proxy.example:8080'}
+        params = mock.MagicMock()
+        params.server = 'keepersecurity.com'
+        params.rest_context.proxies = proxies
+        params.ssl_verify = '/path/to/ca.pem'
+
+        with mock.patch('keepercommander.commands.tunnel_and_connections.get_relay_host',
+                        return_value='relay.example'), \
+                mock.patch('keepercommander.commands.tunnel_and_connections.get_router_host',
+                           return_value='router.example'), \
+                mock.patch('keepercommander.commands.tunnel_and_connections.get_or_create_tube_registry',
+                           return_value=None), \
+                mock.patch('keepercommander.commands.tunnel_and_connections.socket.getaddrinfo',
+                           return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('203.0.113.1', 0))]), \
+                mock.patch('keepercommander.commands.tunnel_and_connections.socket.gethostbyname',
+                           return_value='203.0.113.1'), \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_https',
+                                  return_value=(True, 'reachable', 1)) as test_https, \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_websocket',
+                                  return_value=(True, 'reachable', 1)) as test_websocket, \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_tcp_stun',
+                                  return_value=(True, 'reachable', 1, None)), \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_udp_stun',
+                                  return_value=(True, 'reachable', 1, None)), \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_turn',
+                                  return_value=(True, 'reachable', 1)), \
+                mock.patch.object(PAMTunnelDiagnoseCommand, '_test_udp_port',
+                                  return_value=(True, 1)):
+            PAMTunnelDiagnoseCommand().execute(params)
+
+        test_https.assert_called_once_with(
+            'keepersecurity.com', proxies=proxies, verify='/path/to/ca.pem')
+        test_websocket.assert_called_once_with(
+            'router.example', proxies=proxies, verify='/path/to/ca.pem')
+
+    def test_https_uses_configured_proxy(self):
+        proxies = {'http': 'http://proxy.example:8080', 'https': 'http://proxy.example:8080'}
+        with mock.patch('keepercommander.commands.tunnel_and_connections.requests.get') as mock_get:
+            mock_get.return_value.status_code = 200
+
+            passed, _, _ = PAMTunnelDiagnoseCommand._test_https(
+                'api.example', proxies=proxies, verify='/path/to/ca.pem')
+
+        self.assertTrue(passed)
+        mock_get.assert_called_once_with(
+            'https://api.example:443/',
+            headers={'User-Agent': 'keeper-pam-diagnose/1.0'},
+            proxies=proxies,
+            verify='/path/to/ca.pem',
+            timeout=10,
+            stream=True,
+        )
+
+    def test_websocket_uses_configured_proxy(self):
+        proxies = {'http': 'http://proxy.example:8080', 'https': 'http://proxy.example:8080'}
+        with mock.patch('keepercommander.commands.tunnel_and_connections.requests.get') as mock_get:
+            mock_get.return_value.status_code = 101
+
+            passed, _, _ = PAMTunnelDiagnoseCommand._test_websocket(
+                'router.example', proxies=proxies, verify='/path/to/ca.pem')
+
+        self.assertTrue(passed)
+        mock_get.assert_called_once_with(
+            'https://router.example:443/',
+            headers={
+                'Upgrade': 'websocket',
+                'Connection': 'Upgrade',
+                'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
+                'Sec-WebSocket-Version': '13',
+                'User-Agent': 'keeper-pam-diagnose/1.0',
+            },
+            proxies=proxies,
+            verify='/path/to/ca.pem',
+            timeout=10,
+            stream=True,
+        )