diff --git a/docs/conf.py b/docs/conf.py
index e9ff1a2..b2efe69 100644
--- a/docs/conf.py
+++ b/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'
diff --git a/libtaxii/__init__.py b/libtaxii/__init__.py
index 6d4268d..bea7dd4 100644
--- a/libtaxii/__init__.py
+++ b/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)
diff --git a/libtaxii/scripts/__init__.py b/libtaxii/scripts/__init__.py
index c6574fa..76facf7 100644
--- a/libtaxii/scripts/__init__.py
+++ b/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)
diff --git a/libtaxii/test/messages_11_test.py b/libtaxii/test/messages_11_test.py
index 96e33a1..97284b6 100644
--- a/libtaxii/test/messages_11_test.py
+++ b/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):
&xxe;
"""
- # 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 (, line 4)")):
+ parse(xee_remote)
def test_xee_local(self):
"""
@@ -1263,12 +1262,10 @@ def test_xee_local(self):
]>
&xxe;
"""
- # 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 (, line 4)")):
+ parse(xee_local)
def test_ssrf(self):
"""
diff --git a/libtaxii/test/test_message_from_response.py b/libtaxii/test/test_message_from_response.py
new file mode 100644
index 0000000..f280d5c
--- /dev/null
+++ b/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: (
+ '',
+ libtaxii.messages_10.DiscoveryRequest(message_id='1')
+ ),
+ VID_TAXII_XML_11: (
+ '',
+ 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'''
+ )
+
+
diff --git a/libtaxii/validation.py b/libtaxii/validation.py
index 8440751..1569cca 100644
--- a/libtaxii/validation.py
+++ b/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"
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..b1e1a46
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,2 @@
+[build-system]
+requires = ["setuptools"]
diff --git a/requirements.txt b/requirements.txt
index 28065c1..4988234 100644
--- a/requirements.txt
+++ b/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 .
diff --git a/setup.py b/setup.py
index 22b7e5c..92e59b4 100644
--- a/setup.py
+++ b/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',
diff --git a/tox.ini b/tox.ini
index 9e41056..b324447 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
[tox]
-envlist = py26, py27, rhel26, py34, py35, py36, py37, py38, docs, packaging
+envlist = 3.10, 3.11, 3.12, 3.13, 3.14, docs, packaging
[testenv]
commands =
@@ -8,20 +8,6 @@ commands =
sphinx-build -b html docs docs/_build/html
deps = -rrequirements.txt
-[testenv:py26]
-commands =
- pytest libtaxii
-deps = pytest
-
-[testenv:rhel26]
-commands =
- pytest libtaxii
-deps =
- lxml==2.2.3
- python-dateutil==1.4.1
- six==1.9.0
- pytest
-
[testenv:docs]
commands =
sphinx-build -b doctest docs docs/_build/doctest
@@ -30,15 +16,14 @@ commands =
[testenv:packaging]
deps =
readme_renderer
+ setuptools
commands =
python setup.py check -r -s
[travis]
python =
- 2.6: py26, rhel26
- 2.7: py27, docs, packaging
- 3.4: py34
- 3.5: py35
- 3.6: py36, docs, packaging
- 3.7: py37
- 3.8: py38
+ 3.10: 3.10
+ 3.11: 3.11
+ 3.12: 3.12
+ 3.13: 3.13
+ 3.14: 3.14, docs, packaging