Merge pull request #6 from rpiazza/cyber-observables

More Cyber Observable Object classes

stix2/__init__.py

@@ -3,7 +3,9 @@
 # flake8: noqa
 
 from .bundle import Bundle
-from .observables import Artifact, AutonomousSystem, EmailAddress, File
+from .observables import Artifact, AutonomousSystem, Directory, DomainName, EmailAddress, File, IPv4Address, \
+    IPv6Address, MACAddress, Mutex, NetworkTraffic, Process, Software, URL, UserAccount, WindowsRegistryKey, \
+    X509Certificate
 from .other import ExternalReference, KillChainPhase, MarkingDefinition, \
     GranularMarking, StatementMarking, TLPMarking
 from .sdo import AttackPattern, Campaign, CourseOfAction, Identity, Indicator, \
@@ -35,8 +37,21 @@ OBJ_MAP = {
 OBJ_MAP_OBSERVABLE = {
     'artifact': Artifact,
     'autonomous-system': AutonomousSystem,
+    'directory': Directory,
+    'domain-name': DomainName,
     'email-address': EmailAddress,
     'file': File,
+    'ipv4-addr': IPv4Address,
+    'ipv6-addr': IPv6Address,
+    'mac-addr': MACAddress,
+    'mutex': Mutex,
+    'network-traffic': NetworkTraffic,
+    'process': Process,
+    'software': Software,
+    'url': URL,
+    'user-account': UserAccount,
+    'windows-registry-key': WindowsRegistryKey,
+    'x509-certificate': X509Certificate,
 }
 
 

stix2/base.py

@@ -148,18 +148,21 @@ class _STIXBase(collections.Mapping):
 class Observable(_STIXBase):
 
     def __init__(self, **kwargs):
-        self._STIXBase__valid_refs = kwargs.pop('_valid_refs')
+        # the constructor might be called independently of an observed data object
+        if '_valid_refs' in kwargs:
+            self._STIXBase__valid_refs = kwargs.pop('_valid_refs')
+        else:
+            self._STIXBase__valid_refs = []
         super(Observable, self).__init__(**kwargs)
 
     def _check_property(self, prop_name, prop, kwargs):
         super(Observable, self)._check_property(prop_name, prop, kwargs)
-        if prop_name.endswith('_ref'):
+        if prop_name.endswith('_ref') and prop_name in kwargs:
-            ref = kwargs[prop_name].split('--', 1)[0]
+            ref = kwargs[prop_name]
             if ref not in self._STIXBase__valid_refs:
                 raise InvalidObjRefError(self.__class__, prop_name, "'%s' is not a valid object in local scope" % ref)
-        if prop_name.endswith('_refs'):
+        if prop_name.endswith('_refs') and prop_name in kwargs:
-            for r in kwargs[prop_name]:
+            for ref in kwargs[prop_name]:
-                ref = r.split('--', 1)[0]
                 if ref not in self._STIXBase__valid_refs:
                     raise InvalidObjRefError(self.__class__, prop_name, "'%s' is not a valid object in local scope" % ref)
         # TODO also check the type of the object referenced, not just that the key exists

stix2/observables.py

@@ -1,11 +1,12 @@
 """STIX 2.0 Cyber Observable Objects"""
 
-from .base import Observable
+from .base import _STIXBase, Observable
 # from .properties import (BinaryProperty, BooleanProperty, DictionaryProperty,
 #                          HashesProperty, HexProperty, IDProperty,
 #                          IntegerProperty, ListProperty, ReferenceProperty,
 #                          StringProperty, TimestampProperty, TypeProperty)
-from .properties import BinaryProperty, HashesProperty, IntegerProperty, ObjectReferenceProperty, StringProperty, TypeProperty
+from .properties import BinaryProperty, BooleanProperty, DictionaryProperty, HashesProperty, HexProperty, \
+    IntegerProperty, ListProperty, ObjectReferenceProperty, Property, StringProperty, TimestampProperty, TypeProperty
 
 
 class Artifact(Observable):
@@ -29,6 +30,29 @@ class AutonomousSystem(Observable):
     }
 
 
+class Directory(Observable):
+    _type = 'directory'
+    _properties = {
+        'type': TypeProperty(_type),
+        'path': StringProperty(required=True),
+        'path_enc': StringProperty(),
+        # these are not the created/modified timestamps of the object itself
+        'created': TimestampProperty(),
+        'modified': TimestampProperty(),
+        'accessed': TimestampProperty(),
+        'contains_refs': ListProperty(ObjectReferenceProperty),
+    }
+
+
+class DomainName(Observable):
+    _type = 'domain-name'
+    _properties = {
+        'type': TypeProperty(_type),
+        'value': StringProperty(required=True),
+        'resolves_to_refs': ListProperty(ObjectReferenceProperty),
+    }
+
+
 class EmailAddress(Observable):
     _type = 'email-address'
     _properties = {
@@ -43,4 +67,187 @@ class File(Observable):
     _type = 'file'
     _properties = {
         'type': TypeProperty(_type),
+        # extensions
+        'hashes': HashesProperty(),
+        'size': IntegerProperty(),
+        'name': StringProperty(),
+        'name_enc': StringProperty(),
+        'magic_number_hex': HexProperty(),
+        'mime_type': StringProperty(),
+        # these are not the created/modified timestamps of the object itself
+        'created': TimestampProperty(),
+        'modified': TimestampProperty(),
+        'accessed': TimestampProperty(),
+        'parent_directory_ref': ObjectReferenceProperty(),
+        'is_encrypted': BooleanProperty(),
+        'encyption_algorithm': StringProperty(),
+        'decryption_key': StringProperty(),
+        'contains_refs': ListProperty(ObjectReferenceProperty),
+        'content_ref': ObjectReferenceProperty(),
+    }
+
+
+class IPv4Address(Observable):
+    _type = 'ipv4-addr'
+    _properties = {
+        'type': TypeProperty(_type),
+        'value': StringProperty(required=True),
+        'resolves_to_refs': ListProperty(ObjectReferenceProperty),
+        'belongs_to_refs': ListProperty(ObjectReferenceProperty),
+    }
+
+
+class IPv6Address(Observable):
+    _type = 'ipv6-addr'
+    _properties = {
+        'type': TypeProperty(_type),
+        'value': StringProperty(required=True),
+        'resolves_to_refs': ListProperty(ObjectReferenceProperty),
+        'belongs_to_refs': ListProperty(ObjectReferenceProperty),
+    }
+
+
+class MACAddress(Observable):
+    _type = 'mac-addr'
+    _properties = {
+        'type': TypeProperty(_type),
+        'value': StringProperty(required=True),
+    }
+
+
+class Mutex(Observable):
+    _type = 'mutex'
+    _properties = {
+        'type': TypeProperty(_type),
+        'name': StringProperty(),
+    }
+
+
+class NetworkTraffic(Observable):
+    _type = 'network-traffic'
+    _properties = {
+        'type': TypeProperty(_type),
+        # extensions
+        'start': TimestampProperty(),
+        'end': TimestampProperty(),
+        'is_active': BooleanProperty(),
+        'src_ref': ObjectReferenceProperty(),
+        'dst_ref': ObjectReferenceProperty(),
+        'src_port': IntegerProperty(),
+        'dst_port': IntegerProperty(),
+        'protocols': ListProperty(StringProperty),
+        'src_byte_count': IntegerProperty(),
+        'dst_byte_count': IntegerProperty(),
+        'src_packets': IntegerProperty(),
+        'dst_packets': IntegerProperty(),
+        'ipfix': DictionaryProperty(),
+        'src_payload_ref': ObjectReferenceProperty(),
+        'dst_payload_ref': ObjectReferenceProperty(),
+        'encapsulates_refs': ListProperty(ObjectReferenceProperty),
+        'encapsulates_by_ref': ObjectReferenceProperty(),
+    }
+
+
+class Process(Observable):
+    _type = 'process'
+    _properties = {
+        'type': TypeProperty(_type),
+        # extensions
+        'is_hidden': BooleanProperty(),
+        'pid': IntegerProperty(),
+        'name': StringProperty(),
+        # this is not the created timestamps of the object itself
+        'created': TimestampProperty(),
+        'cwd': StringProperty(),
+        'arguments': ListProperty(StringProperty),
+        'command_line': StringProperty(),
+        'environment_variables': DictionaryProperty(),
+        'opened_connection_refs': ListProperty(ObjectReferenceProperty),
+        'creator_user_ref': ObjectReferenceProperty(),
+        'binary_ref': ObjectReferenceProperty(),
+        'parent_ref': ObjectReferenceProperty(),
+        'child_refs': ListProperty(ObjectReferenceProperty),
+    }
+
+
+class Software(Observable):
+    _type = 'software'
+    _properties = {
+        'type': TypeProperty(_type),
+        'name': StringProperty(required=True),
+        'cpe': StringProperty(),
+        'languages': ListProperty(StringProperty),
+        'vendor': StringProperty(),
+        'version': StringProperty(),
+    }
+
+
+class URL(Observable):
+    _type = 'url'
+    _properties = {
+        'type': TypeProperty(_type),
+        'value': StringProperty(required=True),
+    }
+
+
+class UserAccount(Observable):
+    _type = 'user-account'
+    _properties = {
+        'type': TypeProperty(_type),
+        # extensions
+        'user_id': StringProperty(required=True),
+        'account_login': StringProperty(),
+        'account_type': StringProperty(),
+        'display_name': StringProperty(),
+        'is_service_account': BooleanProperty(),
+        'is_privileged': BooleanProperty(),
+        'can_escalate_privs': BooleanProperty(),
+        'is_disabled': BooleanProperty(),
+        'account_created': TimestampProperty(),
+        'account_expires': TimestampProperty(),
+        'password_last_changed': TimestampProperty(),
+        'account_first_login': TimestampProperty(),
+        'account_last_login': TimestampProperty(),
+    }
+
+
+class WindowsRegistryValueType(_STIXBase):
+    _type = 'windows-registry-value-type'
+    _properties = {
+        'name': StringProperty(required=True),
+        'data': StringProperty(),
+        'data_type': Property()
+    }
+
+
+class WindowsRegistryKey(Observable):
+    _type = 'windows-registry-key'
+    _properties = {
+        'type': TypeProperty(_type),
+        'key': StringProperty(required=True),
+        'values': ListProperty(WindowsRegistryValueType),
+        # this is not the modified timestamps of the object itself
+        'modified': TimestampProperty(),
+        'creator_user_ref': ObjectReferenceProperty(),
+        'number_of_subkeys': IntegerProperty(),
+    }
+
+
+class X509Certificate(Observable):
+    _type = 'x509-certificate'
+    _properties = {
+        'type': TypeProperty(_type),
+        'is_self_signed': BooleanProperty(),
+        'hashes': HashesProperty(),
+        'version': StringProperty(),
+        'serial_number': StringProperty(),
+        'signature_algorithm': StringProperty(),
+        'issuer': StringProperty(),
+        'validity_not_before': TimestampProperty(),
+        'validity_not_after': TimestampProperty(),
+        'subject': StringProperty(),
+        'subject_public_key_algorithm': StringProperty(),
+        'subject_public_key_modulus': StringProperty(),
+        'subject_public_key_exponent': IntegerProperty(),
+        'x509_v3_extensions': Property(),
     }

stix2/test/test_observed_data.py

@@ -44,6 +44,84 @@ def test_observed_data_example():
     assert str(observed_data) == EXPECTED
 
 
+EXPECTED_WITH_REF = """{
+    "created": "2016-04-06T19:58:16Z",
+    "created_by_ref": "identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+    "first_observed": "2015-12-21T19:00:00Z",
+    "id": "observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
+    "last_observed": "2015-12-21T19:00:00Z",
+    "modified": "2016-04-06T19:58:16Z",
+    "number_observed": 50,
+    "objects": {
+        "0": {
+            "name": "foo.exe",
+            "type": "file"
+        },
+        "1": {
+            "contains_refs": [
+                "0"
+            ],
+            "path": "/usr/home",
+            "type": "directory"
+        }
+    },
+    "type": "observed-data"
+}"""
+
+
+def test_observed_data_example_with_refs():
+    observed_data = stix2.ObservedData(
+        id="observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
+        created_by_ref="identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+        created="2016-04-06T19:58:16Z",
+        modified="2016-04-06T19:58:16Z",
+        first_observed="2015-12-21T19:00:00Z",
+        last_observed="2015-12-21T19:00:00Z",
+        number_observed=50,
+        objects={
+            "0": {
+                "type": "file",
+                "name": "foo.exe"
+            },
+            "1": {
+                "type": "directory",
+                "path": "/usr/home",
+                "contains_refs": ["0"]
+            }
+        },
+    )
+
+    assert str(observed_data) == EXPECTED_WITH_REF
+
+
+def test_observed_data_example_with_bad_refs():
+    with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo:
+        stix2.ObservedData(
+            id="observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
+            created_by_ref="identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+            created="2016-04-06T19:58:16Z",
+            modified="2016-04-06T19:58:16Z",
+            first_observed="2015-12-21T19:00:00Z",
+            last_observed="2015-12-21T19:00:00Z",
+            number_observed=50,
+            objects={
+                "0": {
+                    "type": "file",
+                    "name": "foo.exe"
+                },
+                "1": {
+                    "type": "directory",
+                    "path": "/usr/home",
+                    "contains_refs": ["2"]
+                }
+            },
+        )
+
+    assert excinfo.value.cls == stix2.ObservedData
+    assert excinfo.value.prop_name == "objects"
+    assert excinfo.value.reason == "Invalid object reference for 'Directory:contains_refs': '2' is not a valid object in local scope"
+
+
 @pytest.mark.parametrize("data", [
     EXPECTED,
     {
@@ -152,4 +230,124 @@ def test_parse_email_address(data):
         stix2.parse(odata_str)
 
 
-# TODO: Add other examples
+#  creating cyber observables directly
+
+def test_directory_example():
+    dir = stix2.Directory(_valid_refs=["1"],
+                          path='/usr/lib',
+                          created="2015-12-21T19:00:00Z",
+                          modified="2015-12-24T19:00:00Z",
+                          accessed="2015-12-21T20:00:00Z",
+                          contains_refs=["1"])
+
+    assert dir.path == '/usr/lib'
+    assert dir.created == dt.datetime(2015, 12, 21, 19, 0, 0, tzinfo=pytz.utc)
+    assert dir.modified == dt.datetime(2015, 12, 24, 19, 0, 0, tzinfo=pytz.utc)
+    assert dir.accessed == dt.datetime(2015, 12, 21, 20, 0, 0, tzinfo=pytz.utc)
+    assert dir.contains_refs == ["1"]
+
+
+def test_directory_example_ref_error():
+    with pytest.raises(stix2.exceptions.InvalidObjRefError) as excinfo:
+        stix2.Directory(_valid_refs=[],
+                        path='/usr/lib',
+                        created="2015-12-21T19:00:00Z",
+                        modified="2015-12-24T19:00:00Z",
+                        accessed="2015-12-21T20:00:00Z",
+                        contains_refs=["1"])
+
+    assert excinfo.value.cls == stix2.Directory
+    assert excinfo.value.prop_name == "contains_refs"
+
+
+def test_domain_name_example():
+    dn = stix2.DomainName(_valid_refs=["1"],
+                          value="example.com",
+                          resolves_to_refs=["1"])
+
+    assert dn.value == "example.com"
+    assert dn.resolves_to_refs == ["1"]
+
+
+def test_file_example():
+    f = stix2.File(name="qwerty.dll",
+                   hashes={
+                    "SHA-256": "ceafbfd424be2ca4a5f0402cae090dda2fb0526cf521b60b60077c0f622b285a"},
+                   size=100,
+                   magic_number_hex="1C",
+                   mime_type="application/msword",
+                   created="2016-12-21T19:00:00Z",
+                   modified="2016-12-24T19:00:00Z",
+                   accessed="2016-12-21T20:00:00Z",
+                   is_encrypted=True,
+                   encyption_algorithm="AES128-CBC",
+                   decryption_key="fred"
+                   )
+
+    assert f.name == "qwerty.dll"
+    assert f.size == 100
+    assert f.magic_number_hex == "1C"
+    assert f.hashes["SHA-256"] == "ceafbfd424be2ca4a5f0402cae090dda2fb0526cf521b60b60077c0f622b285a"
+    assert f.mime_type == "application/msword"
+    assert f.created == dt.datetime(2016, 12, 21, 19, 0, 0, tzinfo=pytz.utc)
+    assert f.modified == dt.datetime(2016, 12, 24, 19, 0, 0, tzinfo=pytz.utc)
+    assert f.accessed == dt.datetime(2016, 12, 21, 20, 0, 0, tzinfo=pytz.utc)
+    assert f.is_encrypted
+    assert f.encyption_algorithm == "AES128-CBC"
+    assert f.decryption_key == "fred"   # does the key have a format we can test for?
+
+
+# def test_file_example_encyption_error():
+#     f = stix2.File(name="qwerty.dll",
+#                    is_encrypted=False,
+#                    encyption_algorithm="AES128-CBC"
+#                    )
+#
+#     assert f.name == "qwerty.dll"
+#     assert f.is_encrypted == False
+#     assert f.encyption_algorithm == "AES128-CBC"
+
+
+def test_ip4_address_example():
+    ip4 = stix2.IPv4Address(_valid_refs=["1", "4", "5"],
+                            value="198.51.100.3",
+                            resolves_to_refs=["4", "5"])
+
+    assert ip4.value == "198.51.100.3"
+    assert ip4.resolves_to_refs == ["4", "5"]
+
+
+def test_ip4_address_example_cidr():
+    ip4 = stix2.IPv4Address(value="198.51.100.0/24")
+
+    assert ip4.value == "198.51.100.0/24"
+
+
+def test_ip6_address_example():
+    ip6 = stix2.IPv6Address(value="2001:0db8:85a3:0000:0000:8a2e:0370:7334")
+
+    assert ip6.value == "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
+
+
+def test_mac_address_example():
+    ip6 = stix2.MACAddress(value="d2:fb:49:24:37:18")
+
+    assert ip6.value == "d2:fb:49:24:37:18"
+
+
+def test_mutex_example():
+    m = stix2.Mutex(name="barney")
+
+    assert m.name == "barney"
+
+
+def test_software_example():
+    s = stix2.Software(name="Word",
+                       cpe="cpe:2.3:a:microsoft:word:2000:*:*:*:*:*:*:*",
+                       version="2002",
+                       vendor="Microsoft")
+
+    assert s.name == "Word"
+    assert s.cpe == "cpe:2.3:a:microsoft:word:2000:*:*:*:*:*:*:*"
+    assert s.version == "2002"
+    assert s.vendor == "Microsoft"