Python 3.14 support

docs/conf.py

@@ -39,7 +39,6 @@
 if not on_rtd:
     import sphinx_rtd_theme
     html_theme = 'sphinx_rtd_theme'
-    html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
 else:
     html_theme = 'default'
 

libtaxii/__init__.py

@@ -5,16 +5,12 @@
 The main libtaxii module
 """
 
-import six
-from six.moves import urllib
+from urllib.error import HTTPError
 
 import libtaxii.messages_10 as tm10
 import libtaxii.messages_11 as tm11
-import libtaxii.clients as tc
 from .constants import *
 
-import cgi
-
 from .version import __version__  # noqa
 
 
@@ -39,70 +35,23 @@ def get_message_from_http_response(http_response, in_response_to):
             parse
         in_response_to (str): the default value for in_response_to
     """
-    if isinstance(http_response, six.moves.http_client.HTTPResponse):
-        return get_message_from_httplib_http_response(http_response, in_response_to)
-    elif isinstance(http_response, urllib.error.HTTPError):
-        return get_message_from_urllib2_httperror(http_response, in_response_to)
-    elif isinstance(http_response, urllib.response.addinfourl):
-        return get_message_from_urllib_addinfourl(http_response, in_response_to)
-    else:
-        raise ValueError('Unsupported response type: %s.' % http_response.__class__.__name__)
-
 
-def get_message_from_urllib2_httperror(http_response, in_response_to):
-    """ This function should not be called by libtaxii users directly. """
-    info = http_response.info()
-
-    if hasattr(info, 'getheader'):
-        taxii_content_type = info.getheader('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(info.getheader('Content-Type'))
-    else:
-        taxii_content_type = info.get('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(info.get('Content-Type'))
+    taxii_content_type = http_response.getheader('X-TAXII-Content-Type')
+    encoding = http_response.headers.get_content_charset() or 'utf-8'
 
-    encoding = params.get('charset', 'utf-8')
-    response_message = six.ensure_text(http_response.read(), errors='replace')
+    response_message = http_response.read()
 
     if taxii_content_type is None:
-        m = str(http_response) + '\r\n' + str(http_response.info()) + '\r\n' + response_message
-        return tm11.StatusMessage(message_id='0', in_response_to=in_response_to, status_type=ST_FAILURE, message=m)
-    elif taxii_content_type == VID_TAXII_XML_10:  # It's a TAXII XML 1.0 message
-        return tm10.get_message_from_xml(response_message, encoding)
-    elif taxii_content_type == VID_TAXII_XML_11:  # It's a TAXII XML 1.1 message
-        return tm11.get_message_from_xml(response_message, encoding)
-    elif taxii_content_type == VID_CERT_EU_JSON_10:
-        return tm10.get_message_from_json(response_message, encoding)
-    else:
-        raise ValueError('Unsupported X-TAXII-Content-Type: %s' % taxii_content_type)
-
-
-def get_message_from_urllib_addinfourl(http_response, in_response_to):
-    """ This function should not be called by libtaxii users directly. """
-    info = http_response.info()
-
-    if hasattr(info, 'getheader'):
-        taxii_content_type = info.getheader('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(info.getheader('Content-Type'))
-    else:
-        taxii_content_type = info.get('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(info.get('Content-Type'))
-
-    encoding = params.get('charset', 'utf-8')
-    response_message = six.ensure_text(http_response.read(), errors='replace')
-
-    if taxii_content_type is None:  # Treat it as a Failure Status Message, per the spec
-
-        message = []
-        header_dict = six.iteritems(http_response.info().dict)
-        for k, v in header_dict:
-            message.append(k + ': ' + v + '\r\n')
-        message.append('\r\n')
-        message.append(response_message)
-
-        m = ''.join(message)
+        if isinstance(http_response, HTTPError):
+            m = str(http_response) + '\r\n'
+        else:
+            m = ''
+        for header, value in http_response.headers.items():
+            m += f'{header}: {value}\r\n'
+        m += '\r\n'
+        m += response_message.decode(encoding, 'replace')
 
         return tm11.StatusMessage(message_id='0', in_response_to=in_response_to, status_type=ST_FAILURE, message=m)
-
     elif taxii_content_type == VID_TAXII_XML_10:  # It's a TAXII XML 1.0 message
         return tm10.get_message_from_xml(response_message, encoding)
     elif taxii_content_type == VID_TAXII_XML_11:  # It's a TAXII XML 1.1 message
@@ -111,36 +60,3 @@ def get_message_from_urllib_addinfourl(http_response, in_response_to):
         return tm10.get_message_from_json(response_message, encoding)
     else:
         raise ValueError('Unsupported X-TAXII-Content-Type: %s' % taxii_content_type)
-
-
-def get_message_from_httplib_http_response(http_response, in_response_to):
-    """ This function should not be called by libtaxii users directly. """
-    if hasattr(http_response, 'getheader'):
-        taxii_content_type = http_response.getheader('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(http_response.getheader('Content-Type'))
-    else:
-        taxii_content_type = http_response.get('X-TAXII-Content-Type')
-        _, params = cgi.parse_header(http_response.get('Content-Type'))
-
-    encoding = params.get('charset', 'utf-8')
-    response_message = six.ensure_text(http_response.read(), errors='replace')
-
-    if taxii_content_type is None:  # Treat it as a Failure Status Message, per the spec
-
-        message = []
-        header_tuples = http_response.getheaders()
-        for k, v in header_tuples:
-            message.append(k + ': ' + v + '\r\n')
-        message.append('\r\n')
-        message.append(response_message)
-
-        m = ''.join(message)
-
-        return tm11.StatusMessage(message_id='0', in_response_to=in_response_to, status_type=ST_FAILURE, message=m)
-
-    elif taxii_content_type == VID_TAXII_XML_10:  # It's a TAXII XML 1.0 message
-        return tm10.get_message_from_xml(response_message, encoding)
-    elif taxii_content_type == VID_TAXII_XML_11:  # It's a TAXII XML 1.1 message
-        return tm11.get_message_from_xml(response_message, encoding)
-    else:
-        raise ValueError('Unsupported X-TAXII-Content-Type: %s' % taxii_content_type)

libtaxii/scripts/__init__.py

@@ -9,7 +9,7 @@
 import datetime
 import libtaxii.clients as tc
 import six
-from six.moves.configparser import SafeConfigParser
+from configparser import ConfigParser
 from six.moves.urllib.parse import urlparse
 
 import libtaxii as t
@@ -71,7 +71,7 @@ def add_poll_response_args(parser):
                              "to \'clobber\'")
 
 
-class ArgParserConfig(SafeConfigParser):
+class ArgParserConfig(ConfigParser):
 
     def as_args(self, section, raw=False, vars=None):
         """
@@ -102,7 +102,7 @@ class LoadFromFile(argparse.Action):
     def __call__(self, parser, namespace, values, option_string=None):
         config_file = ArgParserConfig()
         with values as f:
-            config_file.readfp(f)
+            config_file.read_file(f)
 
         # Overrides initial values if present on config file
         parser.parse_known_args(config_file.as_args('libtaxii'), namespace=namespace)

libtaxii/test/messages_11_test.py

@@ -7,11 +7,13 @@
 
 import datetime
 import io
+import re
 import sys
 import unittest
 import warnings
 import inspect
 
+import pytest
 from dateutil.tz import tzutc
 from lxml import etree
 
@@ -1244,12 +1246,9 @@ def test_xee_remote(self):
                         <foo>&xxe;</foo>
                       """
 
-        # If an XML Syntax Error is received, an attack would have succeeded
-
-        try:
-            e = parse(xee_remote)
-        except etree.XMLSyntaxError:
-            raise ValueError("An XML Syntax Error was raised, meaning a real attack would have succeeded!")
+        # Undefined xxe entity should be expected, because lxml should not attempt to define it via external reference
+        with pytest.raises(etree.XMLSyntaxError, match=re.escape("Entity 'xxe' not defined, line 4, column 35 (<string>, line 4)")):
+             parse(xee_remote)
 
     def test_xee_local(self):
         """
@@ -1263,12 +1262,10 @@ def test_xee_local(self):
                           <!ENTITY xxe SYSTEM "file:///etc/passwd" >]>
                         <foo>&xxe;</foo>
                       """
-        # If an XML Syntax Error is received, an attack would have succeeded
 
-        try:
-            e = parse(xee_local)
-        except etree.XMLSyntaxError:
-            raise ValueError("An XML Syntax Error was raised, meaning a real attack would have succeeded!")
+        # Undefined xxe entity is expected, because lxml should not attempt to define it via external reference
+        with pytest.raises(etree.XMLSyntaxError, match=re.escape("Entity 'xxe' not defined, line 4, column 35 (<string>, line 4)")):
+             parse(xee_local)
 
     def test_ssrf(self):
         """

libtaxii/test/test_message_from_response.py

@@ -0,0 +1,126 @@
+from http.client import HTTPResponse
+from urllib.error import HTTPError
+
+import pytest
+
+import libtaxii
+from libtaxii.constants import VID_TAXII_XML_10, VID_TAXII_XML_11, VID_CERT_EU_JSON_10
+import libtaxii.clients
+import libtaxii.messages
+
+
+IN_RESPONSE_TO = "test in_response_to value"
+TAXII_CONTENT_TYPE = libtaxii.constants.VID_TAXII_XML_11
+TEST_MESSAGE_ID = "test message id"
+
+MESSAGES = {
+    VID_TAXII_XML_10: (
+        '<taxii:Discovery_Request xmlns:taxii="http://taxii.mitre.org/messages/taxii_xml_binding-1" message_id="1"/>',
+        libtaxii.messages_10.DiscoveryRequest(message_id='1')
+    ),
+    VID_TAXII_XML_11: (
+        '<taxii_11:Discovery_Request xmlns:taxii_11="http://taxii.mitre.org/messages/taxii_xml_binding-1.1" message_id="non-ascii message id Ħ"/>',
+        libtaxii.messages_11.DiscoveryRequest(message_id='non-ascii message id Ħ'),
+    ),
+    VID_CERT_EU_JSON_10: (
+        '{"extended_headers": {}, "message_type": "Discovery_Request", "message_id": "1"}',
+        libtaxii.messages_10.DiscoveryRequest(message_id='1')
+    )
+}
+
+
+@pytest.mark.parametrize('taxii_content_type', MESSAGES.keys())
+def test_httplib_http_response(taxii_content_type, httpserver):
+    content, expected_message = MESSAGES[taxii_content_type]
+
+    httpserver.expect_request("/poll_service_path/").respond_with_data(
+        content.encode('ISO-8859-3'),
+        content_type="application/xml; charset=ISO-8859-3",
+        headers={"X-TAXII-Content-Type": taxii_content_type}
+    )
+
+    client = libtaxii.clients.HttpClient()
+    http_response = client.call_taxii_service2(httpserver.host, '/poll_service_path/', libtaxii.constants.VID_TAXII_XML_10, b"", port=httpserver.port)
+    assert isinstance(http_response, HTTPResponse)
+
+    message = libtaxii.get_message_from_http_response(http_response, IN_RESPONSE_TO)
+    print(message.message_id)
+    assert message == expected_message
+
+
+@pytest.mark.parametrize('taxii_content_type', MESSAGES.keys())
+def test_httplib_http_response_error(taxii_content_type, httpserver):
+    content, expected_message = MESSAGES[taxii_content_type]
+
+    httpserver.expect_request("/poll_service_path/").respond_with_data(
+        content.encode('ISO-8859-3'),
+        status=500,
+        content_type="application/xml; charset=ISO-8859-3",
+        headers={"X-TAXII-Content-Type": taxii_content_type}
+    )
+
+    client = libtaxii.clients.HttpClient()
+    http_response = client.call_taxii_service2(httpserver.host, '/poll_service_path/', libtaxii.constants.VID_TAXII_XML_10, b"", port=httpserver.port)
+    assert isinstance(http_response, HTTPError)
+
+    message = libtaxii.get_message_from_http_response(http_response, IN_RESPONSE_TO)
+    assert message == expected_message
+
+
+def test_httplib_http_response_no_taxii_content_type(httpserver):
+    httpserver.expect_request("/poll_service_path/").respond_with_data(
+        "some content".encode('ISO-8859-3'),
+        content_type="application/xml; charset=ISO-8859-3",
+    )
+
+    client = libtaxii.clients.HttpClient()
+    http_response = client.call_taxii_service2(httpserver.host, '/poll_service_path/', libtaxii.constants.VID_TAXII_XML_10, b"", port=httpserver.port)
+    message = libtaxii.get_message_from_http_response(http_response, IN_RESPONSE_TO)
+
+    assert isinstance(message, libtaxii.messages_11.StatusMessage)
+    assert message.in_response_to == IN_RESPONSE_TO
+    assert message.message_id == '0'
+
+    server = http_response.headers.get("Server")
+    date = http_response.headers.get("Date")
+    assert message.message == (
+        f'''Server: {server}\r\n'''
+        f'''Date: {date}\r\n'''
+        '''Content-Type: application/xml; charset=ISO-8859-3\r\n'''
+        '''Content-Length: 12\r\n'''
+        '''Connection: close\r\n'''
+        '''\r\n'''
+        '''some content'''
+    )
+
+
+def test_httplib_http_response_error_no_taxii_content_type(httpserver):
+    httpserver.expect_request("/poll_service_path/").respond_with_data(
+        "some content".encode('ISO-8859-3'),
+        status=500,
+        content_type="application/xml; charset=ISO-8859-3",
+    )
+
+    client = libtaxii.clients.HttpClient()
+    http_response = client.call_taxii_service2(httpserver.host, '/poll_service_path/', libtaxii.constants.VID_TAXII_XML_10, b"", port=httpserver.port)
+    assert isinstance(http_response, HTTPError)
+
+    message = libtaxii.get_message_from_http_response(http_response, IN_RESPONSE_TO)
+    assert isinstance(message, libtaxii.messages_11.StatusMessage)
+    assert message.in_response_to == IN_RESPONSE_TO
+    assert message.message_id == '0'
+
+    server = http_response.headers.get("Server")
+    date = http_response.headers.get("Date")
+    assert message.message == (
+        f'''HTTP Error 500: INTERNAL SERVER ERROR\r\n'''
+        f'''Server: {server}\r\n'''
+        f'''Date: {date}\r\n'''
+        '''Content-Type: application/xml; charset=ISO-8859-3\r\n'''
+        '''Content-Length: 12\r\n'''
+        '''Connection: close\r\n'''
+        '''\r\n'''
+        '''some content'''
+    )
+
+

libtaxii/validation.py

@@ -19,9 +19,9 @@
 
 RegexTuple = collections.namedtuple('_RegexTuple', ['regex', 'title'])
 # URI regex per http://tools.ietf.org/html/rfc3986
-uri_regex = RegexTuple("(([^:/?#]+):)?(//([^/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?", "URI Format")
-message_id_regex_10 = RegexTuple("^[0-9]+$", "Numbers only")
-targeting_expression_regex = RegexTuple("^(@?\w+|\*{1,2})(/(@?\w+|\*{1,2}))*$", "Targeting Expression Syntax")
+uri_regex = RegexTuple(r"(([^:/?#]+):)?(//([^/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?", "URI Format")
+message_id_regex_10 = RegexTuple(r"^[0-9]+$", "Numbers only")
+targeting_expression_regex = RegexTuple(r"^(@?\w+|\*{1,2})(/(@?\w+|\*{1,2}))*$", "Targeting Expression Syntax")
 
 _none_error = "%s is not allowed to be None and the provided value was None"
 _type_error = "%s must be of type %s. The incorrect value was of type %s"

pyproject.toml

@@ -0,0 +1,2 @@
+[build-system]
+requires = ["setuptools"]

requirements.txt

@@ -1,7 +1,8 @@
-Sphinx==1.6.1
-sphinx_rtd_theme==0.2.4
-pytest==3.0.7
-tox==2.7.0
-bumpversion==0.5.3
+Sphinx==8.1.3
+sphinx_rtd_theme==3.1.0
+pytest==7.4.4
+pytest-httpserver==1.1.1
+tox==4.47.3
+bump-my-version==1.2.7
 
 -e  .

setup.py

@@ -91,7 +91,6 @@ def get_long_description():
     classifiers=[
         'Development Status :: 5 - Production/Stable',
         'Intended Audience :: Developers',
-        'License :: OSI Approved :: BSD License',
         'Operating System :: OS Independent',
         'Programming Language :: Python :: 2',
         'Programming Language :: Python :: 2.6',