Fix deprecation and resource warnings.

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

src/saml2/__init__.py

@@ -586,7 +586,8 @@ def get_ns_map(self, elements, uri_set):
 
         for elem in elements:
             uri_set = self.get_ns_map_attribute(elem.attrib, uri_set)
-            uri_set = self.get_ns_map(elem.getchildren(), uri_set)
+            children = list(elem)
+            uri_set = self.get_ns_map(children, uri_set)
             uri = self.tag_get_uri(elem)
             if uri is not None:
                 uri_set.add(uri)
@@ -651,7 +652,7 @@ def set_prefixes(self, elem, prefix_map):
 
         # fixup all elements in the tree
         memo = {}
-        for elem in elem.getiterator():
+        for elem in elem.iter():
             self.fixup_element_prefixes(elem, uri_map, memo)
 
     def fixup_element_prefixes(self, elem, uri_map, memo):

src/saml2/cert.py

@@ -150,7 +150,6 @@ def create_certificate(self, cert_info, request=False, valid_from=0,
         cert.set_pubkey(k)
         cert.sign(k, hash_alg)
 
-        filesCreated = False
         try:
             if request:
                 tmp_cert = crypto.dump_certificate_request(crypto.FILETYPE_PEM,
@@ -169,33 +168,18 @@ def create_certificate(self, cert_info, request=False, valid_from=0,
             else:
                 tmp_key = crypto.dump_privatekey(crypto.FILETYPE_PEM, k)
             if write_to_file:
-                fc = open(c_f, "wt")
-                fk = open(k_f, "wt")
-
-                if request:
-                    fc.write(tmp_cert.decode('utf-8'))
-                else:
+                with open(c_f, 'wt') as fc:
                     fc.write(tmp_cert.decode('utf-8'))
-                fk.write(tmp_key.decode('utf-8'))
-                filesCreated = True
-                try:
-                    fc.close()
-                except:
-                    pass
-
-                try:
-                    fk.close()
-                except:
-                    pass
+                with open(k_f, 'wt') as fk:
+                    fk.write(tmp_key.decode('utf-8'))
                 return c_f, k_f
             return tmp_cert, tmp_key
         except Exception as ex:
             raise CertificateError("Certificate cannot be generated.", ex)
 
     def write_str_to_file(self, file, str_data):
-        f = open(file, "wt")
-        f.write(str_data)
-        f.close()
+        with open(file, 'wt') as f:
+            f.write(str_data)
 
     def read_str_from_file(self, file, type="pem"):
         with open(file, 'rb') as f:

src/saml2/config.py

@@ -373,7 +373,6 @@ def load_file(self, config_file, metadata_construction=False):
             config_file = config_file[:-3]
 
         mod = self._load(config_file)
-        # return self.load(eval(open(config_file).read()))
         return self.load(copy.deepcopy(mod.CONFIG), metadata_construction)
 
     def load_metadata(self, metadata_conf):

src/saml2/eptid.py

@@ -56,6 +56,9 @@ def get(self, idp, sp, *args):
             self[key] = val
             return val
 
+    def close(self):
+        pass
+
 
 class EptidShelve(Eptid):
     def __init__(self, secret, filename):
@@ -64,3 +67,6 @@ def __init__(self, secret, filename):
             if filename.endswith('.db'):
                 filename = filename.rsplit('.db', 1)[0]
         self._db = shelve.open(filename, writeback=True, protocol=2)
+
+    def close(self):
+        self._db.close()

src/saml2/mdbcache.py

@@ -31,7 +31,7 @@ def __init__(self, server=None, debug=0, db=None):
         self.debug = debug
 
     def delete(self, subject_id):
-        self._cache.remove({"subject_id": subject_id})
+        self._cache.delete_many({'subject_id': subject_id})
 
     def get_identity(self, subject_id, entities=None,
                      check_not_on_or_after=True):
@@ -196,4 +196,4 @@ def valid_to(self, subject_id, entity_id, newtime):
                            {"$set": {"timestamp": newtime}})
 
     def clear(self):
-        self._cache.remove()
+        self._cache.delete_many({})

src/saml2/mongo_store.py

@@ -178,11 +178,7 @@ def handle_name_id_mapping_request(self, name_id, name_id_policy):
         # else create and return a new one
         return self.construct_nameid(_id, name_id_policy=name_id_policy)
 
-    def close(self):
-        pass
-
 
-#------------------------------------------------------------------------------
 class MDB(object):
     primary_key = "mdb"
 

src/saml2/s2repoze/plugins/formswithhidden.py

@@ -116,7 +116,8 @@ def make_plugin(login_form_qs='__do_login', rememberer_name=None, form=None):
         raise ValueError(
             'must include rememberer key (name of another IIdentifier plugin)')
     if form is not None:
-        form = open(form).read()
+        with open(form, 'r') as f:
+            form = f.read()
     plugin = FormHiddenPlugin(login_form_qs, rememberer_name, form)
     return plugin
 

src/saml2/saml.py

@@ -3,6 +3,8 @@
 #
 # Generated Mon May  2 14:23:33 2011 by parse_xsd.py version 0.4.
 #
+import base64
+
 from saml2.validate import valid_ipv4, MustValueError
 from saml2.validate import valid_ipv6
 from saml2.validate import ShouldValueError
@@ -98,9 +100,8 @@ def _decode_attribute_value(typ, text):
     if typ == XSD + "boolean":
         return "%s" % (text == "true" or text == "True")
     if typ == XSD + "base64Binary":
-        import base64
-
-        return base64.decodestring(text)
+        decode_fn = getattr(base64, 'decodebytes', base64.decodestring)
+        return decode_fn(text)
     raise ValueError("type %s not supported" % type)
 
 
@@ -124,9 +125,8 @@ def _verify_value_type(typ, val):
         else:
             raise ValueError("Faulty boolean value")
     if typ == XSD + "base64Binary":
-        import base64
-
-        return base64.decodestring(val.encode('utf-8'))
+        decode_fn = getattr(base64, 'decodebytes', base64.decodestring)
+        return decode_fn(val.encode())
 
 
 class AttributeValueBase(SamlBase):
@@ -204,13 +204,11 @@ def clear_type(self):
     def set_text(self, val, base64encode=False):
         typ = self.get_type()
         if base64encode:
-            import base64
-
             val = base64.encodestring(val)
             self.set_type("xs:base64Binary")
         else:
             if isinstance(val, six.binary_type):
-                val = val.decode('utf-8')
+                val = val.decode()
             if isinstance(val, six.string_types):
                 if not typ:
                     self.set_type("xs:string")

src/saml2/server.py

@@ -26,6 +26,9 @@
 from saml2.samlp import NameIDMappingResponse
 from saml2.sdb import SessionStorage
 from saml2.schema import soapenv
+from saml2.mongo_store import EptidMDB
+from saml2.mongo_store import IdentMDB
+from saml2.mongo_store import SessionStorageMDB
 
 from saml2.request import AuthnRequest
 from saml2.request import AssertionIDRequest
@@ -112,8 +115,6 @@ def choose_session_storage(self):
         else:  # Should be tuple
             typ, data = _spec
             if typ.lower() == "mongodb":
-                from saml2.mongo_store import SessionStorageMDB
-
                 return SessionStorageMDB(database=data, collection="session")
 
         raise NotImplementedError("No such storage type implemented")
@@ -142,13 +143,10 @@ def init_config(self, stype="idp"):
                 idb = _shelve_compat(addr, writeback=True, protocol=2)
             elif typ == "memcached":
                 import memcache
-
                 idb = memcache.Client(addr)
             elif typ == "dict":  # in-memory dictionary
                 idb = {}
             elif typ == "mongodb":
-                from saml2.mongo_store import IdentMDB
-
                 self.ident = IdentMDB(database=addr, collection="ident")
 
             elif typ == "identdb":
@@ -181,8 +179,6 @@ def init_config(self, stype="idp"):
                 if typ == "shelve":
                     self.eptid = EptidShelve(secret, addr)
                 elif typ == "mongodb":
-                    from saml2.mongo_store import EptidMDB
-
                     self.eptid = EptidMDB(secret, database=addr,
                                           collection="eptid")
                 else:

tests/test_51_client.py

@@ -53,6 +53,8 @@
     "authn_auth": "http://www.example.com/login"
 }
 
+encode_fn = getattr(base64, 'encodebytes', base64.encodestring)
+
 
 def generate_cert():
     sn = uuid.uuid4().urn
@@ -177,7 +179,7 @@ def test_create_attribute_query1(self):
             "E8042FB4-4D5B-48C3-8E14-8EDD852790DD",
             format=saml.NAMEID_FORMAT_PERSISTENT,
             message_id="id1")
-        reqstr = "%s" % req.to_string().decode('utf-8')
+        reqstr = "%s" % req.to_string().decode()
 
         assert req.destination == "https://idp.example.com/idp/"
         assert req.id == "id1"
@@ -411,7 +413,7 @@ def test_response_1(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = self.client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -455,7 +457,7 @@ def test_response_1(self):
             userid="[email protected]",
             authn=AUTHN)
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         self.client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -505,7 +507,7 @@ def test_response_2(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -540,7 +542,7 @@ def test_response_3(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -575,7 +577,7 @@ def test_response_4(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -619,7 +621,7 @@ def test_response_5(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -672,7 +674,7 @@ def test_response_6(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -708,7 +710,7 @@ def test_response_7(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -751,7 +753,7 @@ def test_response_8(self):
 
         resp_str = "%s" % resp
 
-        resp_str = base64.encodestring(resp_str.encode('utf-8'))
+        resp_str = encode_fn(resp_str.encode())
 
         authn_response = _client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -927,7 +929,7 @@ def test_sign_then_encrypt_assertion2(self):
 
         # seresp = samlp.response_from_string(enctext)
 
-        resp_str = base64.encodestring(enctext.encode('utf-8'))
+        resp_str = encode_fn(enctext.encode())
         # Now over to the client side
         # Explicitely allow unsigned responses for this and the following 2 tests
         self.client.want_response_signed = False
@@ -1030,7 +1032,7 @@ def test_sign_then_encrypt_assertion_advice_1(self):
 
         # seresp = samlp.response_from_string(enctext)
 
-        resp_str = base64.encodestring(enctext.encode('utf-8'))
+        resp_str = encode_fn(enctext.encode())
         # Now over to the client side
         resp = self.client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,
@@ -1315,7 +1317,7 @@ def test_sign_then_encrypt_assertion_advice_2(self):
 
         # seresp = samlp.response_from_string(enctext)
 
-        resp_str = base64.encodestring(str(response).encode('utf-8'))
+        resp_str = encode_fn(str(response).encode())
         # Now over to the client side
         resp = self.client.parse_authn_request_response(
             resp_str, BINDING_HTTP_POST,