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

Merge branch 'master' of github.com:oasis-open/cti-python-stix2 into dev-extensions-proposal

pull/1/head
Emmanuelle Vargas-Gonzalez 2021-01-15 13:01:30 -05:00
parent a86b21dbf7 9ed62699ca
commit 82390ba648
33 changed files with 488 additions and 370 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

@ -178,8 +178,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
@ -208,8 +210,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("_"):

12
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(
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)

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.:

12
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(
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.

22
stix2/parsing.py
View File

@ -197,8 +197,10 @@ 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)
@ -294,8 +296,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,
@ -310,8 +316,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,

6
stix2/pattern_visitor.py
View File

@ -261,11 +261,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(
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)

12
stix2/properties.py
View File

@ -247,9 +247,11 @@ class ListProperty(Property):
valid = self.contained(**item)
else:
raise ValueError("Can't create a {} out of {}".format(
raise ValueError(
"Can't create a {} out of {}".format(
self.contained._type, str(item),
))
),
)
result.append(valid)
@ -690,8 +692,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

6
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(
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

12
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=[{
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=[{
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():

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

@ -1180,21 +1180,24 @@ def test_process_example_extensions_empty():
def test_process_example_with_WindowsProcessExt_Object():
p = stix2.v20.Process(extensions={
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={
p = stix2.v20.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
@ -1202,14 +1205,16 @@ def test_process_example_with_WindowsServiceExt():
"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={
p = stix2.v20.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
@ -1223,7 +1228,8 @@ def test_process_example_with_WindowsProcessServiceExt():
"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"

6
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(
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):

12
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=[{
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():

6
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(
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():

34
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
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={
stix2.v21.Process(
extensions={
"windows-process-ext": {},
})
},
)
assert excinfo.value.cls == stix2.v21.Process
@ -1276,21 +1280,24 @@ def test_process_example_extensions_empty():
def test_process_example_with_WindowsProcessExt_Object():
p = stix2.v21.Process(extensions={
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={
p = stix2.v21.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
@ -1298,14 +1305,16 @@ def test_process_example_with_WindowsServiceExt():
"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={
p = stix2.v21.Process(
extensions={
"windows-service-ext": {
"service_name": "sirvizio",
"display_name": "Sirvizio",
@ -1319,7 +1328,8 @@ def test_process_example_with_WindowsProcessServiceExt():
"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"

10
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(
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):

6
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=[{
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

6
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(
raise TypeError(
"Not a valid {}: {}".format(
enum_type.__name__, value,
))
),
)
return value

42
stix2/v20/observables.py
View File

@ -440,24 +440,28 @@ class SocketExt(_Extension):
('is_blocking', BooleanProperty()),
('is_listening', BooleanProperty()),
(
'protocol_family', EnumProperty(allowed=[
'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=[
'socket_type', EnumProperty(
allowed=[
"SOCK_STREAM",
"SOCK_DGRAM",
"SOCK_RAW",
"SOCK_RDM",
"SOCK_SEQPACKET",
]),
],
),
),
('socket_descriptor', IntegerProperty()),
('socket_handle', IntegerProperty()),
@ -537,25 +541,30 @@ class WindowsServiceExt(_Extension):
('display_name', StringProperty()),
('group_name', StringProperty()),
(
'start_type', EnumProperty(allowed=[
'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_type', EnumProperty(
allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
]),
],
),
),
(
'service_status', EnumProperty(allowed=[
'service_status', EnumProperty(
allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
@ -563,7 +572,8 @@ class WindowsServiceExt(_Extension):
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
]),
],
),
),
])
@ -687,7 +697,8 @@ class WindowsRegistryValueType(_STIXBase20):
('name', StringProperty(required=True)),
('data', StringProperty()),
(
'data_type', EnumProperty(allowed=[
'data_type', EnumProperty(
allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
@ -701,7 +712,8 @@ class WindowsRegistryValueType(_STIXBase20):
"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([
_properties = list(
itertools.chain.from_iterable([
[('type', TypeProperty(type, spec_version='2.0'))],
properties,
[('extensions', ExtensionsProperty(spec_version='2.0'))],
]))
]),
)
return _custom_observable_builder(cls, type, _properties, '2.0', _Observable)
return wrapper

6
stix2/v20/sdo.py
View File

@ -356,7 +356,8 @@ def CustomObject(type='x-custom-type', properties=None):
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
_properties = list(
itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.0')),
('id', IDProperty(type, spec_version='2.0')),
@ -373,6 +374,7 @@ def CustomObject(type='x-custom-type', properties=None):
('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

42
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=[
'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=[
'integrity_level', EnumProperty(
allowed=[
"low",
"medium",
"high",
"system",
]),
],
),
),
])
@ -634,25 +638,30 @@ class WindowsServiceExt(_Extension):
('display_name', StringProperty()),
('group_name', StringProperty()),
(
'start_type', EnumProperty(allowed=[
'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_type', EnumProperty(
allowed=[
"SERVICE_KERNEL_DRIVER",
"SERVICE_FILE_SYSTEM_DRIVER",
"SERVICE_WIN32_OWN_PROCESS",
"SERVICE_WIN32_SHARE_PROCESS",
]),
],
),
),
(
'service_status', EnumProperty(allowed=[
'service_status', EnumProperty(
allowed=[
"SERVICE_CONTINUE_PENDING",
"SERVICE_PAUSE_PENDING",
"SERVICE_PAUSED",
@ -660,7 +669,8 @@ class WindowsServiceExt(_Extension):
"SERVICE_START_PENDING",
"SERVICE_STOP_PENDING",
"SERVICE_STOPPED",
]),
],
),
),
])
@ -808,7 +818,8 @@ class WindowsRegistryValueType(_STIXBase21):
('name', StringProperty()),
('data', StringProperty()),
(
'data_type', EnumProperty(allowed=[
'data_type', EnumProperty(
allowed=[
"REG_NONE",
"REG_SZ",
"REG_EXPAND_SZ",
@ -822,7 +833,8 @@ class WindowsRegistryValueType(_STIXBase21):
"REG_RESOURCE_REQUIREMENTS_LIST",
"REG_QWORD",
"REG_INVALID_TYPE",
]),
],
),
),
])
@ -935,14 +947,16 @@ def CustomObservable(type='x-custom-observable', properties=None, id_contrib_pro
"""
def wrapper(cls):
_properties = list(itertools.chain.from_iterable([
_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'))],
[('spec_version', StringProperty(fixed='2.1'))],
properties,
[('extensions', ExtensionsProperty(spec_version='2.1'))],
]))
]),
)
if extension_name:
@CustomExtension(type=extension_name, properties=properties)
class NameExtension:

6
stix2/v21/sdo.py
View File

@ -837,7 +837,8 @@ def CustomObject(type='x-custom-type', properties=None, extension_name=None):
"""
def wrapper(cls):
extension_properties = [x for x in properties if not x[0].startswith('x_')]
_properties = list(itertools.chain.from_iterable([
_properties = list(
itertools.chain.from_iterable([
[
('type', TypeProperty(type, spec_version='2.1')),
('spec_version', StringProperty(fixed='2.1')),
@ -858,7 +859,8 @@ def CustomObject(type='x-custom-type', properties=None, extension_name=None):
('extensions', ExtensionsProperty(spec_version='2.1')),
],
sorted([x for x in properties if x[0].startswith('x_')], key=lambda x: x[0]),
]))
]),
)
if extension_name:
@observables.CustomExtension(type=extension_name, properties=extension_properties)
class NameExtension:

15
stix2/workbench.py
View File

@ -21,7 +21,9 @@
"""
import functools
import stix2
from . import AttackPattern as _AttackPattern
from . import Campaign as _Campaign
from . import CourseOfAction as _CourseOfAction
@ -40,24 +42,25 @@ 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 . 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.

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