Merge pull request #240 from MagsenAbbeThales/master

fix: spec_version & version from modified/created

Author: eric-eclecticiq

opentaxii/persistence/sqldb/api.py

@@ -14,7 +14,7 @@
 from opentaxii.persistence import OpenTAXII2PersistenceAPI, OpenTAXIIPersistenceAPI
 from opentaxii.persistence.sqldb import taxii2models
 from opentaxii.taxii2 import entities
-from opentaxii.taxii2.utils import DATETIMEFORMAT
+from opentaxii.taxii2.utils import get_object_version
 
 from . import converters as conv
 from .models import (
@@ -1005,9 +1005,7 @@ def add_objects(
         self.db.session.commit()
         job_details = []
         for obj in objects:
-            version = datetime.datetime.strptime(
-                obj["modified"], DATETIMEFORMAT
-            ).replace(tzinfo=datetime.timezone.utc)
+            version = get_object_version(obj)
             if (
                 not self.db.session.query(literal(True))
                 .filter(

opentaxii/taxii2/utils.py

@@ -15,3 +15,33 @@ def taxii2_datetimeformat(input_value: datetime.datetime) -> str:
     :rtype: string
     """
     return input_value.astimezone(datetime.timezone.utc).strftime(DATETIMEFORMAT)
+
+
+def get_object_version(obj: dict) -> datetime.datetime:
+    """
+    The TAXII version is inferred in this order:
+
+    1. ``modified`` field, if present
+    2. ``created`` field, if present
+    3. 1970-01-01T00:00:00+00:00 otherwise
+
+    This is guided by 3.4.1 Supported Fields for Match [#]_:
+
+        If a STIX object is not versioned (and therefore does not have a modified
+        timestamp) then this version parameter MUST use the created timestamp. If
+        an object does not have a created or modified timestamp or any other
+        version information that can be used, then the server should use a value for
+        the version that is consistent to the server.
+
+    .. [#] https://docs.oasis-open.org/cti/taxii/v2.1/os/taxii-v2.1-os.html#_Toc31107518
+    """
+    if "modified" in obj:
+        return datetime.datetime.strptime(obj["modified"], DATETIMEFORMAT).replace(
+            tzinfo=datetime.timezone.utc
+        )
+    elif "created" in obj:
+        return datetime.datetime.strptime(obj["created"], DATETIMEFORMAT).replace(
+            tzinfo=datetime.timezone.utc
+        )
+    else:
+        return datetime.datetime.fromtimestamp(0, tz=datetime.timezone.utc)

tests/taxii2/test_taxii2_sqldb.py

@@ -6,7 +6,7 @@
 from opentaxii.persistence.sqldb.api import Taxii2SQLDatabaseAPI
 from opentaxii.persistence.sqldb.taxii2models import Job, JobDetail, STIXObject
 from opentaxii.taxii2 import entities
-from opentaxii.taxii2.utils import DATETIMEFORMAT
+from opentaxii.taxii2.utils import get_object_version
 from tests.taxii2.utils import (
     API_ROOTS,
     API_ROOTS_WITH_DEFAULT,
@@ -794,7 +794,37 @@ def test_get_objects(
                     "valid_from": "2016-01-01T00:00:00Z",
                 }
             ],  # objects
-            id="single object",
+            id="single-object-version-modified",
+        ),
+        pytest.param(
+            API_ROOTS[0].id,  # api_root_id
+            COLLECTIONS[5].id,  # collection_id
+            [
+                {
+                    "definition": {
+                        "statement": "Copyright 2015-2025, The MITRE Corporation. [...]"
+                    },
+                    "id": "marking-definition--fa42a846-8d90-4e51-bc29-71d5b4802168",
+                    "type": "marking-definition",
+                    "created": "2017-06-01T00:00:00.000Z",
+                    "created_by_ref": "identity--c78cb6e5-0c4b-4611-8297-d1b8b55e40b5",
+                    "definition_type": "statement",
+                    "spec_version": "2.1",
+                }
+            ],  # objects
+            id="single-object-version-created",
+        ),
+        pytest.param(
+            API_ROOTS[0].id,  # api_root_id
+            COLLECTIONS[5].id,  # collection_id
+            [
+                {
+                    "type": "x-no-version",
+                    "id": "x-no-version--5e113376-8a13-432d-b711-92f566ebbd92",
+                    "spec_version": "2.1",
+                }
+            ],  # objects
+            id="single-object-version-created",
         ),
         pytest.param(
             API_ROOTS[0].id,  # api_root_id
@@ -869,9 +899,7 @@ def test_add_objects(
                     id=job_detail.id,
                     job_id=job.id,
                     stix_id=obj["id"],
-                    version=datetime.datetime.strptime(
-                        obj["modified"], DATETIMEFORMAT
-                    ).replace(tzinfo=datetime.timezone.utc),
+                    version=get_object_version(obj),
                     message="",
                     status="success",
                 )
@@ -894,10 +922,7 @@ def test_add_objects(
             taxii2_sqldb_api.db.session.query(STIXObject)
             .filter(
                 STIXObject.id == obj["id"],
-                STIXObject.version
-                == datetime.datetime.strptime(obj["modified"], DATETIMEFORMAT).replace(
-                    tzinfo=datetime.timezone.utc
-                ),
+                STIXObject.version == get_object_version(obj),
             )
             .one()
         )
@@ -906,9 +931,7 @@ def test_add_objects(
         assert db_obj.type == obj["type"]
         assert db_obj.spec_version == obj["spec_version"]
         assert isinstance(db_obj.date_added, datetime.datetime)
-        assert db_obj.version == datetime.datetime.strptime(
-            obj["modified"], DATETIMEFORMAT
-        ).replace(tzinfo=datetime.timezone.utc)
+        assert db_obj.version == get_object_version(obj)
         assert db_obj.serialized_data == {
             key: value
             for (key, value) in obj.items()
@@ -921,9 +944,7 @@ def test_add_objects(
         )
         assert db_job_detail.job_id == db_job.id
         assert db_job_detail.stix_id == obj["id"]
-        assert db_job_detail.version == datetime.datetime.strptime(
-            obj["modified"], DATETIMEFORMAT
-        ).replace(tzinfo=datetime.timezone.utc)
+        assert db_job_detail.version == get_object_version(obj)
         assert db_job_detail.message == ""
         assert db_job_detail.status == "success"
 

tests/taxii2/test_taxii2_utils.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from opentaxii.taxii2.utils import taxii2_datetimeformat
+from opentaxii.taxii2.utils import get_object_version, taxii2_datetimeformat
 
 
 @pytest.mark.parametrize(
@@ -73,3 +73,18 @@
 )
 def test_taxii2_datetimeformat(input_value, expected):
     assert taxii2_datetimeformat(input_value) == expected
+
+
+def test_get_object_version():
+    assert get_object_version(
+        {
+            "modified": "2022-01-01T13:30:00.000000Z",
+            "created": "2016-04-06T20:06:37.000Z",
+        }
+    ) == datetime.datetime(2022, 1, 1, 13, 30, 00, 0, datetime.timezone.utc)
+    assert get_object_version(
+        {"created": "2016-04-06T20:06:37.000Z"}
+    ) == datetime.datetime(2016, 4, 6, 20, 6, 37, 0, datetime.timezone.utc)
+    assert get_object_version({}) == datetime.datetime.fromtimestamp(
+        0, tz=datetime.timezone.utc
+    )