Handle hex

.github/workflows/python-ci-tests.yml

@@ -3,10 +3,31 @@
 name: cti-python-stix2 test harness
 on: [push, pull_request]
 
-jobs:
-  build:
+env:
+  POSTGRES_USER: postgres
+  POSTGRES_PASSWORD: postgres
+  POSTGRES_DB: postgres
 
+jobs:
+  test-job:
     runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:11
+        # Provide the password for postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: postgres
+        ports: [ '5432:5432' ]
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     strategy:
       matrix:
         python-version: [3.8, 3.9, '3.10', '3.11', '3.12']

.pre-commit-config.yaml

@@ -1,22 +1,22 @@
 repos:
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v3.4.0
+    rev: v4.5.0
     hooks:
     -   id: trailing-whitespace
     -   id: check-merge-conflict
 -   repo: https://github.com/asottile/add-trailing-comma
-    rev: v2.0.2
+    rev: v3.1.0
     hooks:
     -   id: add-trailing-comma
 -   repo: https://github.com/PyCQA/flake8
-    rev: 3.8.4
+    rev: 7.0.0
     hooks:
     -   id: flake8
         name: Check project styling
         args:
         - --max-line-length=160
 -   repo: https://github.com/PyCQA/isort
-    rev: 5.12.0
+    rev: 5.13.2
     hooks:
     -   id: isort
         name: Sort python imports (shows diff)

setup.py

@@ -52,6 +52,7 @@ def get_long_description():
         'requests',
         'simplejson',
         'stix2-patterns>=1.2.0',
+        'inflection',
     ],
     project_urls={
         'Documentation': 'https://stix2.readthedocs.io/',
@@ -61,5 +62,10 @@ def get_long_description():
     extras_require={
         'taxii': ['taxii2-client>=2.3.0'],
         'semantic': ['haversine', 'rapidfuzz'],
+        'relationaldb': [
+            'sqlalchemy',
+            'sqlalchemy_utils',
+            'psycopg2',
+        ],
     },
 )

stix2/base.py

@@ -470,6 +470,10 @@ def _check_object_constraints(self):
         self._check_at_least_one_property()
 
 
+class _MetaObject(_STIXBase):
+    pass
+
+
 def _choose_one_hash(hash_dict):
     if "MD5" in hash_dict:
         return {"MD5": hash_dict["MD5"]}

stix2/custom.py

@@ -89,7 +89,7 @@ def __init__(self, **kwargs):
     return _CustomObservable
 
 
-def _custom_extension_builder(cls, type, properties, version, base_class):
+def _custom_extension_builder(cls, applies_to, type, properties, version, base_class):
 
     properties = _get_properties_dict(properties)
     toplevel_properties = None
@@ -98,6 +98,7 @@ def _custom_extension_builder(cls, type, properties, version, base_class):
     # it exists.  How to treat the other properties which were given depends on
     # the extension type.
     extension_type = getattr(cls, "extension_type", None)
+    applies_to = applies_to
     if extension_type:
         # I suppose I could also go with a plain string property, since the
         # value is fixed... but an enum property seems more true to the
@@ -128,6 +129,7 @@ class _CustomExtension(cls, base_class):
 
         _type = type
         _properties = nested_properties
+        _applies_to = applies_to
         if extension_type == "toplevel-property-extension":
             _toplevel_properties = toplevel_properties
 

stix2/datastore/relational_db/add_method.py

@@ -0,0 +1,28 @@
+import re
+
+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))
+_ALLOWABLE_CLASSES.extend([Property])
+
+
+def create_real_method_name(name, klass_name):
+    classnames = map(lambda x: x.__name__, _ALLOWABLE_CLASSES)
+    if klass_name not in classnames:
+        raise NameError
+
+    split_up_klass_name = re.findall('[A-Z][^A-Z]*', klass_name)
+    return name + "_" + "_".join([x.lower() for x in split_up_klass_name])
+
+
+def add_method(cls):
+
+    def decorator(fn):
+        method_name = fn.__name__
+        fn.__name__ = create_real_method_name(fn.__name__, cls.__name__)
+        setattr(cls, method_name, fn)
+        return fn
+    return decorator

stix2/datastore/relational_db/database_backends/database_backend_base.py

@@ -0,0 +1,114 @@
+from typing import Any
+
+from sqlalchemy import Boolean, Float, Integer, Text, create_engine
+from sqlalchemy_utils import create_database, database_exists, drop_database
+
+from stix2.base import (
+    _DomainObject, _MetaObject, _Observable, _RelationshipObject,
+)
+
+
+class DatabaseBackend:
+    def __init__(self, database_connection_url, force_recreate=False, **kwargs: Any):
+        self.database_connection_url = database_connection_url
+        self.database_exists = database_exists(database_connection_url)
+
+        if force_recreate:
+            if self.database_exists:
+                drop_database(database_connection_url)
+            create_database(database_connection_url)
+            self.database_exists = database_exists(database_connection_url)
+
+        self.database_connection = create_engine(database_connection_url)
+
+    def _create_schemas(self):
+        pass
+
+    @staticmethod
+    def determine_schema_name(stix_object):
+        return ""
+
+    @staticmethod
+    def determine_stix_type(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 _create_database(self):
+        if self.database_exists:
+            drop_database(self.database_connection.url)
+        create_database(self.database_connection.url)
+        self.database_exists = database_exists(self.database_connection.url)
+
+    def schema_for(stix_class):
+        return ""
+
+    @staticmethod
+    def schema_for_core():
+        return ""
+
+    # you must implement the next 4 methods in the subclass
+
+    @staticmethod
+    def determine_sql_type_for_property():  # noqa: F811
+        pass
+
+    @staticmethod
+    def determine_sql_type_for_binary_property():  # noqa: F811
+        pass
+
+    @staticmethod
+    def determine_sql_type_for_hex_property():  # noqa: F811
+        pass
+
+    @staticmethod
+    def determine_sql_type_for_timestamp_property():  # noqa: F811
+        pass
+
+    @staticmethod
+    def determine_sql_type_for_kill_chain_phase():  # noqa: F811
+        return None
+
+    @staticmethod
+    def determine_sql_type_for_boolean_property():  # noqa: F811
+        return Boolean
+
+    @staticmethod
+    def determine_sql_type_for_float_property():  # noqa: F811
+        return Float
+
+    @staticmethod
+    def determine_sql_type_for_integer_property():  # noqa: F811
+        return Integer
+
+    @staticmethod
+    def determine_sql_type_for_reference_property():  # noqa: F811
+        return Text
+
+    @staticmethod
+    def determine_sql_type_for_string_property():  # noqa: F811
+        return Text
+
+    @staticmethod
+    def determine_sql_type_for_key_as_int():  # noqa: F811
+        return Integer
+
+    @staticmethod
+    def determine_sql_type_for_key_as_id():  # noqa: F811
+        return Text
+
+    @staticmethod
+    def array_allowed():
+        return False
+
+    def generate_value(self, stix_type, value):
+        sql_type = stix_type.determine_sql_type(self)
+        if sql_type == self.determine_sql_type_for_string_property():
+            return value
+        elif sql_type == self.determine_sql_type_for_hex_property():
+            return bytes.fromhex(value)

stix2/datastore/relational_db/database_backends/postgres_backend.py

@@ -0,0 +1,66 @@
+import os
+from typing import Any
+
+from sqlalchemy import TIMESTAMP, LargeBinary, Text
+from sqlalchemy.schema import CreateSchema
+
+from stix2.base import (
+    _DomainObject, _MetaObject, _Observable, _RelationshipObject,
+)
+from stix2.datastore.relational_db.utils import schema_for
+
+from .database_backend_base import DatabaseBackend
+
+
+class PostgresBackend(DatabaseBackend):
+    default_database_connection_url = \
+        f"postgresql://{os.getenv('POSTGRES_USER', 'postgres')}:" + \
+        f"{os.getenv('POSTGRES_PASSWORD', 'postgres')}@" + \
+        f"{os.getenv('POSTGRES_IP_ADDRESS', '0.0.0.0')}:" + \
+        f"{os.getenv('POSTGRES_PORT', '5432')}/postgres"
+
+    def __init__(self, database_connection_url=default_database_connection_url, force_recreate=False, **kwargs: Any):
+        super().__init__(database_connection_url, force_recreate=force_recreate, **kwargs)
+
+    def _create_schemas(self):
+        with self.database_connection.begin() as trans:
+            trans.execute(CreateSchema("common", if_not_exists=True))
+            trans.execute(CreateSchema("sdo", if_not_exists=True))
+            trans.execute(CreateSchema("sco", if_not_exists=True))
+            trans.execute(CreateSchema("sro", if_not_exists=True))
+
+    @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"
+
+    @staticmethod
+    def schema_for(stix_class):
+        return schema_for(stix_class)
+
+    @staticmethod
+    def schema_for_core():
+        return "common"
+
+    @staticmethod
+    def determine_sql_type_for_binary_property():  # noqa: F811
+        return PostgresBackend.determine_sql_type_for_string_property()
+
+    @staticmethod
+    def determine_sql_type_for_hex_property():  # noqa: F811
+        # return LargeBinary
+        return PostgresBackend.determine_sql_type_for_string_property()
+
+    @staticmethod
+    def determine_sql_type_for_timestamp_property():  # noqa: F811
+        return TIMESTAMP(timezone=True)
+
+    @staticmethod
+    def array_allowed():
+        return True