Skip to content

Commit 9183447

Browse files
rpiazzarpiazza
committed
constraitns for reg-ex db dependent
1 parent f11cce1 commit 9183447

File tree

3 files changed

+26
-46
lines changed

3 files changed

+26
-46
lines changed

‎stix2/datastore/relational_db/database_backends/database_backend_base.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,10 @@ def determine_stix_type(stix_object):
120120
def array_allowed():
121121
return False
122122

123+
@staticmethod
124+
def create_regex_constraint_expression(column, pattern):
125+
pass
126+
123127
def process_value_for_insert(self, stix_type, value):
124128
sql_type = stix_type.determine_sql_type(self)
125129
if sql_type == self.determine_sql_type_for_string_property():

‎stix2/datastore/relational_db/database_backends/postgres_backend.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,3 +73,7 @@ def determine_sql_type_for_timestamp_property(): # noqa: F811
7373
@staticmethod
7474
def array_allowed():
7575
return True
76+
77+
@staticmethod
78+
def create_regex_constraint_expression(column_name, pattern):
79+
return f"{column_name} ~ {pattern}"

‎stix2/datastore/relational_db/table_creation.py‎

Lines changed: 18 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,7 @@ def create_kill_chain_phases_table(name, metadata, db_backend, schema_name, tabl
162162
def create_granular_markings_table(metadata, db_backend, sco_or_sdo):
163163
schema_name = db_backend.schema_for_core()
164164
tables = list()
165+
reg_ex = f"'^marking-definition--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: E131
165166
columns = [
166167
Column(
167168
"id",
@@ -173,10 +174,7 @@ def create_granular_markings_table(metadata, db_backend, sco_or_sdo):
173174
Column(
174175
"marking_ref",
175176
db_backend.determine_sql_type_for_reference_property(),
176-
CheckConstraint(
177-
"marking_ref ~ '^marking-definition--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'",
178-
# noqa: E131
179-
),
177+
CheckConstraint(db_backend.create_regex_constraint_expression("marking_ref", reg_ex)),
180178
),
181179
]
182180
if db_backend.array_allowed():
@@ -230,15 +228,13 @@ def create_granular_markings_table(metadata, db_backend, sco_or_sdo):
230228

231229

232230
def create_external_references_tables(metadata, db_backend):
231+
reg_ex = "'^[a-z][a-z0-9-]+[a-z0-9]--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: E131
233232
columns = [
234233
Column(
235234
"id",
236235
db_backend.determine_sql_type_for_key_as_id(),
237236
ForeignKey("common.core_sdo" + ".id", ondelete="CASCADE"),
238-
CheckConstraint(
239-
"id ~ '^[a-z][a-z0-9-]+[a-z0-9]--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'", # noqa: E131
240-
),
241-
),
237+
CheckConstraint(db_backend.create_regex_constraint_expression("id", reg_ex))),
242238
Column("source_name", db_backend.determine_sql_type_for_string_property()),
243239
Column("description", db_backend.determine_sql_type_for_string_property()),
244240
Column("url", db_backend.determine_sql_type_for_string_property()),
@@ -255,26 +251,23 @@ def create_external_references_tables(metadata, db_backend):
255251
def create_core_table(metadata, db_backend, stix_type_name):
256252
tables = list()
257253
table_name = "core_" + stix_type_name
254+
reg_ex = "'^[a-z][a-z0-9-]+[a-z0-9]--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: E131
258255
columns = [
259256
Column(
260257
"id",
261258
db_backend.determine_sql_type_for_key_as_id(),
262-
CheckConstraint(
263-
"id ~ '^[a-z][a-z0-9-]+[a-z0-9]--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'", # noqa: E131
264-
),
259+
CheckConstraint(db_backend.create_regex_constraint_expression("id", reg_ex)),
265260
primary_key=True,
266261
),
267262
Column("spec_version", db_backend.determine_sql_type_for_string_property(), default="2.1"),
268263
]
269264
if stix_type_name == "sdo":
265+
reg_ex = "'^identity--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: E131
270266
sdo_columns = [
271267
Column(
272268
"created_by_ref",
273269
db_backend.determine_sql_type_for_reference_property(),
274-
CheckConstraint(
275-
"created_by_ref ~ '^identity--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'", # noqa: E131
276-
),
277-
),
270+
CheckConstraint(db_backend.create_regex_constraint_expression("created_by_ref", reg_ex))),
278271
Column("created", db_backend.determine_sql_type_for_timestamp_property()),
279272
Column("modified", db_backend.determine_sql_type_for_timestamp_property()),
280273
Column("revoked", db_backend.determine_sql_type_for_boolean_property()),
@@ -405,9 +398,8 @@ def generate_table_information(self, name, db_backend, **kwargs): # noqa: F811
405398
return Column(
406399
name,
407400
self.determine_sql_type(db_backend),
408-
CheckConstraint(
409-
# this regular expression might accept or reject some legal base64 strings
410-
f"{name} ~ " + "'^[-A-Za-z0-9+/]*={0,3}$'",
401+
# this regular expression might accept or reject some legal base64 strings
402+
CheckConstraint(db_backend.create_regex_constraint_expression(name, "'^[-A-Za-z0-9+/]*={0,3}$'")
411403
),
412404
nullable=not self.required,
413405
)
@@ -534,9 +526,7 @@ def generate_table_information(self, name, db_backend, **kwargs): # noqa: F811
534526
return Column(
535527
name,
536528
self.determine_sql_type(db_backend),
537-
CheckConstraint(
538-
f"{name} ~ '^{enum_re}$'",
539-
),
529+
CheckConstraint(db_backend.create_regex_constraint_expression(name, f"'^{enum_re}$'")),
540530
nullable=not self.required,
541531
)
542532

@@ -609,18 +599,7 @@ def generate_table_information(self, name, db_backend, **kwargs): # noqa: F811
609599
schema_name = kwargs.get('schema_name')
610600
table_name = kwargs.get("table_name")
611601
core_table = kwargs.get("core_table")
612-
# if schema_name == "common":
613-
# return Column(
614-
# name,
615-
# Text,
616-
# CheckConstraint(
617-
# f"{name} ~ '^{table_name}" + "--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'",
618-
# # noqa: E131
619-
# ),
620-
# primary_key=True,
621-
# nullable=not (self.required),
622-
# )
623-
# else:
602+
id_req_exp = f"'^{table_name}" + "--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: E131
624603
if schema_name:
625604
foreign_key_column = f"common.core_{core_table}.id"
626605
else:
@@ -630,8 +609,7 @@ def generate_table_information(self, name, db_backend, **kwargs): # noqa: F811
630609
db_backend.determine_sql_type_for_key_as_id(),
631610
ForeignKey(foreign_key_column, ondelete="CASCADE"),
632611
CheckConstraint(
633-
f"{name} ~ '^{table_name}" + "--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'",
634-
# noqa: E131
612+
db_backend.create_regex_constraint_expression(name, id_req_exp)
635613
),
636614
primary_key=True,
637615
nullable=not (self.required),
@@ -743,17 +721,14 @@ def ref_column(name, specifics, db_backend, auth_type=0):
743721
if specifics:
744722
types = "|".join(specifics)
745723
if auth_type == 0:
724+
reg_ex = f"'^({types})" + "--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'" # noqa: F811
746725
constraint = \
747-
CheckConstraint(
748-
f"{name} ~ '^({types})" +
749-
"--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$'",
750-
)
726+
CheckConstraint(db_backend.create_regex_constraint_expression(name, reg_ex))
751727
else:
728+
reg_ex = "'--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$')"
752729
constraint = \
753-
CheckConstraint(
754-
f"(NOT({name} ~ '^({types})')) AND ({name} ~ " +
755-
"'--[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$')",
756-
)
730+
CheckConstraint(db_backend.create_regex_constraint_expression(f"NOT({name}", f"'^({types})'") + " AND " +
731+
db_backend.create_regex_constraint_expression(name, reg_ex))
757732
return Column(name, db_backend.determine_sql_type_for_reference_property(), constraint)
758733
else:
759734
return Column(
@@ -789,9 +764,6 @@ def generate_table_information(self, name, db_backend, **kwargs): # noqa: F811
789764
return Column(
790765
name,
791766
self.determine_sql_type(db_backend),
792-
# CheckConstraint(
793-
# f"{name} ~ '^{enum_re}$'"
794-
# ),
795767
nullable=not (self.required),
796768
)
797769

0 commit comments

Comments
 (0)