Drop python2 support (#62)

* Drop Python 2 support (#53)

* drop python 2 support

* Remove python 3.7 from CI (#54)

it's not available on Ubuntu anymore since it's EOL

---------

Co-authored-by: Aaron McConnell <[email protected]>

* Fix missing 'not'

* add Python 3.11, 3.12 to CI (#56)

---------

Co-authored-by: Denitsa Hristova <[email protected]>
Co-authored-by: Aaron McConnell <[email protected]>
Co-authored-by: Denitsa Hristova <[email protected]>

Author: 3 people

.github/workflows/openvpn-ci.yml

@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
 
     steps:
       - uses: actions/checkout@v2

duo_openvpn.py

@@ -16,10 +16,9 @@
 import socket
 import sys
 import syslog
+import http.client as http_client
+import urllib.parse as urllib_parse
 
-import six
-from six.moves import http_client
-from six.moves.urllib.parse import quote, urlencode
 
 def log(msg):
     msg = 'Duo OpenVPN: %s' % msg
@@ -50,8 +49,8 @@ def canon_params(params):
     # http://tools.ietf.org/html/rfc5849#section-3.4.1.3.2
     args = []
     for (key, vals) in sorted(
-        (quote(key, '~'), vals) for (key, vals) in params.items()):
-        for val in sorted(quote(val, '~') for val in vals):
+        (urllib_parse.quote(key, '~'), vals) for (key, vals) in params.items()):
+        for val in sorted(urllib_parse.quote(val, '~') for val in vals):
             args.append('%s=%s' % (key, val))
     return '&'.join(args)
 
@@ -80,19 +79,19 @@ def sign(ikey, skey, method, host, uri, date, sig_version, params):
     """
     canonical = canonicalize(method, host, uri, params, date, sig_version)
 
-    if isinstance(skey, six.text_type):
+    if isinstance(skey, str):
         skey = skey.encode('utf-8')
-    if isinstance(canonical, six.text_type):
+    if isinstance(canonical, str):
         canonical = canonical.encode('utf-8')
 
     sig = hmac.new(skey, canonical, hashlib.sha512)
     auth = '%s:%s' % (ikey, sig.hexdigest())
 
-    if isinstance(auth, six.text_type):
+    if isinstance(auth, str):
         auth = auth.encode('utf-8')
 
     b64 = base64.b64encode(auth)
-    if not isinstance(b64, six.text_type):
+    if not isinstance(b64, str):
         b64 = b64.decode('utf-8')
 
     return 'Basic %s' % b64
@@ -106,12 +105,12 @@ def normalize_params(params):
     # urllib cannot handle unicode strings properly. quote() excepts,
     # and urlencode() replaces them with '?'.
     def encode(value):
-        if isinstance(value, six.text_type):
+        if isinstance(value, str):
             return value.encode("utf-8")
         return value
 
     def to_list(value):
-        if value is None or isinstance(value, six.string_types):
+        if value is None or isinstance(value, str):
             return [value]
         return value
 
@@ -185,11 +184,11 @@ def api_call(self, method, path, params):
 
         if method in ['POST', 'PUT']:
             headers['Content-type'] = 'application/x-www-form-urlencoded'
-            body = urlencode(params, doseq=True)
+            body = urllib_parse.urlencode(params, doseq=True)
             uri = path
         else:
             body = None
-            uri = path + '?' + urlencode(params, doseq=True)
+            uri = path + '?' + urllib_parse.urlencode(params, doseq=True)
 
         return self._make_request(method, uri, body, headers)
 
@@ -278,7 +277,7 @@ def raise_error(msg):
             error.data = data
             raise error
 
-        if not isinstance(data, six.text_type):
+        if not isinstance(data, str):
             data = data.decode('utf-8')
 
         if response.status != 200:
@@ -293,14 +292,14 @@ def raise_error(msg):
                         ))
                     else:
                         raise_error('Received %s %s' % (
-                                response.status,
+                            response.status,
                             data['message'],
                         ))
             except (ValueError, KeyError, TypeError):
                 pass
             raise_error('Received %s %s' % (
-                    response.status,
-                    response.reason,
+                response.status,
+                response.reason,
             ))
         try:
             data = json.loads(data)
@@ -313,18 +312,16 @@ def raise_error(msg):
 def success(control):
     log('writing success code to %s' % control)
 
-    f = open(control, 'w')
-    f.write('1')
-    f.close()
+    with open(control, 'w') as f:
+        f.write('1')
 
     sys.exit(0)
 
 def failure(control):
     log('writing failure code to %s' % control)
 
-    f = open(control, 'w')
-    f.write('0')
-    f.close()
+    with open(control, 'w') as f:
+        f.write('0')
 
     sys.exit(1)
 

https_wrapper.py

@@ -22,9 +22,9 @@
 import re
 import socket
 import ssl
-
-from six.moves import http_client
-from six.moves import urllib
+import http.client as http_client
+import urllib.request
+import urllib.error
 
 
 class InvalidCertificateException(http_client.HTTPException):
@@ -37,7 +37,7 @@ def __init__(self, host, cert, reason):
       host: The hostname the connection was made to.
       cert: The SSL certificate (as a dictionary) the host returned.
     """
-    http_client.HTTPException.__init__(self)
+    super().__init__()
     self.host = host
     self.cert = cert
     self.reason = reason
@@ -68,7 +68,7 @@ def __init__(self, host, port=None, key_file=None, cert_file=None,
       strict: When true, causes BadStatusLine to be raised if the status line
           can't be parsed as a valid HTTP/1.0 or 1.1 status line.
     """
-    http_client.HTTPConnection.__init__(self, host, port, strict, **kwargs)
+    super().__init__(host, port, strict, **kwargs)
     self.key_file = key_file
     self.cert_file = cert_file
     self.ca_certs = ca_certs
@@ -136,7 +136,7 @@ class CertValidatingHTTPSHandler(urllib.request.HTTPSHandler):
 
   def __init__(self, **kwargs):
     """Constructor. Any keyword args are passed to the http_client handler."""
-    urllib.request.HTTPSHandler.__init__(self)
+    super().__init__()
     self._connection_args = kwargs
 
   def https_open(self, req):
@@ -147,9 +147,9 @@ def http_class_wrapper(host, **kwargs):
     try:
       return self.do_open(http_class_wrapper, req)
     except urllib.error.URLError as e:
-      if type(e.reason) == ssl.SSLError and e.reason.args[0] == 1:
+      if isinstance(e.reason, ssl.SSLError) and e.reason.args[0] == 1:
         raise InvalidCertificateException(req.host, '',
                                           e.reason.args[1])
       raise
 
-  https_request = urllib.request.HTTPSHandler.do_request_
+https_request = urllib.request.HTTPSHandler.do_request_

test_duo_openvpn.py

@@ -1,11 +1,10 @@
 import email.utils
 import json
-import os
 import tempfile
 import unittest
 
-from mox3 import mox
-import six
+from unittest.mock import MagicMock
+import io
 
 import duo_openvpn
 
@@ -19,22 +18,22 @@ def mock_client_factory(mock):
     class MockClient(duo_openvpn.Client):
         def __init__(self, *args, **kwargs):
             mock.duo_client_init(*args, **kwargs)
-            super(MockClient, self).__init__(*args, **kwargs)
+            super().__init__(*args, **kwargs)
 
         def set_proxy(self, *args, **kwargs):
             mock.duo_client_set_proxy(*args, **kwargs)
-            return super(MockClient, self).set_proxy(*args, **kwargs)
+            return super().set_proxy(*args, **kwargs)
 
         def _connect(self):
             return mock
 
     return MockClient
 
-class MockResponse(six.StringIO, object):
+class MockResponse(io.StringIO):
     def __init__(self, status, body, reason='some reason'):
         self.status = status
         self.reason = reason
-        super(MockResponse, self).__init__(body)
+        super().__init__(body)
 
 class TestIntegration(unittest.TestCase):
     IKEY = 'expected ikey'
@@ -60,13 +59,9 @@ class TestIntegration(unittest.TestCase):
     )
 
     def setUp(self):
-        self.mox = mox.Mox()
-        self.expected_calls = self.mox.CreateMockAnything()
-
-    def assert_auth(self, environ, expected_control,
-                    send_control=True):
-        self.mox.ReplayAll()
+        self.expected_calls = MagicMock()
 
+    def assert_auth(self, environ, expected_control, send_control=True):
         with tempfile.NamedTemporaryFile() as control:
             if send_control:
                 environ['control'] = control.name
@@ -76,12 +71,9 @@ def assert_auth(self, environ, expected_control,
                     environ=environ,
                     Client=mock_client_factory(self.expected_calls),
                 )
-            self.mox.VerifyAll()
 
-            control.seek(0, os.SEEK_SET)
-            output = control.read()
-            if not isinstance(output, six.text_type):
-                output = output.decode('utf-8')
+            control.seek(0)
+            output = control.read().decode('utf-8')
             self.assertEqual(expected_control, output)
             if expected_control == '1':
                 self.assertEqual(0, cm.exception.args[0])
@@ -101,7 +93,7 @@ def normal_environ(self):
             ikey=self.IKEY,
             skey=self.SKEY,
             host=self.HOST,
-            user_agent=('duo_openvpn/' + duo_openvpn.__version__),
+            user_agent=self.EXPECTED_USER_AGENT,
         )
         self.expected_calls.duo_client_set_proxy(
             host=None,
@@ -114,22 +106,22 @@ def compare_params(self, recv_params, sent_params):
         return len(recv_params.split('&')) == len(stanzas) and all([s in recv_params for s in stanzas])
 
     def expect_request(self, method, path, params, params_func=None, response=None, raises=None):
-        if params_func == None:
+        if params_func is None:
             params_func = lambda p: self.compare_params(p, self.EXPECTED_PREAUTH_PARAMS)
-        self.expected_calls.request(method, path, mox.Func(params_func), {
+
+        self.expected_calls.request(
+            method, path, params_func, {
                 'User-Agent': self.EXPECTED_USER_AGENT,
                 'Host': self.HOST,
                 'Content-type': 'application/x-www-form-urlencoded',
-                'Authorization': mox.Func((lambda s: s.startswith('Basic ') and not s.startswith('Basic b\''))),
-                'Date': mox.Func((lambda s: bool(email.utils.parsedate_tz(s))))
-            },
+                'Authorization': MagicMock(side_effect=lambda s: s.startswith('Basic ') and not s.startswith('Basic b\'')),
+                'Date': MagicMock(side_effect=lambda s: bool(email.utils.parsedate_tz(s)))
+            }
         )
-        meth = self.expected_calls.getresponse()
-        if raises is not None:
-            meth.AndRaise(raises)
+        if raises:
+            self.expected_calls.getresponse.side_effect = raises
         else:
-            meth.AndReturn(response)
-            self.expected_calls.close()
+            self.expected_calls.getresponse.return_value = response
 
     def expect_preauth(self, result, path=EXPECTED_PREAUTH_PATH, factor='push1'):
         self.expect_request(
@@ -139,12 +131,12 @@ def expect_preauth(self, result, path=EXPECTED_PREAUTH_PATH, factor='push1'):
             response=MockResponse(
                 status=200,
                 body=json.dumps({
-                        'stat': 'OK',
-                        'response': {
-                            'result': result,
-                            'status': 'expected status',
-                            'factors': {'default': factor},
-                        },
+                    'stat': 'OK',
+                    'response': {
+                        'result': result,
+                        'status': 'expected status',
+                        'factors': {'default': factor},
+                    },
                 }),
             ),
         )
@@ -158,11 +150,11 @@ def expect_auth(self, result, path=EXPECTED_AUTH_PATH):
             response=MockResponse(
                 status=200,
                 body=json.dumps({
-                        'stat': 'OK',
-                        'response': {
-                            'result': result,
-                            'status': 'expected status',
-                        },
+                    'stat': 'OK',
+                    'response': {
+                        'result': result,
+                        'status': 'expected status',
+                    },
                 }),
             ),
         )