MISP
/
cti-python-stix2
mirror of https://github.com/MISP/cti-python-stix2
2
0
Fork 0
Code Issues Releases Wiki Activity

Merge 'master' and fix imports for consistency

pull/1/head
Chris Lenk 2021-01-15 12:34:10 -05:00
parent 7de5c458bb 9ed62699ca
commit a0d535336e
37 changed files with 532 additions and 422 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
.github/workflows/python-ci-tests.yml vendored Normal file
View File

@ -0,0 +1,33 @@
# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
name: cti-python-stix2 test harness
on: [push, pull_request]
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.6, 3.7, 3.8, 3.9]
name: Python ${{ matrix.python-version }} Build
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install and update essential dependencies
run: |
pip install -U pip setuptools
pip install tox-gh-actions
pip install codecov
- name: Test with Tox
run: |
tox
- name: Upload coverage information to Codecov
uses: codecov/codecov-action@v1
with:
fail_ci_if_error: true # optional (default = false)
verbose: true # optional (default = false)

24
.pre-commit-config.yaml
View File

@ -1,17 +1,25 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v1.3.0
rev: v3.4.0
hooks:
- id: trailing-whitespace
- id: flake8
args:
- --max-line-length=160
- id: check-merge-conflict
- repo: https://github.com/asottile/add-trailing-comma
rev: v0.6.4
rev: v2.0.2
hooks:
- id: add-trailing-comma
- repo: https://github.com/FalconSocial/pre-commit-python-sorter
sha: b57843b0b874df1d16eb0bef00b868792cb245c2
- repo: https://github.com/PyCQA/flake8
rev: 3.8.4
hooks:
- id: python-import-sorter
- id: flake8
name: Check project styling
args:
- --max-line-length=160
- repo: https://github.com/PyCQA/isort
rev: 5.7.0
hooks:
- id: isort
name: Sort python imports (shows diff)
args: ["-c", "--diff"]
- id: isort
name: Sort python imports (fixes files)

19
.travis.yml
View File

@ -1,19 +0,0 @@
os: linux
language: python
cache: pip
dist: bionic
python:
- "3.5"
- "3.6"
- "3.7"
- "3.8"
install:
- pip install -U pip setuptools
- pip install tox-travis
- pip install codecov
- pip install pre-commit
script:
- tox
- pre-commit run --all-files
after_success:
- codecov

4
README.rst
View File

@ -163,8 +163,8 @@ questions about TC Open Repository participation to OASIS Staff at
repository-admin@oasis-open.org and any specific CLA-related questions
to repository-cla@oasis-open.org.
.. |Build_Status| image:: https://travis-ci.org/oasis-open/cti-python-stix2.svg?branch=master
:target: https://travis-ci.org/oasis-open/cti-python-stix2
.. |Build_Status| image:: https://github.com/oasis-open/cti-python-stix2/workflows/cti-python-stix2%20test%20harness/badge.svg
:target: https://github.com/oasis-open/cti-python-stix2/actions?query=workflow%3A%22cti-python-stix2+test+harness%22
:alt: Build Status
.. |Coverage| image:: https://codecov.io/gh/oasis-open/cti-python-stix2/branch/master/graph/badge.svg
:target: https://codecov.io/gh/oasis-open/cti-python-stix2

2
setup.py
View File

@ -40,10 +40,10 @@ setup(
'Topic :: Security',
'License :: OSI Approved :: BSD License',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
],
keywords='stix stix2 json cti cyber threat intelligence',
packages=find_packages(exclude=['*.test', '*.test.*']),

12
stix2/base.py
View File

@ -164,8 +164,10 @@ class _STIXBase(Mapping):
defaulted = []
for name, prop in self._properties.items():
try:
if (not prop.required and not hasattr(prop, '_fixed_value') and
prop.default() == setting_kwargs[name]):
if (
not prop.required and not hasattr(prop, '_fixed_value') and
prop.default() == setting_kwargs[name]
):
defaulted.append(name)
except (AttributeError, KeyError):
continue
@ -194,8 +196,10 @@ class _STIXBase(Mapping):
unpickling = '_inner' not in self.__dict__
if not unpickling and name in self:
return self.__getitem__(name)
raise AttributeError("'%s' object has no attribute '%s'" %
(self.__class__.__name__, name))
raise AttributeError(
"'%s' object has no attribute '%s'" %
(self.__class__.__name__, name),
)
def __setattr__(self, name, value):
if not name.startswith("_"):

18
stix2/datastore/memory.py
View File

@ -75,8 +75,10 @@ class _ObjectFamily(object):
def add(self, obj):
self.all_versions[obj["modified"]] = obj
if (self.latest_version is None or
obj["modified"] > self.latest_version["modified"]):
if (
self.latest_version is None or
obj["modified"] > self.latest_version["modified"]
):
self.latest_version = obj
def __str__(self):
@ -188,11 +190,13 @@ class MemorySink(DataSink):
def save_to_file(self, path, encoding="utf-8"):
path = os.path.abspath(path)
all_objs = list(itertools.chain.from_iterable(
value.all_versions.values() if isinstance(value, _ObjectFamily)
else [value]
for value in self._data.values()
))
all_objs = list(
itertools.chain.from_iterable(
value.all_versions.values() if isinstance(value, _ObjectFamily)
else [value]
for value in self._data.values()
),
)
if any("spec_version" in x for x in all_objs):
bundle = v21.Bundle(all_objs, allow_custom=self.allow_custom)

26
stix2/equivalence/pattern/__init__.py
View File

@ -9,18 +9,14 @@
|
"""
import stix2
from stix2.equivalence.pattern.compare.observation import (
observation_expression_cmp,
)
from stix2.equivalence.pattern.transform import (
ChainTransformer, SettleTransformer,
)
from stix2.equivalence.pattern.transform.observation import (
from ... import pattern_visitor
from ...version import DEFAULT_VERSION
from .compare.observation import observation_expression_cmp
from .transform import ChainTransformer, SettleTransformer
from .transform.observation import (
AbsorptionTransformer, CanonicalizeComparisonExpressionsTransformer,
DNFTransformer, FlattenTransformer, OrderDedupeTransformer,
)
import stix2.pattern_visitor
# Lazy-initialize
_pattern_canonicalizer = None
@ -61,7 +57,7 @@ def _get_pattern_canonicalizer():
return _pattern_canonicalizer
def equivalent_patterns(pattern1, pattern2, stix_version=stix2.DEFAULT_VERSION):
def equivalent_patterns(pattern1, pattern2, stix_version=DEFAULT_VERSION):
"""
Determine whether two STIX patterns are semantically equivalent.
@ -74,10 +70,10 @@ def equivalent_patterns(pattern1, pattern2, stix_version=stix2.DEFAULT_VERSION):
Returns:
True if the patterns are semantically equivalent; False if not
"""
patt_ast1 = stix2.pattern_visitor.create_pattern_object(
patt_ast1 = pattern_visitor.create_pattern_object(
pattern1, version=stix_version,
)
patt_ast2 = stix2.pattern_visitor.create_pattern_object(
patt_ast2 = pattern_visitor.create_pattern_object(
pattern2, version=stix_version,
)
@ -91,7 +87,7 @@ def equivalent_patterns(pattern1, pattern2, stix_version=stix2.DEFAULT_VERSION):
def find_equivalent_patterns(
search_pattern, patterns, stix_version=stix2.DEFAULT_VERSION,
search_pattern, patterns, stix_version=DEFAULT_VERSION,
):
"""
Find patterns from a sequence which are equivalent to a given pattern.
@ -109,7 +105,7 @@ def find_equivalent_patterns(
Returns:
A generator iterator producing the semantically equivalent patterns
"""
search_pattern_ast = stix2.pattern_visitor.create_pattern_object(
search_pattern_ast = pattern_visitor.create_pattern_object(
search_pattern, version=stix_version,
)
@ -119,7 +115,7 @@ def find_equivalent_patterns(
)
for pattern in patterns:
pattern_ast = stix2.pattern_visitor.create_pattern_object(
pattern_ast = pattern_visitor.create_pattern_object(
pattern, version=stix_version,
)
canon_pattern_ast, _ = pattern_canonicalizer.transform(pattern_ast)

4
stix2/equivalence/pattern/transform/comparison.py
View File

@ -144,7 +144,7 @@ class ComparisonExpressionTransformer(Transformer):
class OrderDedupeTransformer(
ComparisonExpressionTransformer
ComparisonExpressionTransformer,
):
"""
Canonically order the children of all nodes in the AST. Because the
@ -247,7 +247,7 @@ class FlattenTransformer(ComparisonExpressionTransformer):
class AbsorptionTransformer(
ComparisonExpressionTransformer
ComparisonExpressionTransformer,
):
"""
Applies boolean "absorption" rules for AST simplification. E.g.:

14
stix2/equivalence/pattern/transform/observation.py
View File

@ -152,9 +152,11 @@ class ObservationExpressionTransformer(Transformer):
changed = True
else:
raise TypeError("Not an observation expression: {}: {}".format(
type(ast).__name__, str(ast),
))
raise TypeError(
"Not an observation expression: {}: {}".format(
type(ast).__name__, str(ast),
),
)
return result, changed
@ -229,7 +231,7 @@ class FlattenTransformer(ObservationExpressionTransformer):
class OrderDedupeTransformer(
ObservationExpressionTransformer
ObservationExpressionTransformer,
):
"""
Canonically order AND/OR expressions, and dedupe ORs. E.g.:
@ -272,7 +274,7 @@ class OrderDedupeTransformer(
class AbsorptionTransformer(
ObservationExpressionTransformer
ObservationExpressionTransformer,
):
"""
Applies boolean "absorption" rules for observation expressions, for AST
@ -479,7 +481,7 @@ class DNFTransformer(ObservationExpressionTransformer):
class CanonicalizeComparisonExpressionsTransformer(
ObservationExpressionTransformer
ObservationExpressionTransformer,
):
"""
Canonicalize all comparison expressions.

6
stix2/parsing.py
View File

@ -182,7 +182,9 @@ def parse_observable(data, _valid_refs=None, allow_custom=False, version=None):
# flag allows for unknown custom objects too, but will not
# be parsed into STIX observable object, just returned as is
return obj
raise ParseError("Can't parse unknown observable type '%s'! For custom observables, "
"use the CustomObservable decorator." % obj['type'])
raise ParseError(
"Can't parse unknown observable type '%s'! For custom observables, "
"use the CustomObservable decorator." % obj['type'],
)
return obj_class(allow_custom=allow_custom, **obj)

17
stix2/pattern_visitor.py
View File

@ -17,10 +17,9 @@ from stix2patterns.v21.grammars.STIXPatternVisitor import \
STIXPatternVisitor as STIXPatternVisitor21
from stix2patterns.v21.pattern import Pattern as Pattern21
import stix2
from .patterns import *
from .patterns import _BooleanExpression
from .version import DEFAULT_VERSION
# flake8: noqa F405
@ -261,11 +260,13 @@ class STIXPatternVisitorForSTIX2():
property_path.append(self.instantiate("ListObjectPathComponent", current.property_name, next.getText()))
i += 2
elif isinstance(next, IntegerConstant):
property_path.append(self.instantiate(
"ListObjectPathComponent",
current.property_name if isinstance(current, BasicObjectPathComponent) else text_type(current),
next.value,
))
property_path.append(
self.instantiate(
"ListObjectPathComponent",
current.property_name if isinstance(current, BasicObjectPathComponent) else text_type(current),
next.value,
),
)
i += 2
else:
property_path.append(current)
@ -389,7 +390,7 @@ class STIXPatternVisitorForSTIX20(STIXPatternVisitorForSTIX2, STIXPatternVisitor
super(STIXPatternVisitor20, self).__init__()
def create_pattern_object(pattern, module_suffix="", module_name="", version=stix2.DEFAULT_VERSION):
def create_pattern_object(pattern, module_suffix="", module_name="", version=DEFAULT_VERSION):
"""
Create a STIX pattern AST from a pattern string.
"""

31
stix2/properties.py
View File

@ -9,8 +9,6 @@ import uuid
from six import string_types, text_type
import stix2
from .base import _STIXBase
from .exceptions import (
CustomContentError, DictionaryKeyError, MissingPropertiesError,
@ -19,6 +17,7 @@ from .exceptions import (
from .parsing import parse, parse_observable
from .registry import STIX2_OBJ_MAPS
from .utils import _get_dict, get_class_hierarchy_names, parse_into_datetime
from .version import DEFAULT_VERSION
try:
from collections.abc import Mapping
@ -248,9 +247,11 @@ class ListProperty(Property):
valid = self.contained(**item)
else:
raise ValueError("Can't create a {} out of {}".format(
self.contained._type, str(item),
))
raise ValueError(
"Can't create a {} out of {}".format(
self.contained._type, str(item),
),
)
result.append(valid)
@ -274,7 +275,7 @@ class StringProperty(Property):
class TypeProperty(Property):
def __init__(self, type, spec_version=stix2.DEFAULT_VERSION):
def __init__(self, type, spec_version=DEFAULT_VERSION):
_validate_type(type, spec_version)
self.spec_version = spec_version
super(TypeProperty, self).__init__(fixed=type)
@ -282,7 +283,7 @@ class TypeProperty(Property):
class IDProperty(Property):
def __init__(self, type, spec_version=stix2.DEFAULT_VERSION):
def __init__(self, type, spec_version=DEFAULT_VERSION):
self.required_prefix = type + "--"
self.spec_version = spec_version
super(IDProperty, self).__init__()
@ -381,7 +382,7 @@ class TimestampProperty(Property):
class DictionaryProperty(Property):
def __init__(self, spec_version=stix2.DEFAULT_VERSION, **kwargs):
def __init__(self, spec_version=DEFAULT_VERSION, **kwargs):
self.spec_version = spec_version
super(DictionaryProperty, self).__init__(**kwargs)
@ -470,7 +471,7 @@ class HexProperty(Property):
class ReferenceProperty(Property):
def __init__(self, valid_types=None, invalid_types=None, spec_version=stix2.DEFAULT_VERSION, **kwargs):
def __init__(self, valid_types=None, invalid_types=None, spec_version=DEFAULT_VERSION, **kwargs):
"""
references sometimes must be to a specific object type
"""
@ -604,7 +605,7 @@ class ObservableProperty(Property):
"""Property for holding Cyber Observable Objects.
"""
def __init__(self, spec_version=stix2.DEFAULT_VERSION, allow_custom=False, *args, **kwargs):
def __init__(self, spec_version=DEFAULT_VERSION, allow_custom=False, *args, **kwargs):
self.allow_custom = allow_custom
self.spec_version = spec_version
super(ObservableProperty, self).__init__(*args, **kwargs)
@ -639,7 +640,7 @@ class ExtensionsProperty(DictionaryProperty):
"""Property for representing extensions on Observable objects.
"""
def __init__(self, spec_version=stix2.DEFAULT_VERSION, allow_custom=False, enclosing_type=None, required=False):
def __init__(self, spec_version=DEFAULT_VERSION, allow_custom=False, enclosing_type=None, required=False):
self.allow_custom = allow_custom
self.enclosing_type = enclosing_type
super(ExtensionsProperty, self).__init__(spec_version=spec_version, required=required)
@ -681,7 +682,7 @@ class ExtensionsProperty(DictionaryProperty):
class STIXObjectProperty(Property):
def __init__(self, spec_version=stix2.DEFAULT_VERSION, allow_custom=False, *args, **kwargs):
def __init__(self, spec_version=DEFAULT_VERSION, allow_custom=False, *args, **kwargs):
self.allow_custom = allow_custom
self.spec_version = spec_version
super(STIXObjectProperty, self).__init__(*args, **kwargs)
@ -689,8 +690,10 @@ class STIXObjectProperty(Property):
def clean(self, value):
# Any STIX Object (SDO, SRO, or Marking Definition) can be added to
# a bundle with no further checks.
if any(x in ('_DomainObject', '_RelationshipObject', 'MarkingDefinition')
for x in get_class_hierarchy_names(value)):
if any(
x in ('_DomainObject', '_RelationshipObject', 'MarkingDefinition')
for x in get_class_hierarchy_names(value)
):
# A simple "is this a spec version 2.1+ object" test. For now,
# limit 2.0 bundles to 2.0 objects. It's not possible yet to
# have validation co-constraints among properties, e.g. have

50
stix2/registration.py
View File

@ -1,14 +1,14 @@
import re
import stix2
import stix2.registry
from . import registry
from .base import _DomainObject, _Observable
from .exceptions import DuplicateRegistrationError
from .properties import _validate_type
from .utils import PREFIX_21_REGEX, get_class_hierarchy_names
from .version import DEFAULT_VERSION
def _register_object(new_type, version=stix2.DEFAULT_VERSION):
def _register_object(new_type, version=DEFAULT_VERSION):
"""Register a custom STIX Object type.
Args:
@ -40,15 +40,15 @@ def _register_object(new_type, version=stix2.DEFAULT_VERSION):
v = 'v' + version.replace('.', '')
else:
# Use default version (latest) if no version was provided.
v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
v = 'v' + DEFAULT_VERSION.replace('.', '')
OBJ_MAP = stix2.registry.STIX2_OBJ_MAPS[v]['objects']
OBJ_MAP = registry.STIX2_OBJ_MAPS[v]['objects']
if new_type._type in OBJ_MAP.keys():
raise DuplicateRegistrationError("STIX Object", new_type._type)
OBJ_MAP[new_type._type] = new_type
def _register_marking(new_marking, version=stix2.DEFAULT_VERSION):
def _register_marking(new_marking, version=DEFAULT_VERSION):
"""Register a custom STIX Marking Definition type.
Args:
@ -61,7 +61,7 @@ def _register_marking(new_marking, version=stix2.DEFAULT_VERSION):
mark_type = new_marking._type
properties = new_marking._properties
stix2.properties._validate_type(mark_type, version)
_validate_type(mark_type, version)
if version == "2.1":
for prop_name, prop_value in properties.items():
@ -72,15 +72,15 @@ def _register_marking(new_marking, version=stix2.DEFAULT_VERSION):
v = 'v' + version.replace('.', '')
else:
# Use default version (latest) if no version was provided.
v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
v = 'v' + DEFAULT_VERSION.replace('.', '')
OBJ_MAP_MARKING = stix2.registry.STIX2_OBJ_MAPS[v]['markings']
OBJ_MAP_MARKING = registry.STIX2_OBJ_MAPS[v]['markings']
if mark_type in OBJ_MAP_MARKING.keys():
raise DuplicateRegistrationError("STIX Marking", mark_type)
OBJ_MAP_MARKING[mark_type] = new_marking
def _register_observable(new_observable, version=stix2.DEFAULT_VERSION):
def _register_observable(new_observable, version=DEFAULT_VERSION):
"""Register a custom STIX Cyber Observable type.
Args:
@ -99,8 +99,12 @@ def _register_observable(new_observable, version=stix2.DEFAULT_VERSION):
"'%s' is named like an object reference property but "
"is not an ObjectReferenceProperty." % prop_name,
)
elif (prop_name.endswith('_refs') and ('ListProperty' not in get_class_hierarchy_names(prop) or
'ObjectReferenceProperty' not in get_class_hierarchy_names(prop.contained))):
elif (
prop_name.endswith('_refs') and (
'ListProperty' not in get_class_hierarchy_names(prop) or
'ObjectReferenceProperty' not in get_class_hierarchy_names(prop.contained)
)
):
raise ValueError(
"'%s' is named like an object reference list property but "
"is not a ListProperty containing ObjectReferenceProperty." % prop_name,
@ -115,8 +119,12 @@ def _register_observable(new_observable, version=stix2.DEFAULT_VERSION):
"'%s' is named like a reference property but "
"is not a ReferenceProperty." % prop_name,
)
elif (prop_name.endswith('_refs') and ('ListProperty' not in get_class_hierarchy_names(prop) or
'ReferenceProperty' not in get_class_hierarchy_names(prop.contained))):
elif (
prop_name.endswith('_refs') and (
'ListProperty' not in get_class_hierarchy_names(prop) or
'ReferenceProperty' not in get_class_hierarchy_names(prop.contained)
)
):
raise ValueError(
"'%s' is named like a reference list property but "
"is not a ListProperty containing ReferenceProperty." % prop_name,
@ -126,16 +134,16 @@ def _register_observable(new_observable, version=stix2.DEFAULT_VERSION):
v = 'v' + version.replace('.', '')
else:
# Use default version (latest) if no version was provided.
v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
v = 'v' + DEFAULT_VERSION.replace('.', '')
OBJ_MAP_OBSERVABLE = stix2.registry.STIX2_OBJ_MAPS[v]['observables']
OBJ_MAP_OBSERVABLE = registry.STIX2_OBJ_MAPS[v]['observables']
if new_observable._type in OBJ_MAP_OBSERVABLE.keys():
raise DuplicateRegistrationError("Cyber Observable", new_observable._type)
OBJ_MAP_OBSERVABLE[new_observable._type] = new_observable
def _register_observable_extension(
observable, new_extension, version=stix2.DEFAULT_VERSION,
observable, new_extension, version=DEFAULT_VERSION,
):
"""Register a custom extension to a STIX Cyber Observable type.
@ -155,7 +163,7 @@ def _register_observable_extension(
if not issubclass(obs_class, _Observable):
raise ValueError("'observable' must be a valid Observable class!")
stix2.properties._validate_type(ext_type, version)
_validate_type(ext_type, version)
if not new_extension._properties:
raise ValueError(
@ -184,8 +192,8 @@ def _register_observable_extension(
"created with the @CustomObservable decorator.",
)
OBJ_MAP_OBSERVABLE = stix2.registry.STIX2_OBJ_MAPS[v]['observables']
EXT_MAP = stix2.registry.STIX2_OBJ_MAPS[v]['observable-extensions']
OBJ_MAP_OBSERVABLE = registry.STIX2_OBJ_MAPS[v]['observables']
EXT_MAP = registry.STIX2_OBJ_MAPS[v]['observable-extensions']
try:
if ext_type in EXT_MAP[observable_type].keys():

14
stix2/test/v20/test_datastore_memory.py
View File

@ -175,12 +175,14 @@ def test_memory_source_get_nonexistant_object(mem_source):
def test_memory_store_all_versions(mem_store):
# Add bundle of items to sink
mem_store.add(dict(
id="bundle--%s" % make_id(),
objects=STIX_OBJS2,
spec_version="2.0",
type="bundle",
))
mem_store.add(
dict(
id="bundle--%s" % make_id(),
objects=STIX_OBJS2,
spec_version="2.0",
type="bundle",
),
)
resp = mem_store.all_versions("indicator--00000000-0000-4000-8000-000000000001")
assert len(resp) == 3

20
stix2/test/v20/test_environment.py
View File

@ -39,15 +39,19 @@ def ds2():
cam = stix2.v20.Campaign(id=CAMPAIGN_ID, **CAMPAIGN_KWARGS)
idy = stix2.v20.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS)
ind = stix2.v20.Indicator(id=INDICATOR_ID, created_by_ref=idy.id, **INDICATOR_KWARGS)
indv2 = ind.new_version(external_references=[{
"source_name": "unknown",
"url": "https://examplewebsite.com/",
}])
indv2 = ind.new_version(
external_references=[{
"source_name": "unknown",
"url": "https://examplewebsite.com/",
}],
)
mal = stix2.v20.Malware(id=MALWARE_ID, created_by_ref=idy.id, **MALWARE_KWARGS)
malv2 = mal.new_version(external_references=[{
"source_name": "unknown",
"url": "https://examplewebsite2.com/",
}])
malv2 = mal.new_version(
external_references=[{
"source_name": "unknown",
"url": "https://examplewebsite2.com/",
}],
)
rel1 = stix2.v20.Relationship(ind, 'indicates', mal, id=RELATIONSHIP_IDS[0])
rel2 = stix2.v20.Relationship(mal, 'targets', idy, id=RELATIONSHIP_IDS[1])
rel3 = stix2.v20.Relationship(cam, 'uses', mal, id=RELATIONSHIP_IDS[2])

6
stix2/test/v20/test_indicator.py
View File

@ -20,7 +20,8 @@ EXPECTED_INDICATOR = """{
]
}"""
EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join("""
EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join(
"""
type='indicator',
id='indicator--a740531e-63ff-4e49-a9e1-a0a3eed0e3e7',
created='2017-01-01T00:00:01.000Z',
@ -28,7 +29,8 @@ EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join("""
pattern="[file:hashes.MD5 = 'd41d8cd98f00b204e9800998ecf8427e']",
valid_from='1970-01-01T00:00:01Z',
labels=['malicious-activity']
""".split()) + ")"
""".split(),
) + ")"
def test_indicator_with_all_required_properties():

66
stix2/test/v20/test_observed_data.py
View File

@ -1180,50 +1180,56 @@ def test_process_example_extensions_empty():
def test_process_example_with_WindowsProcessExt_Object():
p = stix2.v20.Process(extensions={
"windows-process-ext": stix2.v20.WindowsProcessExt(
aslr_enabled=True,
dep_enabled=True,
priority="HIGH_PRIORITY_CLASS",
owner_sid="S-1-5-21-186985262-1144665072-74031268-1309",
), # noqa
})
p = stix2.v20.Process(
extensions={
"windows-process-ext": stix2.v20.WindowsProcessExt(
aslr_enabled=True,
dep_enabled=True,
priority="HIGH_PRIORITY_CLASS",
owner_sid="S-1-5-21-186985262-1144665072-74031268-1309",
), # noqa
},
)
assert p.extensions["windows-process-ext"].dep_enabled
assert p.extensions["windows-process-ext"].owner_sid == "S-1-5-21-186985262-1144665072-74031268-1309"
def test_process_example_with_WindowsServiceExt():
p = stix2.v20.Process(extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
p = stix2.v20.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
},
},
})
)
assert p.extensions["windows-service-ext"].service_name == "sirvizio"
assert p.extensions["windows-service-ext"].service_type == "SERVICE_WIN32_OWN_PROCESS"
def test_process_example_with_WindowsProcessServiceExt():
p = stix2.v20.Process(extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
p = stix2.v20.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
},
"windows-process-ext": {
"aslr_enabled": True,
"dep_enabled": True,
"priority": "HIGH_PRIORITY_CLASS",
"owner_sid": "S-1-5-21-186985262-1144665072-74031268-1309",
},
},
"windows-process-ext": {
"aslr_enabled": True,
"dep_enabled": True,
"priority": "HIGH_PRIORITY_CLASS",
"owner_sid": "S-1-5-21-186985262-1144665072-74031268-1309",
},
})
)
assert p.extensions["windows-service-ext"].service_name == "sirvizio"
assert p.extensions["windows-service-ext"].service_type == "SERVICE_WIN32_OWN_PROCESS"

5
stix2/test/v20/test_parsing.py
View File

@ -2,8 +2,7 @@ from collections import OrderedDict
import pytest
import stix2
from stix2 import exceptions, parsing, registration, registry
from stix2 import DEFAULT_VERSION, exceptions, parsing, registration, registry
BUNDLE = {
"type": "bundle",
@ -59,7 +58,7 @@ def test_parse_observable_with_version():
assert v in str(obs_obj.__class__)
@pytest.mark.xfail(reason="The default version is no longer 2.0", condition=stix2.DEFAULT_VERSION != "2.0")
@pytest.mark.xfail(reason="The default version is no longer 2.0", condition=DEFAULT_VERSION != "2.0")
def test_parse_observable_with_no_version():
observable = {"type": "file", "name": "foo.exe"}
obs_obj = parsing.parse_observable(observable)

10
stix2/test/v20/test_pattern_expressions.py
View File

@ -306,10 +306,12 @@ def test_multiple_qualifiers():
def test_set_op():
exp = stix2.ObservationExpression(stix2.IsSubsetComparisonExpression(
"network-traffic:dst_ref.value",
"2001:0db8:dead:beef:0000:0000:0000:0000/64",
))
exp = stix2.ObservationExpression(
stix2.IsSubsetComparisonExpression(
"network-traffic:dst_ref.value",
"2001:0db8:dead:beef:0000:0000:0000:0000/64",
),
)
assert str(exp) == "[network-traffic:dst_ref.value ISSUBSET '2001:0db8:dead:beef:0000:0000:0000:0000/64']"

2
stix2/test/v20/test_utils.py
View File

@ -71,7 +71,7 @@ def test_parse_datetime_invalid(ts):
{"a": 1},
'{"a": 1}',
StringIO(u'{"a": 1}'),
[("a", 1,)],
[("a", 1)],
],
)
def test_get_dict(data):

16
stix2/test/v20/test_versioning.py
View File

@ -46,10 +46,12 @@ def test_making_new_version_with_embedded_object():
**CAMPAIGN_MORE_KWARGS
)
campaign_v2 = campaign_v1.new_version(external_references=[{
"source_name": "capec",
"external_id": "CAPEC-164",
}])
campaign_v2 = campaign_v1.new_version(
external_references=[{
"source_name": "capec",
"external_id": "CAPEC-164",
}],
)
assert campaign_v1.id == campaign_v2.id
assert campaign_v1.created_by_ref == campaign_v2.created_by_ref
@ -237,8 +239,10 @@ def test_remove_custom_stix_property():
mal_nc = stix2.versioning.remove_custom_stix(mal)
assert "x_custom" not in mal_nc
assert (stix2.utils.parse_into_datetime(mal["modified"], precision="millisecond") <
stix2.utils.parse_into_datetime(mal_nc["modified"], precision="millisecond"))
assert (
stix2.utils.parse_into_datetime(mal["modified"], precision="millisecond") <
stix2.utils.parse_into_datetime(mal_nc["modified"], precision="millisecond")
)
def test_remove_custom_stix_object():

12
stix2/test/v21/test_datastore_memory.py
View File

@ -191,11 +191,13 @@ def test_memory_source_get_nonexistant_object(mem_source):
def test_memory_store_all_versions(mem_store):
# Add bundle of items to sink
mem_store.add(dict(
id="bundle--%s" % make_id(),
objects=STIX_OBJS2,
type="bundle",
))
mem_store.add(
dict(
id="bundle--%s" % make_id(),
objects=STIX_OBJS2,
type="bundle",
),
)
resp = mem_store.all_versions("indicator--00000000-0000-4000-8000-000000000001")
assert len(resp) == 3

6
stix2/test/v21/test_indicator.py
View File

@ -20,7 +20,8 @@ EXPECTED_INDICATOR = """{
"valid_from": "1970-01-01T00:00:01Z"
}"""
EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join("""
EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join(
"""
type='indicator',
spec_version='2.1',
id='indicator--a740531e-63ff-4e49-a9e1-a0a3eed0e3e7',
@ -30,7 +31,8 @@ EXPECTED_INDICATOR_REPR = "Indicator(" + " ".join("""
pattern_type='stix',
pattern_version='2.1',
valid_from='1970-01-01T00:00:01Z'
""".split()) + ")"
""".split(),
) + ")"
def test_indicator_with_all_required_properties():

12
stix2/test/v21/test_location.py
View File

@ -19,14 +19,16 @@ EXPECTED_LOCATION_1 = """{
"longitude": 2.3522
}"""
EXPECTED_LOCATION_1_REPR = "Location(" + " ".join("""
EXPECTED_LOCATION_1_REPR = "Location(" + " ".join(
"""
type='location',
spec_version='2.1',
id='location--a6e9345f-5a15-4c29-8bb3-7dcc5d168d64',
created='2016-04-06T20:03:00.000Z',
modified='2016-04-06T20:03:00.000Z',
latitude=48.8566,
longitude=2.3522""".split()) + ")"
longitude=2.3522""".split(),
) + ")"
EXPECTED_LOCATION_2 = """{
"type": "location",
@ -38,13 +40,15 @@ EXPECTED_LOCATION_2 = """{
}
"""
EXPECTED_LOCATION_2_REPR = "Location(" + " ".join("""
EXPECTED_LOCATION_2_REPR = "Location(" + " ".join(
"""
type='location',
spec_version='2.1',
id='location--a6e9345f-5a15-4c29-8bb3-7dcc5d168d64',
created='2016-04-06T20:03:00.000Z',
modified='2016-04-06T20:03:00.000Z',
region='north-america'""".split()) + ")"
region='north-america'""".split(),
) + ")"
def test_location_with_some_required_properties():

92
stix2/test/v21/test_observed_data.py
View File

@ -496,12 +496,14 @@ def test_parse_email_message_not_multipart(data):
def test_parse_file_archive(data):
odata_str = OBJECTS_REGEX.sub('"objects": { %s }' % data, EXPECTED)
odata = stix2.parse(odata_str, version="2.1")
assert all(x in odata.objects["3"].extensions['archive-ext'].contains_refs
for x in [
"file--ecd47d73-15e4-5250-afda-ef8897b22340",
"file--65f2873d-38c2-56b4-bfa5-e3ef21e8a3c3",
"file--ef2d6dca-ec7d-5ab7-8dd9-ec9c0dee0eac",
])
assert all(
x in odata.objects["3"].extensions['archive-ext'].contains_refs
for x in [
"file--ecd47d73-15e4-5250-afda-ef8897b22340",
"file--65f2873d-38c2-56b4-bfa5-e3ef21e8a3c3",
"file--ef2d6dca-ec7d-5ab7-8dd9-ec9c0dee0eac",
]
)
@pytest.mark.parametrize(
@ -904,14 +906,14 @@ def test_file_with_archive_ext_object():
f_obj = stix2.v21.File(
name="foo", extensions={
"archive-ext": {
"contains_refs": [ad, ],
"contains_refs": [ad],
},
},
)
f_ref = stix2.v21.File(
name="foo", extensions={
"archive-ext": {
"contains_refs": [ad.id, ],
"contains_refs": [ad.id],
},
},
)
@ -1229,9 +1231,11 @@ def test_process_example_empty_error():
def test_process_example_empty_with_extensions():
with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo:
stix2.v21.Process(extensions={
"windows-process-ext": {},
})
stix2.v21.Process(
extensions={
"windows-process-ext": {},
},
)
assert excinfo.value.cls == stix2.v21.Process
@ -1276,50 +1280,56 @@ def test_process_example_extensions_empty():
def test_process_example_with_WindowsProcessExt_Object():
p = stix2.v21.Process(extensions={
"windows-process-ext": stix2.v21.WindowsProcessExt(
aslr_enabled=True,
dep_enabled=True,
priority="HIGH_PRIORITY_CLASS",
owner_sid="S-1-5-21-186985262-1144665072-74031268-1309",
), # noqa
})
p = stix2.v21.Process(
extensions={
"windows-process-ext": stix2.v21.WindowsProcessExt(
aslr_enabled=True,
dep_enabled=True,
priority="HIGH_PRIORITY_CLASS",
owner_sid="S-1-5-21-186985262-1144665072-74031268-1309",
), # noqa
},
)
assert p.extensions["windows-process-ext"].dep_enabled
assert p.extensions["windows-process-ext"].owner_sid == "S-1-5-21-186985262-1144665072-74031268-1309"
def test_process_example_with_WindowsServiceExt():
p = stix2.v21.Process(extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
p = stix2.v21.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
},
},
})
)
assert p.extensions["windows-service-ext"].service_name == "sirvizio"
assert p.extensions["windows-service-ext"].service_type == "SERVICE_WIN32_OWN_PROCESS"
def test_process_example_with_WindowsProcessServiceExt():
p = stix2.v21.Process(extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
p = stix2.v21.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
"start_type": "SERVICE_AUTO_START",
"service_type": "SERVICE_WIN32_OWN_PROCESS",
"service_status": "SERVICE_RUNNING",
},
"windows-process-ext": {
"aslr_enabled": True,
"dep_enabled": True,
"priority": "HIGH_PRIORITY_CLASS",
"owner_sid": "S-1-5-21-186985262-1144665072-74031268-1309",
},
},
"windows-process-ext": {
"aslr_enabled": True,
"dep_enabled": True,
"priority": "HIGH_PRIORITY_CLASS",
"owner_sid": "S-1-5-21-186985262-1144665072-74031268-1309",
},
})
)
assert p.extensions["windows-service-ext"].service_name == "sirvizio"
assert p.extensions["windows-service-ext"].service_type == "SERVICE_WIN32_OWN_PROCESS"

7
stix2/test/v21/test_parsing.py
View File

@ -2,8 +2,7 @@ from collections import OrderedDict
import pytest
import stix2
from stix2 import exceptions, parsing, registration, registry
from stix2 import DEFAULT_VERSION, exceptions, parsing, registration, registry
BUNDLE = {
"type": "bundle",
@ -64,7 +63,7 @@ def test_parse_observable_with_version():
assert v in str(obs_obj.__class__)
@pytest.mark.xfail(reason="The default version is not 2.1", condition=stix2.DEFAULT_VERSION != "2.1")
@pytest.mark.xfail(reason="The default version is not 2.1", condition=DEFAULT_VERSION != "2.1")
def test_parse_observable_with_no_version():
observable = {"type": "file", "name": "foo.exe", "spec_version": "2.1"}
obs_obj = parsing.parse_observable(observable)
@ -85,7 +84,7 @@ def test_register_marking_with_version():
assert v in str(registry.STIX2_OBJ_MAPS[v]['markings'][NewMarking1._type])
@pytest.mark.xfail(reason="The default version is not 2.1", condition=stix2.DEFAULT_VERSION != "2.1")
@pytest.mark.xfail(reason="The default version is not 2.1", condition=DEFAULT_VERSION != "2.1")
def test_register_marking_with_no_version():
# Uses default version (2.1 in this case)
class NewMarking2:

14
stix2/test/v21/test_pattern_expressions.py
View File

@ -444,10 +444,12 @@ def test_multiple_qualifiers():
def test_set_op():
exp = stix2.ObservationExpression(stix2.IsSubsetComparisonExpression(
"network-traffic:dst_ref.value",
"2001:0db8:dead:beef:0000:0000:0000:0000/64",
))
exp = stix2.ObservationExpression(
stix2.IsSubsetComparisonExpression(
"network-traffic:dst_ref.value",
"2001:0db8:dead:beef:0000:0000:0000:0000/64",
),
)
assert str(exp) == "[network-traffic:dst_ref.value ISSUBSET '2001:0db8:dead:beef:0000:0000:0000:0000/64']"
@ -712,12 +714,12 @@ def test_parsing_boolean():
def test_parsing_mixed_boolean_expression_1():
patt_obj = create_pattern_object("[a:b = 1 AND a:b = 2 OR a:b = 3]",)
patt_obj = create_pattern_object("[a:b = 1 AND a:b = 2 OR a:b = 3]")
assert str(patt_obj) == "[a:b = 1 AND a:b = 2 OR a:b = 3]"
def test_parsing_mixed_boolean_expression_2():
patt_obj = create_pattern_object("[a:b = 1 OR a:b = 2 AND a:b = 3]",)
patt_obj = create_pattern_object("[a:b = 1 OR a:b = 2 AND a:b = 3]")
assert str(patt_obj) == "[a:b = 1 OR a:b = 2 AND a:b = 3]"

2
stix2/test/v21/test_utils.py
View File

@ -71,7 +71,7 @@ def test_parse_datetime_invalid(ts):
{"a": 1},
'{"a": 1}',
StringIO(u'{"a": 1}'),
[("a", 1,)],
[("a", 1)],
],
)
def test_get_dict(data):

10
stix2/test/v21/test_versioning.py
View File

@ -50,10 +50,12 @@ def test_making_new_version_with_embedded_object():
**CAMPAIGN_MORE_KWARGS
)
campaign_v2 = campaign_v1.new_version(external_references=[{
"source_name": "capec",
"external_id": "CAPEC-164",
}])
campaign_v2 = campaign_v1.new_version(
external_references=[{
"source_name": "capec",
"external_id": "CAPEC-164",
}],
)
assert campaign_v1.id == campaign_v2.id
assert campaign_v1.spec_version == campaign_v2.spec_version

8
stix2/utils.py
View File

@ -71,9 +71,11 @@ def _to_enum(value, enum_type, enum_default=None):
elif isinstance(value, six.string_types):
value = enum_type[value.upper()]
else:
raise TypeError("Not a valid {}: {}".format(
enum_type.__name__, value,
))
raise TypeError(
"Not a valid {}: {}".format(
enum_type.__name__, value,
),
)
return value

128
stix2/v20/observables.py
View File

@ -440,24 +440,28 @@ class SocketExt(_Extension):
('is_blocking', BooleanProperty()),
('is_listening', BooleanProperty()),
(
'protocol_family', EnumProperty(allowed=[
"PF_INET",
"PF_IPX",
"PF_APPLETALK",
"PF_INET6",
"PF_AX25",
"PF_NETROM",
]),
'protocol_family', EnumProperty(
allowed=[
"PF_INET",
"PF_IPX",
"PF_APPLETALK",
"PF_INET6",
"PF_AX25",
"PF_NETROM",
],
),
),
('options', DictionaryProperty(spec_version="2.0")),
(
'socket_type', EnumProperty(allowed=[
"SOCK_STREAM",
"SOCK_DGRAM",
"SOCK_RAW",
"SOCK_RDM",
"SOCK_SEQPACKET",
]),
'socket_type', EnumProperty(
allowed=[
"SOCK_STREAM",
"SOCK_DGRAM",
"SOCK_RAW",
"SOCK_RDM",
"SOCK_SEQPACKET",
],
),
),
('socket_descriptor', IntegerProperty()),
('socket_handle', IntegerProperty()),
@ -537,33 +541,39 @@ class WindowsServiceExt(_Extension):
('display_name', StringProperty()),
('group_name', StringProperty()),
(
'start_type', EnumProperty(allowed=[
"SERVICE_AUTO_START",
"SERVICE_BOOT_START",
"SERVICE_DEMAND_START",
"SERVICE_DISABLED",
"SERVICE_SYSTEM_ALERT",
]),
'start_type', EnumProperty(
allowed=[
"SERVICE_AUTO_START",
"SERVICE_BOOT_START",
"SERVICE_DEMAND_START",
"SERVICE_DISABLED",
"SERVICE_SYSTEM_ALERT",
],
),
),
('service_dll_refs', ListProperty(ObjectReferenceProperty(valid_types='file'))),
(
'service_type', EnumProperty(allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
]),
'service_type', EnumProperty(
allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
],
),
),
(
'service_status', EnumProperty(allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
"SERVICE_RUNNING",
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
]),
'service_status', EnumProperty(
allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
"SERVICE_RUNNING",
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
],
),
),
])
@ -687,21 +697,23 @@ class WindowsRegistryValueType(_STIXBase20):
('name', StringProperty(required=True)),
('data', StringProperty()),
(
'data_type', EnumProperty(allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
"REG_BINARY",
"REG_DWORD",
"REG_DWORD_BIG_ENDIAN",
"REG_LINK",
"REG_MULTI_SZ",
"REG_RESOURCE_LIST",
"REG_FULL_RESOURCE_DESCRIPTION",
"REG_RESOURCE_REQUIREMENTS_LIST",
"REG_QWORD",
"REG_INVALID_TYPE",
]),
'data_type', EnumProperty(
allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
"REG_BINARY",
"REG_DWORD",
"REG_DWORD_BIG_ENDIAN",
"REG_LINK",
"REG_MULTI_SZ",
"REG_RESOURCE_LIST",
"REG_FULL_RESOURCE_DESCRIPTION",
"REG_RESOURCE_REQUIREMENTS_LIST",
"REG_QWORD",
"REG_INVALID_TYPE",
],
),
),
])
@ -790,11 +802,13 @@ def CustomObservable(type='x-custom-observable', properties=None):
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
[('type', TypeProperty(type, spec_version='2.0'))],
properties,
[('extensions', ExtensionsProperty(spec_version="2.0", enclosing_type=type))],
]))
_properties = list(
itertools.chain.from_iterable([
[('type', TypeProperty(type, spec_version='2.0'))],
properties,
[('extensions', ExtensionsProperty(spec_version="2.0", enclosing_type=type))],
]),
)
return _custom_observable_builder(cls, type, _properties, '2.0', _Observable)
return wrapper

38
stix2/v20/sdo.py
View File

@ -356,23 +356,25 @@ def CustomObject(type='x-custom-type', properties=None):
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.0')),
('id', IDProperty(type, spec_version='2.0')),
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.0')),
('created', TimestampProperty(default=lambda: NOW, precision='millisecond')),
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
],
[x for x in properties if not x[0].startswith('x_')],
[
('revoked', BooleanProperty(default=lambda: False)),
('labels', ListProperty(StringProperty)),
('external_references', ListProperty(ExternalReference)),
('object_marking_refs', ListProperty(ReferenceProperty(valid_types='marking-definition', spec_version='2.0'))),
('granular_markings', ListProperty(GranularMarking)),
],
sorted([x for x in properties if x[0].startswith('x_')], key=lambda x: x[0]),
]))
_properties = list(
itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.0')),
('id', IDProperty(type, spec_version='2.0')),
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.0')),
('created', TimestampProperty(default=lambda: NOW, precision='millisecond')),
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
],
[x for x in properties if not x[0].startswith('x_')],
[
('revoked', BooleanProperty(default=lambda: False)),
('labels', ListProperty(StringProperty)),
('external_references', ListProperty(ExternalReference)),
('object_marking_refs', ListProperty(ReferenceProperty(valid_types='marking-definition', spec_version='2.0'))),
('granular_markings', ListProperty(GranularMarking)),
],
sorted([x for x in properties if x[0].startswith('x_')], key=lambda x: x[0]),
]),
)
return _custom_object_builder(cls, type, _properties, '2.0', _DomainObject)
return wrapper

128
stix2/v21/observables.py
View File

@ -505,13 +505,15 @@ class SocketExt(_Extension):
('is_listening', BooleanProperty()),
('options', DictionaryProperty(spec_version='2.1')),
(
'socket_type', EnumProperty(allowed=[
"SOCK_STREAM",
"SOCK_DGRAM",
"SOCK_RAW",
"SOCK_RDM",
"SOCK_SEQPACKET",
]),
'socket_type', EnumProperty(
allowed=[
"SOCK_STREAM",
"SOCK_DGRAM",
"SOCK_RAW",
"SOCK_RDM",
"SOCK_SEQPACKET",
],
),
),
('socket_descriptor', IntegerProperty(min=0)),
('socket_handle', IntegerProperty()),
@ -612,12 +614,14 @@ class WindowsProcessExt(_Extension):
('window_title', StringProperty()),
('startup_info', DictionaryProperty(spec_version='2.1')),
(
'integrity_level', EnumProperty(allowed=[
"low",
"medium",
"high",
"system",
]),
'integrity_level', EnumProperty(
allowed=[
"low",
"medium",
"high",
"system",
],
),
),
])
@ -634,33 +638,39 @@ class WindowsServiceExt(_Extension):
('display_name', StringProperty()),
('group_name', StringProperty()),
(
'start_type', EnumProperty(allowed=[
"SERVICE_AUTO_START",
"SERVICE_BOOT_START",
"SERVICE_DEMAND_START",
"SERVICE_DISABLED",
"SERVICE_SYSTEM_ALERT",
]),
'start_type', EnumProperty(
allowed=[
"SERVICE_AUTO_START",
"SERVICE_BOOT_START",
"SERVICE_DEMAND_START",
"SERVICE_DISABLED",
"SERVICE_SYSTEM_ALERT",
],
),
),
('service_dll_refs', ListProperty(ReferenceProperty(valid_types='file', spec_version="2.1"))),
(
'service_type', EnumProperty(allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
]),
'service_type', EnumProperty(
allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
],
),
),
(
'service_status', EnumProperty(allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
"SERVICE_RUNNING",
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
]),
'service_status', EnumProperty(
allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
"SERVICE_RUNNING",
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
],
),
),
])
@ -808,21 +818,23 @@ class WindowsRegistryValueType(_STIXBase21):
('name', StringProperty()),
('data', StringProperty()),
(
'data_type', EnumProperty(allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
"REG_BINARY",
"REG_DWORD",
"REG_DWORD_BIG_ENDIAN",
"REG_LINK",
"REG_MULTI_SZ",
"REG_RESOURCE_LIST",
"REG_FULL_RESOURCE_DESCRIPTION",
"REG_RESOURCE_REQUIREMENTS_LIST",
"REG_QWORD",
"REG_INVALID_TYPE",
]),
'data_type', EnumProperty(
allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
"REG_BINARY",
"REG_DWORD",
"REG_DWORD_BIG_ENDIAN",
"REG_LINK",
"REG_MULTI_SZ",
"REG_RESOURCE_LIST",
"REG_FULL_RESOURCE_DESCRIPTION",
"REG_RESOURCE_REQUIREMENTS_LIST",
"REG_QWORD",
"REG_INVALID_TYPE",
],
),
),
])
@ -935,13 +947,15 @@ def CustomObservable(type='x-custom-observable', properties=None, id_contrib_pro
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
[('type', TypeProperty(type, spec_version='2.1'))],
[('spec_version', StringProperty(fixed='2.1'))],
[('id', IDProperty(type, spec_version='2.1'))],
properties,
[('extensions', ExtensionsProperty(spec_version='2.1', enclosing_type=type))],
]))
_properties = list(
itertools.chain.from_iterable([
[('type', TypeProperty(type, spec_version='2.1'))],
[('spec_version', StringProperty(fixed='2.1'))],
[('id', IDProperty(type, spec_version='2.1'))],
properties,
[('extensions', ExtensionsProperty(spec_version='2.1', enclosing_type=type))],
]),
)
return _custom_observable_builder(cls, type, _properties, '2.1', _Observable, id_contrib_props)
return wrapper

44
stix2/v21/sdo.py
View File

@ -789,27 +789,29 @@ def CustomObject(type='x-custom-type', properties=None):
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.1')),
('spec_version', StringProperty(fixed='2.1')),
('id', IDProperty(type, spec_version='2.1')),
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.1')),
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
],
[x for x in properties if not x[0].startswith('x_')],
[
('revoked', BooleanProperty(default=lambda: False)),
('labels', ListProperty(StringProperty)),
('confidence', IntegerProperty()),
('lang', StringProperty()),
('external_references', ListProperty(ExternalReference)),
('object_marking_refs', ListProperty(ReferenceProperty(valid_types='marking-definition', spec_version='2.1'))),
('granular_markings', ListProperty(GranularMarking)),
],
sorted([x for x in properties if x[0].startswith('x_')], key=lambda x: x[0]),
]))
_properties = list(
itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.1')),
('spec_version', StringProperty(fixed='2.1')),
('id', IDProperty(type, spec_version='2.1')),
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.1')),
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
],
[x for x in properties if not x[0].startswith('x_')],
[
('revoked', BooleanProperty(default=lambda: False)),
('labels', ListProperty(StringProperty)),
('confidence', IntegerProperty()),
('lang', StringProperty()),
('external_references', ListProperty(ExternalReference)),
('object_marking_refs', ListProperty(ReferenceProperty(valid_types='marking-definition', spec_version='2.1'))),
('granular_markings', ListProperty(GranularMarking)),
],
sorted([x for x in properties if x[0].startswith('x_')], key=lambda x: x[0]),
]),
)
return _custom_object_builder(cls, type, _properties, '2.1', _DomainObject)
return wrapper

21
stix2/workbench.py
View File

@ -21,7 +21,7 @@
"""
import functools
import stix2
from . import AttackPattern as _AttackPattern
from . import Campaign as _Campaign
from . import CourseOfAction as _CourseOfAction
@ -34,34 +34,37 @@ from . import Location as _Location
from . import Malware as _Malware
from . import MalwareAnalysis as _MalwareAnalysis
from . import Note as _Note
from . import OBJ_MAP
from . import ObservedData as _ObservedData
from . import Opinion as _Opinion
from . import Report as _Report
from . import ThreatActor as _ThreatActor
from . import Tool as _Tool
from . import Vulnerability as _Vulnerability
from . import ( # noqa: F401
from .version import DEFAULT_VERSION
from . import ( # noqa: F401 isort:skip
AlternateDataStream, ArchiveExt, Artifact, AutonomousSystem,
Bundle, CustomExtension, CustomMarking, CustomObservable,
Directory, DomainName, EmailAddress, EmailMessage,
EmailMIMEComponent, Environment, ExternalReference, File,
FileSystemSource, Filter, GranularMarking, HTTPRequestExt,
ICMPExt, IPv4Address, IPv6Address, KillChainPhase, LanguageContent, MACAddress,
MarkingDefinition, MemoryStore, Mutex, NetworkTraffic, NTFSExt,
parse_observable, PDFExt, Process, RasterImageExt, Relationship,
ICMPExt, IPv4Address, IPv6Address, KillChainPhase, LanguageContent,
MACAddress, MarkingDefinition, MemoryStore, Mutex, NetworkTraffic,
NTFSExt, parse_observable, PDFExt, Process, RasterImageExt, Relationship,
Sighting, SocketExt, Software, StatementMarking,
TAXIICollectionSource, TCPExt, TLP_AMBER, TLP_GREEN, TLP_RED,
TLP_WHITE, TLPMarking, UNIXAccountExt, URL, UserAccount,
WindowsPEBinaryExt, WindowsPEOptionalHeaderType,
WindowsPESection, WindowsProcessExt, WindowsRegistryKey,
WindowsRegistryValueType, WindowsServiceExt, X509Certificate,
X509V3ExtensionsType
X509V3ExtensionsType,
)
from .datastore.filters import FilterSet
from .datastore.filters import FilterSet # isort:skip
# Enable some adaptation to the current default supported STIX version.
_STIX_VID = "v" + stix2.DEFAULT_VERSION.replace(".", "")
_STIX_VID = "v" + DEFAULT_VERSION.replace(".", "")
# Use an implicit MemoryStore
@ -161,7 +164,7 @@ def _setup_workbench():
# Add our new "class" to this module's globals and to the library-wide
# mapping. This allows parse() to use the wrapped classes.
globals()[obj_type.__name__] = factory_func
stix2.OBJ_MAP[obj_type._type] = factory_func
OBJ_MAP[obj_type._type] = factory_func
_setup_workbench()

33
tox.ini
View File

@ -1,5 +1,5 @@
[tox]
envlist = py35,py36,py37,py38,style,isort-check,packaging
envlist = py36,py37,py38,py39,packaging,pre-commit-check
[testenv]
deps =
@ -15,33 +15,24 @@ deps =
commands =
python -m pytest --cov=stix2 stix2/test/ --cov-report term-missing -W ignore::stix2.exceptions.STIXDeprecationWarning
passenv = CI TRAVIS TRAVIS_*
[testenv:style]
deps =
flake8
commands =
flake8
[flake8]
max-line-length = 160
[testenv:isort-check]
deps = isort
commands =
isort stix2 examples --df
isort stix2 examples -c
passenv = GITHUB_*
[testenv:packaging]
deps =
twine
commands =
python setup.py bdist_wheel --universal
python setup.py sdist bdist_wheel --universal
twine check dist/*
[travis]
[testenv:pre-commit-check]
deps =
pre-commit
commands =
pre-commit run --all-files
[gh-actions]
python =
3.5: py35
3.6: py36
3.7: py37
3.8: py38, style, packaging
3.8: py38
3.9: py39, packaging, pre-commit-check