Chris Lenk
GitHub
commit
9145bdf5e8
1
setup.py
1
setup.py
|
|
@ -51,6 +51,7 @@ setup(
|
|||
keywords='stix stix2 json cti cyber threat intelligence',
|
||||
packages=find_packages(exclude=['*.test', '*.test.*']),
|
||||
install_requires=[
|
||||
'enum34 ; python_version<"3.4"',
|
||||
'python-dateutil',
|
||||
'pytz',
|
||||
'requests',
|
||||
|
|
|
|||
|
|
@ -323,12 +323,16 @@ class BooleanProperty(Property):
|
|||
|
||||
class TimestampProperty(Property):
|
||||
|
||||
def __init__(self, precision=None, **kwargs):
|
||||
def __init__(self, precision="any", precision_constraint="exact", **kwargs):
|
||||
self.precision = precision
|
||||
self.precision_constraint = precision_constraint
|
||||
|
||||
super(TimestampProperty, self).__init__(**kwargs)
|
||||
|
||||
def clean(self, value):
|
||||
return parse_into_datetime(value, self.precision)
|
||||
return parse_into_datetime(
|
||||
value, self.precision, self.precision_constraint,
|
||||
)
|
||||
|
||||
|
||||
class DictionaryProperty(Property):
|
||||
|
|
|
|||
|
|
@ -88,8 +88,8 @@ def test_attack_pattern_invalid_labels():
|
|||
def test_overly_precise_timestamps():
|
||||
ap = stix2.v21.AttackPattern(
|
||||
id=ATTACK_PATTERN_ID,
|
||||
created="2016-05-12T08:17:27.0000342Z",
|
||||
modified="2016-05-12T08:17:27.000287Z",
|
||||
created="2016-05-12T08:17:27.000000342Z",
|
||||
modified="2016-05-12T08:17:27.000000287Z",
|
||||
name="Spear Phishing",
|
||||
external_references=[{
|
||||
"source_name": "capec",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,164 @@
|
|||
import datetime
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
import stix2
|
||||
from stix2.utils import (
|
||||
Precision, PrecisionConstraint, STIXdatetime, _to_enum, format_datetime,
|
||||
parse_into_datetime,
|
||||
)
|
||||
|
||||
_DT = datetime.datetime.utcnow()
|
||||
# intentionally omit microseconds from the following. We add it in as
|
||||
# needed for each test.
|
||||
_DT_STR = _DT.strftime("%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value, enum_type, enum_default, enum_expected", [
|
||||
("second", Precision, None, Precision.SECOND),
|
||||
(
|
||||
"eXaCt", PrecisionConstraint, PrecisionConstraint.MIN,
|
||||
PrecisionConstraint.EXACT,
|
||||
),
|
||||
(None, Precision, Precision.MILLISECOND, Precision.MILLISECOND),
|
||||
(Precision.ANY, Precision, None, Precision.ANY),
|
||||
],
|
||||
)
|
||||
def test_to_enum(value, enum_type, enum_default, enum_expected):
|
||||
result = _to_enum(value, enum_type, enum_default)
|
||||
assert result == enum_expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value, err_type", [
|
||||
("foo", KeyError),
|
||||
(1, TypeError),
|
||||
(PrecisionConstraint.EXACT, TypeError),
|
||||
(None, TypeError),
|
||||
],
|
||||
)
|
||||
def test_to_enum_errors(value, err_type):
|
||||
with pytest.raises(err_type):
|
||||
_to_enum(value, Precision)
|
||||
|
||||
|
||||
@pytest.mark.xfail(
|
||||
sys.version_info[:2] == (3, 6), strict=True,
|
||||
reason="https://bugs.python.org/issue32404",
|
||||
)
|
||||
def test_stix_datetime_now():
|
||||
dt = STIXdatetime.utcnow()
|
||||
assert dt.precision is Precision.ANY
|
||||
assert dt.precision_constraint is PrecisionConstraint.EXACT
|
||||
|
||||
|
||||
def test_stix_datetime():
|
||||
dt = datetime.datetime.utcnow()
|
||||
|
||||
sdt = STIXdatetime(dt, precision=Precision.SECOND)
|
||||
assert sdt.precision is Precision.SECOND
|
||||
assert sdt == dt
|
||||
|
||||
sdt = STIXdatetime(
|
||||
dt,
|
||||
precision_constraint=PrecisionConstraint.EXACT,
|
||||
)
|
||||
assert sdt.precision_constraint is PrecisionConstraint.EXACT
|
||||
assert sdt == dt
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"us, precision, precision_constraint, expected_truncated_us", [
|
||||
(123456, Precision.ANY, PrecisionConstraint.EXACT, 123456),
|
||||
(123456, Precision.SECOND, PrecisionConstraint.EXACT, 0),
|
||||
(123456, Precision.SECOND, PrecisionConstraint.MIN, 123456),
|
||||
(123456, Precision.MILLISECOND, PrecisionConstraint.EXACT, 123000),
|
||||
(123456, Precision.MILLISECOND, PrecisionConstraint.MIN, 123456),
|
||||
(1234, Precision.MILLISECOND, PrecisionConstraint.EXACT, 1000),
|
||||
(123, Precision.MILLISECOND, PrecisionConstraint.EXACT, 0),
|
||||
],
|
||||
)
|
||||
def test_parse_datetime(
|
||||
us, precision, precision_constraint, expected_truncated_us,
|
||||
):
|
||||
|
||||
# complete the datetime string with microseconds
|
||||
dt_us_str = "{}.{:06d}Z".format(_DT_STR, us)
|
||||
|
||||
sdt = parse_into_datetime(
|
||||
dt_us_str,
|
||||
precision=precision,
|
||||
precision_constraint=precision_constraint,
|
||||
)
|
||||
|
||||
assert sdt.precision is precision
|
||||
assert sdt.precision_constraint is precision_constraint
|
||||
assert sdt.microsecond == expected_truncated_us
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"us, precision, precision_constraint, expected_us_str", [
|
||||
(123456, Precision.ANY, PrecisionConstraint.EXACT, ".123456"),
|
||||
(123456, Precision.SECOND, PrecisionConstraint.EXACT, ""),
|
||||
(123456, Precision.SECOND, PrecisionConstraint.MIN, ".123456"),
|
||||
(123456, Precision.MILLISECOND, PrecisionConstraint.EXACT, ".123"),
|
||||
(123456, Precision.MILLISECOND, PrecisionConstraint.MIN, ".123456"),
|
||||
(0, Precision.SECOND, PrecisionConstraint.MIN, ""),
|
||||
(0, Precision.MILLISECOND, PrecisionConstraint.MIN, ".000"),
|
||||
(0, Precision.MILLISECOND, PrecisionConstraint.EXACT, ".000"),
|
||||
(1000, Precision.MILLISECOND, PrecisionConstraint.EXACT, ".001"),
|
||||
(10000, Precision.MILLISECOND, PrecisionConstraint.EXACT, ".010"),
|
||||
(100000, Precision.MILLISECOND, PrecisionConstraint.EXACT, ".100"),
|
||||
(1000, Precision.ANY, PrecisionConstraint.EXACT, ".001"),
|
||||
(10000, Precision.ANY, PrecisionConstraint.EXACT, ".01"),
|
||||
(100000, Precision.ANY, PrecisionConstraint.EXACT, ".1"),
|
||||
(1001, Precision.MILLISECOND, PrecisionConstraint.MIN, ".001001"),
|
||||
(10010, Precision.MILLISECOND, PrecisionConstraint.MIN, ".01001"),
|
||||
(100100, Precision.MILLISECOND, PrecisionConstraint.MIN, ".1001"),
|
||||
],
|
||||
)
|
||||
def test_format_datetime(us, precision, precision_constraint, expected_us_str):
|
||||
|
||||
dt = _DT.replace(microsecond=us)
|
||||
expected_dt_str = "{}{}Z".format(_DT_STR, expected_us_str)
|
||||
|
||||
sdt = STIXdatetime(
|
||||
dt,
|
||||
precision=precision,
|
||||
precision_constraint=precision_constraint,
|
||||
)
|
||||
s = format_datetime(sdt)
|
||||
assert s == expected_dt_str
|
||||
|
||||
|
||||
def test_sdo_extra_precision():
|
||||
# add extra precision for "modified", ensure it's not lost
|
||||
identity_dict = {
|
||||
"type": "identity",
|
||||
"id": "identity--4a457eeb-6639-4aa3-be81-5930a3000c39",
|
||||
"created": "2015-12-21T19:59:11.000Z",
|
||||
"modified": "2015-12-21T19:59:11.0001Z",
|
||||
"name": "John Smith",
|
||||
"identity_class": "individual",
|
||||
"spec_version": "2.1",
|
||||
}
|
||||
|
||||
identity_obj = stix2.parse(identity_dict)
|
||||
assert identity_obj.modified.microsecond == 100
|
||||
assert identity_obj.modified.precision is Precision.MILLISECOND
|
||||
assert identity_obj.modified.precision_constraint is PrecisionConstraint.MIN
|
||||
|
||||
identity_str = identity_obj.serialize(pretty=True)
|
||||
|
||||
# ensure precision is retained in JSON
|
||||
assert identity_str == """{
|
||||
"type": "identity",
|
||||
"spec_version": "2.1",
|
||||
"id": "identity--4a457eeb-6639-4aa3-be81-5930a3000c39",
|
||||
"created": "2015-12-21T19:59:11.000Z",
|
||||
"modified": "2015-12-21T19:59:11.0001Z",
|
||||
"name": "John Smith",
|
||||
"identity_class": "individual"
|
||||
}"""
|
||||
|
|
@ -1,6 +1,9 @@
|
|||
import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
import stix2
|
||||
import stix2.utils
|
||||
|
||||
from .constants import CAMPAIGN_MORE_KWARGS
|
||||
|
||||
|
|
@ -236,8 +239,7 @@ def test_remove_custom_stix_property():
|
|||
mal_nc = stix2.utils.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 mal["modified"] < mal_nc["modified"]
|
||||
|
||||
|
||||
def test_remove_custom_stix_object():
|
||||
|
|
@ -264,3 +266,33 @@ def test_remove_custom_stix_no_custom():
|
|||
assert len(campaign_v1.keys()) == len(campaign_v2.keys())
|
||||
assert campaign_v1.id == campaign_v2.id
|
||||
assert campaign_v1.description == campaign_v2.description
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"old, candidate_new, expected_new, use_stix21", [
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.001Z", "1999-08-15T00:19:07.001Z", False),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.001Z", False),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:06.000Z", "1999-08-15T00:19:07.001Z", False),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:06.999Z", "1999-08-15T00:19:07.001Z", False),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.0001Z", "1999-08-15T00:19:07.001Z", False),
|
||||
("1999-08-15T00:19:07.999Z", "1999-08-15T00:19:07.9999Z", "1999-08-15T00:19:08.000Z", False),
|
||||
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.001Z", "1999-08-15T00:19:07.001Z", True),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.000001Z", True),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:06.000Z", "1999-08-15T00:19:07.000001Z", True),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:06.999999Z", "1999-08-15T00:19:07.000001Z", True),
|
||||
("1999-08-15T00:19:07.000Z", "1999-08-15T00:19:07.000001Z", "1999-08-15T00:19:07.000001Z", True),
|
||||
("1999-08-15T00:19:07.999Z", "1999-08-15T00:19:07.999999Z", "1999-08-15T00:19:07.999999Z", True),
|
||||
],
|
||||
)
|
||||
def test_fudge_modified(old, candidate_new, expected_new, use_stix21):
|
||||
old_dt = datetime.datetime.strptime(old, "%Y-%m-%dT%H:%M:%S.%fZ")
|
||||
candidate_new_dt = datetime.datetime.strptime(
|
||||
candidate_new, "%Y-%m-%dT%H:%M:%S.%fZ",
|
||||
)
|
||||
expected_new_dt = datetime.datetime.strptime(
|
||||
expected_new, "%Y-%m-%dT%H:%M:%S.%fZ",
|
||||
)
|
||||
|
||||
fudged = stix2.utils._fudge_modified(old_dt, candidate_new_dt, use_stix21)
|
||||
assert fudged == expected_new_dt
|
||||
|
|
|
|||
246
stix2/utils.py
246
stix2/utils.py
|
|
@ -6,10 +6,12 @@ except ImportError:
|
|||
from collections import Mapping
|
||||
import copy
|
||||
import datetime as dt
|
||||
import enum
|
||||
import json
|
||||
|
||||
from dateutil import parser
|
||||
import pytz
|
||||
import six
|
||||
|
||||
import stix2.base
|
||||
|
||||
|
|
@ -29,9 +31,79 @@ TYPE_REGEX = r'^\-?[a-z0-9]+(-[a-z0-9]+)*\-?$'
|
|||
SCO21_EXT_REGEX = r'^\-?[a-z0-9]+(-[a-z0-9]+)*\-ext$'
|
||||
|
||||
|
||||
class Precision(enum.Enum):
|
||||
"""
|
||||
Timestamp format precisions.
|
||||
"""
|
||||
# auto() wasn't introduced until Python 3.6.
|
||||
ANY = 1
|
||||
SECOND = 2
|
||||
MILLISECOND = 3
|
||||
|
||||
|
||||
class PrecisionConstraint(enum.Enum):
|
||||
"""
|
||||
Timestamp precision constraints. These affect how the Precision
|
||||
values are applied when formatting a timestamp.
|
||||
|
||||
These constraints don't really make sense with the ANY precision, so they
|
||||
have no effect in that case.
|
||||
"""
|
||||
EXACT = 1 # format must have exactly the given precision
|
||||
MIN = 2 # format must have at least the given precision
|
||||
# no need for a MAX constraint yet
|
||||
|
||||
|
||||
def _to_enum(value, enum_type, enum_default=None):
|
||||
"""
|
||||
Detect and convert strings to enums and None to a default enum. This
|
||||
allows use of strings and None in APIs, while enforcing the enum type: if
|
||||
you use a string, it must name a valid enum value. This implementation is
|
||||
case-insensitive.
|
||||
|
||||
:param value: A value to be interpreted as an enum (string, Enum instance,
|
||||
or None). If an Enum instance, it must be an instance of enum_type.
|
||||
:param enum_type: The enum type which strings will be interpreted against
|
||||
:param enum_default: The default enum to use if value is None. Must be
|
||||
an instance of enum_type, or None. If None, you are disallowing a
|
||||
default and requiring that value be non-None.
|
||||
:return: An instance of enum_type
|
||||
:raises TypeError: If value was neither an instance of enum_type, None, nor
|
||||
a string
|
||||
:raises KeyError: If value was a string which couldn't be interpreted as an
|
||||
enum value from enum_type
|
||||
"""
|
||||
assert enum_default is None or isinstance(enum_default, enum_type)
|
||||
|
||||
if not isinstance(value, enum_type):
|
||||
if value is None and enum_default is not None:
|
||||
value = enum_default
|
||||
elif isinstance(value, six.string_types):
|
||||
value = enum_type[value.upper()]
|
||||
else:
|
||||
raise TypeError("Not a valid {}: {}".format(
|
||||
enum_type.__name__, value,
|
||||
))
|
||||
|
||||
return value
|
||||
|
||||
|
||||
class STIXdatetime(dt.datetime):
|
||||
"""
|
||||
Bundle a datetime with some format-related metadata, so that JSON
|
||||
serialization has the info it needs to produce compliant timestamps.
|
||||
"""
|
||||
|
||||
def __new__(cls, *args, **kwargs):
|
||||
precision = kwargs.pop('precision', None)
|
||||
precision = _to_enum(
|
||||
kwargs.pop("precision", Precision.ANY),
|
||||
Precision,
|
||||
)
|
||||
precision_constraint = _to_enum(
|
||||
kwargs.pop("precision_constraint", PrecisionConstraint.EXACT),
|
||||
PrecisionConstraint,
|
||||
)
|
||||
|
||||
if isinstance(args[0], dt.datetime): # Allow passing in a datetime object
|
||||
dttm = args[0]
|
||||
args = (
|
||||
|
|
@ -41,6 +113,7 @@ class STIXdatetime(dt.datetime):
|
|||
# self will be an instance of STIXdatetime, not dt.datetime
|
||||
self = dt.datetime.__new__(cls, *args, **kwargs)
|
||||
self.precision = precision
|
||||
self.precision_constraint = precision_constraint
|
||||
return self
|
||||
|
||||
def __repr__(self):
|
||||
|
|
@ -90,7 +163,7 @@ def format_datetime(dttm):
|
|||
2. Convert to UTC
|
||||
3. Format in ISO format
|
||||
4. Ensure correct precision
|
||||
a. Add subsecond value if non-zero and precision not defined
|
||||
a. Add subsecond value if warranted, according to precision settings
|
||||
5. Add "Z"
|
||||
|
||||
"""
|
||||
|
|
@ -101,20 +174,74 @@ def format_datetime(dttm):
|
|||
else:
|
||||
zoned = dttm.astimezone(pytz.utc)
|
||||
ts = zoned.strftime('%Y-%m-%dT%H:%M:%S')
|
||||
ms = zoned.strftime('%f')
|
||||
precision = getattr(dttm, 'precision', None)
|
||||
if precision == 'second':
|
||||
pass # Already precise to the second
|
||||
elif precision == 'millisecond':
|
||||
ts = ts + '.' + ms[:3]
|
||||
elif zoned.microsecond > 0:
|
||||
ts = ts + '.' + ms.rstrip('0')
|
||||
return ts + 'Z'
|
||||
precision = getattr(dttm, 'precision', Precision.ANY)
|
||||
precision_constraint = getattr(
|
||||
dttm, 'precision_constraint', PrecisionConstraint.EXACT,
|
||||
)
|
||||
|
||||
frac_seconds_str = ""
|
||||
if precision == Precision.ANY:
|
||||
# No need to truncate; ignore constraint
|
||||
if zoned.microsecond:
|
||||
frac_seconds_str = "{:06d}".format(zoned.microsecond).rstrip("0")
|
||||
|
||||
elif precision == Precision.SECOND:
|
||||
if precision_constraint == PrecisionConstraint.MIN:
|
||||
# second precision, or better. Winds up being the same as ANY:
|
||||
# just use all our digits
|
||||
if zoned.microsecond:
|
||||
frac_seconds_str = "{:06d}".format(zoned.microsecond)\
|
||||
.rstrip("0")
|
||||
# exact: ignore microseconds entirely
|
||||
|
||||
else:
|
||||
# precision == millisecond
|
||||
if precision_constraint == PrecisionConstraint.EXACT:
|
||||
# can't rstrip() here or we may lose precision
|
||||
frac_seconds_str = "{:06d}".format(zoned.microsecond)[:3]
|
||||
|
||||
else:
|
||||
# millisecond precision, or better. So we can rstrip() zeros, but
|
||||
# only to a length of at least 3 digits (ljust() adds zeros back,
|
||||
# if it stripped too far.)
|
||||
frac_seconds_str = "{:06d}"\
|
||||
.format(zoned.microsecond)\
|
||||
.rstrip("0")\
|
||||
.ljust(3, "0")
|
||||
|
||||
ts = "{}{}{}Z".format(
|
||||
ts,
|
||||
"." if frac_seconds_str else "",
|
||||
frac_seconds_str,
|
||||
)
|
||||
|
||||
return ts
|
||||
|
||||
|
||||
def parse_into_datetime(value, precision=None):
|
||||
"""Parse a value into a valid STIX timestamp object.
|
||||
def parse_into_datetime(
|
||||
value, precision=Precision.ANY,
|
||||
precision_constraint=PrecisionConstraint.EXACT,
|
||||
):
|
||||
"""
|
||||
Parse a value into a valid STIX timestamp object. Also, optionally adjust
|
||||
precision of fractional seconds. This allows alignment with JSON
|
||||
serialization requirements, and helps ensure we're not using extra
|
||||
precision which would be lost upon JSON serialization. The precision
|
||||
info will be embedded in the returned object, so that JSON serialization
|
||||
will format it correctly.
|
||||
|
||||
:param value: A datetime.datetime or datetime.date instance, or a string
|
||||
:param precision: A precision value: either an instance of the Precision
|
||||
enum, or a string naming one of the enum values (case-insensitive)
|
||||
:param precision_constraint: A precision constraint value: either an
|
||||
instance of the PrecisionConstraint enum, or a string naming one of
|
||||
the enum values (case-insensitive)
|
||||
:return: A STIXdatetime instance, which is a datetime but also carries the
|
||||
precision info necessary to properly JSON-serialize it.
|
||||
"""
|
||||
precision = _to_enum(precision, Precision)
|
||||
precision_constraint = _to_enum(precision_constraint, PrecisionConstraint)
|
||||
|
||||
if isinstance(value, dt.date):
|
||||
if hasattr(value, 'hour'):
|
||||
ts = value
|
||||
|
|
@ -138,20 +265,23 @@ def parse_into_datetime(value, precision=None):
|
|||
ts = pytz.utc.localize(parsed)
|
||||
|
||||
# Ensure correct precision
|
||||
if not precision:
|
||||
return STIXdatetime(ts, precision=precision)
|
||||
ms = ts.microsecond
|
||||
if precision == 'second':
|
||||
ts = ts.replace(microsecond=0)
|
||||
elif precision == 'millisecond':
|
||||
ms_len = len(str(ms))
|
||||
if ms_len > 3:
|
||||
# Truncate to millisecond precision
|
||||
factor = 10 ** (ms_len - 3)
|
||||
ts = ts.replace(microsecond=(ts.microsecond // factor) * factor)
|
||||
else:
|
||||
if precision == Precision.SECOND:
|
||||
if precision_constraint == PrecisionConstraint.EXACT:
|
||||
ts = ts.replace(microsecond=0)
|
||||
return STIXdatetime(ts, precision=precision)
|
||||
# else, no need to modify fractional seconds
|
||||
|
||||
elif precision == Precision.MILLISECOND:
|
||||
if precision_constraint == PrecisionConstraint.EXACT:
|
||||
us = (ts.microsecond // 1000) * 1000
|
||||
ts = ts.replace(microsecond=us)
|
||||
# else: at least millisecond precision: the constraint will affect JSON
|
||||
# formatting, but there's nothing we need to do here.
|
||||
|
||||
# else, precision == Precision.ANY: nothing for us to do.
|
||||
|
||||
return STIXdatetime(
|
||||
ts, precision=precision, precision_constraint=precision_constraint,
|
||||
)
|
||||
|
||||
|
||||
def _get_dict(data):
|
||||
|
|
@ -256,6 +386,39 @@ def find_property_index(obj, search_key, search_value):
|
|||
return idx
|
||||
|
||||
|
||||
def _fudge_modified(old_modified, new_modified, use_stix21):
|
||||
"""
|
||||
Ensures a new modified timestamp is newer than the old. When they are
|
||||
too close together, new_modified must be pushed further ahead to ensure
|
||||
it is distinct and later, after JSON serialization (which may mean it's
|
||||
actually being pushed a little ways into the future). JSON serialization
|
||||
can remove precision, which can cause distinct timestamps to accidentally
|
||||
become equal, if we're not careful.
|
||||
|
||||
:param old_modified: A previous "modified" timestamp, as a datetime object
|
||||
:param new_modified: A candidate new "modified" timestamp, as a datetime
|
||||
object
|
||||
:param use_stix21: Whether to use STIX 2.1+ versioning timestamp precision
|
||||
rules (boolean). This is important so that we are aware of how
|
||||
timestamp precision will be truncated, so we know how close together
|
||||
the timestamps can be, and how far ahead to potentially push the new
|
||||
one.
|
||||
:return: A suitable new "modified" timestamp. This may be different from
|
||||
what was passed in, if it had to be pushed ahead.
|
||||
"""
|
||||
if use_stix21:
|
||||
# 2.1+: we can use full precision
|
||||
if new_modified <= old_modified:
|
||||
new_modified = old_modified + dt.timedelta(microseconds=1)
|
||||
else:
|
||||
# 2.0: we must use millisecond precision
|
||||
one_ms = dt.timedelta(milliseconds=1)
|
||||
if new_modified - old_modified < one_ms:
|
||||
new_modified = old_modified + one_ms
|
||||
|
||||
return new_modified
|
||||
|
||||
|
||||
def new_version(data, **kwargs):
|
||||
"""Create a new version of a STIX object, by modifying properties and
|
||||
updating the ``modified`` property.
|
||||
|
|
@ -283,12 +446,32 @@ def new_version(data, **kwargs):
|
|||
if unchangable_properties:
|
||||
raise UnmodifiablePropertyError(unchangable_properties)
|
||||
|
||||
# Different versioning precision rules in STIX 2.0 vs 2.1, so we need
|
||||
# to know which rules to apply.
|
||||
is_21 = "spec_version" in data
|
||||
precision_constraint = "min" if is_21 else "exact"
|
||||
|
||||
cls = type(data)
|
||||
if 'modified' not in kwargs:
|
||||
kwargs['modified'] = get_timestamp()
|
||||
old_modified = parse_into_datetime(
|
||||
data["modified"], precision="millisecond",
|
||||
precision_constraint=precision_constraint,
|
||||
)
|
||||
|
||||
new_modified = get_timestamp()
|
||||
new_modified = _fudge_modified(old_modified, new_modified, is_21)
|
||||
|
||||
kwargs['modified'] = new_modified
|
||||
|
||||
elif 'modified' in data:
|
||||
old_modified_property = parse_into_datetime(data.get('modified'), precision='millisecond')
|
||||
new_modified_property = parse_into_datetime(kwargs['modified'], precision='millisecond')
|
||||
old_modified_property = parse_into_datetime(
|
||||
data.get('modified'), precision='millisecond',
|
||||
precision_constraint=precision_constraint,
|
||||
)
|
||||
new_modified_property = parse_into_datetime(
|
||||
kwargs['modified'], precision='millisecond',
|
||||
precision_constraint=precision_constraint,
|
||||
)
|
||||
if new_modified_property <= old_modified_property:
|
||||
raise InvalidValueError(
|
||||
cls, 'modified',
|
||||
|
|
@ -378,11 +561,6 @@ def remove_custom_stix(stix_obj):
|
|||
|
||||
new_obj = new_version(stix_obj, **(dict(props)))
|
||||
|
||||
while parse_into_datetime(new_obj['modified']) == parse_into_datetime(stix_obj['modified']):
|
||||
# Prevents bug when fast computation allows multiple STIX object
|
||||
# versions to be created in single unit of time
|
||||
new_obj = new_version(stix_obj, **(dict(props)))
|
||||
|
||||
return new_obj
|
||||
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -87,8 +87,8 @@ class LanguageContent(_STIXBase):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('object_ref', ReferenceProperty(valid_types=["SCO", "SDO", "SRO"], spec_version='2.1', required=True)),
|
||||
# TODO: 'object_modified' it MUST be an exact match for the modified time of the STIX Object (SRO or SDO) being referenced.
|
||||
('object_modified', TimestampProperty(precision='millisecond')),
|
||||
|
|
@ -155,7 +155,7 @@ class MarkingDefinition(_STIXBase, _MarkingsMixin):
|
|||
('spec_version', StringProperty(fixed='2.1')),
|
||||
('id', IDProperty(_type)),
|
||||
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.1')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('external_references', ListProperty(ExternalReference)),
|
||||
('object_marking_refs', ListProperty(ReferenceProperty(valid_types='marking-definition', spec_version='2.1'))),
|
||||
('granular_markings', ListProperty(GranularMarking)),
|
||||
|
|
|
|||
|
|
@ -32,8 +32,8 @@ class AttackPattern(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('aliases', ListProperty(StringProperty)),
|
||||
|
|
@ -59,8 +59,8 @@ class Campaign(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('aliases', ListProperty(StringProperty)),
|
||||
|
|
@ -98,8 +98,8 @@ class CourseOfAction(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('revoked', BooleanProperty(default=lambda: False)),
|
||||
|
|
@ -122,8 +122,8 @@ class Grouping(STIXDomainObject):
|
|||
('type', TypeProperty(_type)),
|
||||
('spec_version', StringProperty(fixed='2.1')),
|
||||
('id', IDProperty(_type, spec_version='2.1')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.1')),
|
||||
('revoked', BooleanProperty(default=lambda: False)),
|
||||
('labels', ListProperty(StringProperty)),
|
||||
|
|
@ -150,8 +150,8 @@ class Identity(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('roles', ListProperty(StringProperty)),
|
||||
|
|
@ -179,8 +179,8 @@ class Indicator(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty()),
|
||||
('description', StringProperty()),
|
||||
('indicator_types', ListProperty(StringProperty)),
|
||||
|
|
@ -238,8 +238,8 @@ class Infrastructure(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('revoked', BooleanProperty(default=lambda: False)),
|
||||
('labels', ListProperty(StringProperty)),
|
||||
('confidence', IntegerProperty()),
|
||||
|
|
@ -278,8 +278,8 @@ class IntrusionSet(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('aliases', ListProperty(StringProperty)),
|
||||
|
|
@ -320,8 +320,8 @@ class Location(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty()),
|
||||
('description', StringProperty()),
|
||||
('latitude', FloatProperty(min=-90.0, max=90.0)),
|
||||
|
|
@ -427,8 +427,8 @@ class Malware(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty()),
|
||||
('description', StringProperty()),
|
||||
('malware_types', ListProperty(StringProperty)),
|
||||
|
|
@ -478,8 +478,8 @@ class MalwareAnalysis(STIXDomainObject):
|
|||
('type', TypeProperty(_type)),
|
||||
('spec_version', StringProperty(fixed='2.1')),
|
||||
('id', IDProperty(_type, spec_version='2.1')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('created_by_ref', ReferenceProperty(valid_types='identity', spec_version='2.1')),
|
||||
('revoked', BooleanProperty(default=lambda: False)),
|
||||
('labels', ListProperty(StringProperty)),
|
||||
|
|
@ -523,8 +523,8 @@ class Note(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('abstract', StringProperty()),
|
||||
('content', StringProperty(required=True)),
|
||||
('authors', ListProperty(StringProperty)),
|
||||
|
|
@ -550,8 +550,8 @@ class ObservedData(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('first_observed', TimestampProperty(required=True)),
|
||||
('last_observed', TimestampProperty(required=True)),
|
||||
('number_observed', IntegerProperty(min=1, max=999999999, required=True)),
|
||||
|
|
@ -605,8 +605,8 @@ class Opinion(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('explanation', StringProperty()),
|
||||
('authors', ListProperty(StringProperty)),
|
||||
(
|
||||
|
|
@ -642,8 +642,8 @@ class Report(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('report_types', ListProperty(StringProperty)),
|
||||
|
|
@ -670,8 +670,8 @@ class ThreatActor(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('threat_actor_types', ListProperty(StringProperty)),
|
||||
|
|
@ -716,8 +716,8 @@ class Tool(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('tool_types', ListProperty(StringProperty)),
|
||||
|
|
@ -745,8 +745,8 @@ class Vulnerability(STIXDomainObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('name', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('revoked', BooleanProperty(default=lambda: False)),
|
||||
|
|
@ -795,8 +795,8 @@ def CustomObject(type='x-custom-type', properties=None):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('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_')],
|
||||
[
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@ class Relationship(STIXRelationshipObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('relationship_type', StringProperty(required=True)),
|
||||
('description', StringProperty()),
|
||||
('source_ref', ReferenceProperty(invalid_types=_invalid_source_target_types, spec_version='2.1', required=True)),
|
||||
|
|
@ -78,8 +78,8 @@ class Sighting(STIXRelationshipObject):
|
|||
('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')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond')),
|
||||
('created', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('modified', TimestampProperty(default=lambda: NOW, precision='millisecond', precision_constraint='min')),
|
||||
('description', StringProperty()),
|
||||
('first_seen', TimestampProperty()),
|
||||
('last_seen', TimestampProperty()),
|
||||
|
|
|
|||
Loading…
Reference in New Issue