fixed kill_chains, granular_markings, extension-definitions

stix2/datastore/relational_db/add_method.py

@@ -1,7 +1,8 @@
 import re
-from stix2.v21.base import _STIXBase21
+
 from stix2.datastore.relational_db.utils import get_all_subclasses
 from stix2.properties import Property
+from stix2.v21.base import _STIXBase21
 
 _ALLOWABLE_CLASSES = get_all_subclasses(_STIXBase21)
 _ALLOWABLE_CLASSES.extend(get_all_subclasses(Property))

stix2/datastore/relational_db/input_creation.py

@@ -11,6 +11,8 @@
     HashesProperty, HexProperty, IDProperty, IntegerProperty, ListProperty,
     Property, ReferenceProperty, StringProperty, TimestampProperty,
 )
+from stix2.utils import STIXdatetime
+from stix2.v21.common import KillChainPhase
 
 
 @add_method(Property)
@@ -29,7 +31,7 @@ def generate_insert_information(self, name, stix_object, **kwargs):  # noqa: F81
 
 
 @add_method(DictionaryProperty)
-def generate_insert_information(self, name, stix_object, **kwargs):  # noqa: F811
+def generate_insert_information(self, dictionary_name, stix_object, **kwargs):  # noqa: F811
     bindings = dict()
     data_sink = kwargs.get("data_sink")
     table_name = kwargs.get("table_name")
@@ -42,21 +44,29 @@ def generate_insert_information(self, name, stix_object, **kwargs):  # noqa: F81
         bindings["id"] = foreign_key_value
     table = data_sink.tables_dictionary[
         canonicalize_table_name(
-            table_name + "_" + name,
+            table_name + "_" + dictionary_name,
             schema_name,
         )
     ]
-    for name, value in stix_object[name].items():
 
-        name_binding = "name"
-        if not hasattr(self, "value_types") or len(self.value_types) == 1:
+    # binary, boolean, float, hex,
+    # integer, string, timestamp
+    valid_types = stix_object._properties[dictionary_name].valid_types
+    for name, value in stix_object[dictionary_name].items():
+        if not valid_types or len(self.valid_types) == 1:
             value_binding = "value"
-        elif isinstance(value, int):
+        elif isinstance(value, int) and IntegerProperty in valid_types:
             value_binding = "integer_value"
+        elif isinstance(value, str) and StringProperty in valid_types:
+            value_binding = "string_value"
+        elif isinstance(value, bool) and BooleanProperty in valid_types:
+            value_binding = "boolean_value"
+        elif isinstance(value, STIXdatetime) and TimestampProperty in valid_types:
+            value_binding = "timestamp_value"
         else:
             value_binding = "string_value"
 
-        bindings[name_binding] = name
+        bindings["name"] = name
         bindings[value_binding] = value
 
         insert_statements.append(insert(table).values(bindings))
@@ -175,6 +185,29 @@ def generate_insert_information(   # noqa: F811
             }
             insert_statements.append(insert(table).values(bindings))
         return insert_statements
+    elif self.contained == KillChainPhase:
+        insert_statements = list()
+        table = data_sink.tables_dictionary[canonicalize_table_name(table_name + "_" + name, schema_name)]
+
+        for idx, item in enumerate(stix_object[name]):
+            bindings = {
+                "id": stix_object["id"] if id in stix_object else foreign_key_value,
+                "kill_chain_name": item["kill_chain_name"],
+                "phase_name": item["phase_name"],
+            }
+            insert_statements.append(insert(table).values(bindings))
+        return insert_statements
+    elif isinstance(self.contained, EnumProperty):
+        insert_statements = list()
+        table = data_sink.tables_dictionary[canonicalize_table_name(table_name + "_" + name, schema_name)]
+
+        for idx, item in enumerate(stix_object[name]):
+            bindings = {
+                "id": stix_object["id"] if id in stix_object else foreign_key_value,
+                name: item,
+            }
+            insert_statements.append(insert(table).values(bindings))
+        return insert_statements
     elif isinstance(self.contained, EmbeddedObjectProperty):
         insert_statements = list()
         for value in stix_object[name]:
@@ -265,20 +298,19 @@ def generate_insert_for_external_references(data_sink, stix_object):
 
 
 def generate_insert_for_granular_markings(granular_markings_table, stix_object):
+    insert_statements = list()
     granular_markings = stix_object["granular_markings"]
-    bindings = {
-        "id": stix_object["id"],
-    }
     for idx, granular_marking in enumerate(granular_markings):
-        lang_binding_name = f"lang{idx}"
-        marking_ref_binding_name = f"marking_ref{idx}"
-        selectors_binding_name = f"selectors{idx}"
-
-        bindings[lang_binding_name] = granular_marking.get("lang")
-        bindings[marking_ref_binding_name] = granular_marking.get("marking_ref")
-        bindings[selectors_binding_name] = granular_marking.get("selectors")
-
-    return [insert(granular_markings_table).values(bindings)]
+        bindings = {"id": stix_object["id"]}
+        lang_property_value = granular_marking.get("lang")
+        if lang_property_value:
+            bindings["lang"] = lang_property_value
+        marking_ref_value = granular_marking.get("marking_ref")
+        if marking_ref_value:
+            bindings["marking_ref"] = marking_ref_value
+        bindings["selectors"] = granular_marking.get("selectors")
+        insert_statements.append(insert(granular_markings_table).values(bindings))
+    return insert_statements
 
 
 # def generate_insert_for_extensions(extensions, foreign_key_value, type_name, core_properties):
@@ -297,7 +329,7 @@ def generate_insert_for_granular_markings(granular_markings_table, stix_object):
 
 
 def generate_insert_for_core(data_sink, stix_object, core_properties, schema_name):
-    if schema_name == "sdo":
+    if schema_name in ["sdo", "sro"]:
         core_table = data_sink.tables_dictionary["common.core_sdo"]
     else:
         core_table = data_sink.tables_dictionary["common.core_sco"]
@@ -335,7 +367,7 @@ def generate_insert_for_core(data_sink, stix_object, core_properties, schema_nam
             granular_marking_table = data_sink.tables_dictionary["common.granular_marking_sco"]
         granular_input_statements = generate_insert_for_granular_markings(
             granular_marking_table,
-            stix_object.granular_markings,
+            stix_object,
         )
         insert_statements.extend(granular_input_statements)
 
@@ -393,17 +425,20 @@ def generate_insert_for_object(data_sink, stix_object, schema_name, level=0):
     bindings = dict()
     if schema_name == "sco":
         core_properties = SCO_COMMON_PROPERTIES
-    else:
+    elif schema_name in ["sdo", "sro"]:
         core_properties = SDO_COMMON_PROPERTIES
+    else:
+        core_properties = list()
     type_name = stix_object["type"]
-    insert_statements.extend(generate_insert_for_core(data_sink, stix_object, core_properties, schema_name))
+    if core_properties:
+        insert_statements.extend(generate_insert_for_core(data_sink, stix_object, core_properties, schema_name))
     if "id" in stix_object:
         foreign_key_value = stix_object["id"]
     else:
         foreign_key_value = None
     sub_insert_statements = list()
     for name, prop in stix_object._properties.items():
-        if (name == 'id' or name not in core_properties) and name in stix_object:
+        if (name == 'id' or name not in core_properties) and name != "type" and name in stix_object:
             result = prop.generate_insert_information(
                 name, stix_object,
                 data_sink=data_sink,

stix2/datastore/relational_db/relational_db.py

@@ -2,7 +2,9 @@
 from sqlalchemy.schema import CreateSchema, CreateTable, Sequence
 from sqlalchemy_utils import create_database, database_exists, drop_database
 
-from stix2.base import _STIXBase
+from stix2.base import (
+    _DomainObject, _MetaObject, _Observable, _RelationshipObject, _STIXBase,
+)
 from stix2.datastore import DataSink, DataSource, DataStoreMixin
 from stix2.datastore.relational_db.input_creation import (
     generate_insert_for_object,
@@ -187,8 +189,19 @@ def add(self, stix_data, version=None):
         _add(self, stix_data)
     add.__doc__ = _add.__doc__
 
+    @staticmethod
+    def _determine_schema_name(stix_object):
+        if isinstance(stix_object, _DomainObject):
+            return "sdo"
+        elif isinstance(stix_object, _Observable):
+            return "sco"
+        elif isinstance(stix_object, _RelationshipObject):
+            return "sro"
+        elif isinstance(stix_object, _MetaObject):
+            return "common"
+
     def insert_object(self, stix_object):
-        schema_name = "sdo" if "created" in stix_object else "sco"
+        schema_name = self._determine_schema_name(stix_object)
         with self.database_connection.begin() as trans:
             statements = generate_insert_for_object(self, stix_object, schema_name)
             for stmt in statements:

stix2/datastore/relational_db/relational_db_testing.py

@@ -77,6 +77,7 @@ def malware_with_all_required_properties():
 def file_example_with_PDFExt_Object():
     f = stix2.v21.File(
         name="qwerty.dll",
+        magic_number_hex="504B0304",
         extensions={
             "pdf-ext": stix2.v21.PDFExt(
                 version="1.7",
@@ -95,22 +96,94 @@ def file_example_with_PDFExt_Object():
     return f
 
 
+def extension_definition_insert():
+    return stix2.ExtensionDefinition(
+        created_by_ref="identity--8a5fb7e4-aabe-4635-8972-cbcde1fa4792",
+        name="test",
+        schema="a schema",
+        version="1.2.3",
+        extension_types=["property-extension", "new-sdo", "new-sro"],
+    )
+
+
+def dictionary_test():
+    return stix2.File(
+        spec_version="2.1",
+        name="picture.jpg",
+        defanged=True,
+        ctime="1980-02-23T05:43:28.2678Z",
+        extensions={
+          "raster-image-ext": {
+              "exif_tags": {
+                  "Make": "Nikon",
+                  "Model": "D7000",
+                  "XResolution": 4928,
+                  "YResolution": 3264,
+              },
+          },
+        },
+    )
+
+
+def kill_chain_test():
+    return stix2.AttackPattern(
+        spec_version="2.1",
+        id="attack-pattern--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061",
+        created="2016-05-12T08:17:27.000Z",
+        modified="2016-05-12T08:17:27.000Z",
+        name="Spear Phishing",
+        kill_chain_phases=[
+             {
+                 "kill_chain_name": "lockheed-martin-cyber-kill-chain",
+                 "phase_name": "reconnaissance",
+             },
+        ],
+        external_references=[
+             {
+                 "source_name": "capec",
+                 "external_id": "CAPEC-163",
+             },
+        ],
+        granular_markings=[
+             {
+                 "lang": "en_US",
+                 "selectors": ["kill_chain_phases"],
+             },
+             {
+                 "marking_ref": "marking-definition--50902d70-37ae-4f85-af68-3f4095493b42",
+                 "selectors": ["external_references"],
+             },
+        ], )
+
+
 def main():
     store = RelationalDBStore(
         "postgresql://localhost/stix-data-sink",
         False,
         None,
         True,
         False,
-        stix2.Directory,
     )
 
     if store.sink.database_exists:
         store.sink.generate_stix_schema()
         store.sink.clear_tables()
 
+        pdf_file = file_example_with_PDFExt_Object()
+        store.add(pdf_file)
+
+        ap = kill_chain_test()
+        store.add(ap)
+
         store.add(directory_stix_object)
 
+        store.add(s)
+
+        store.add(extension_definition_insert())
+
+        dict_example = dictionary_test()
+        store.add(dict_example)
+
         read_obj = store.get(directory_stix_object.id)
         print(read_obj)
     else:

stix2/datastore/relational_db/table_creation.py

@@ -587,7 +587,7 @@ def generate_table_information(self, name, metadata, schema_name, table_name, **
         )
         return tables
     elif self.contained == KillChainPhase:
-        tables.append(create_kill_chain_phases_table("kill_chain_phase", metadata, schema_name, table_name))
+        tables.append(create_kill_chain_phases_table("kill_chain_phases", metadata, schema_name, table_name))
         return tables
     else:
         if isinstance(self.contained, Property):
@@ -693,7 +693,8 @@ def generate_object_table(
     columns = list()
     tables = list()
     for name, prop in properties.items():
-        if name == 'id' or name not in core_properties:
+        # type is never a column since it is implicit in the table
+        if (name == 'id' or name not in core_properties) and name != 'type':
             col = prop.generate_table_information(
                 name,
                 metadata=metadata,