Ruff format

core/schemas/__init__.py

@@ -9,6 +9,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 def load_entities():
     logger.info("Registering entities")
     modules = dict()
@@ -55,7 +56,10 @@ def load_indicators():
         if indicator_file.stem not in indicator.IndicatorType.__members__:
             aenum.extend_enum(indicator.IndicatorType, indicator_file.stem, enum_value)
         modules[module_name] = enum_value
-    indicator.TYPE_MAPPING = {"indicator": indicator.Indicator, "indicators": indicator.Indicator}
+    indicator.TYPE_MAPPING = {
+        "indicator": indicator.Indicator,
+        "indicators": indicator.Indicator,
+    }
     for module_name, enum_value in modules.items():
         module = importlib.import_module(module_name)
         for _, obj in inspect.getmembers(module, inspect.isclass):
@@ -70,6 +74,7 @@ def load_indicators():
         else:
             indicator.IndicatorTypes |= cls
 
+
 def load_observables():
     logger.info("Registering observables")
     modules = dict()
@@ -88,7 +93,10 @@ def load_observables():
         modules[module_name] = observable_file.stem
     if "guess" not in observable.ObservableType.__members__:
         aenum.extend_enum(observable.ObservableType, "guess", "guess")
-    observable.TYPE_MAPPING = {"observable": observable.Observable, "observables": observable.Observable}
+    observable.TYPE_MAPPING = {
+        "observable": observable.Observable,
+        "observables": observable.Observable,
+    }
     for module_name, enum_value in modules.items():
         module = importlib.import_module(module_name)
         for _, obj in inspect.getmembers(module, inspect.isclass):
@@ -103,6 +111,7 @@ def load_observables():
         else:
             observable.ObservableTypes |= cls
 
+
 load_observables()
 load_entities()
-load_indicators()
+load_indicators()

core/schemas/entities/course_of_action.py

@@ -5,4 +5,6 @@
 
 class CourseOfAction(entity.Entity):
     _type_filter: ClassVar[str] = entity.EntityType.course_of_action
-    type: Literal[entity.EntityType.course_of_action] = entity.EntityType.course_of_action
+    type: Literal[
+        entity.EntityType.course_of_action
+    ] = entity.EntityType.course_of_action

core/schemas/entity.py

@@ -78,4 +78,4 @@ def add_context(
         else:
             context["source"] = source
             self.context.append(context)
-        return self.save()
+        return self.save()

core/schemas/indicator.py

@@ -18,15 +18,16 @@ def future():
 
 DEFAULT_INDICATOR_VALIDITY_DAYS = 30
 
+
 # forward declarations
 class IndicatorType(str, Enum):
     ...
 
+
 IndicatorTypes = ()
 TYPE_MAPPING = {}
 
 
-
 class IndicatorMatch(BaseModel):
     name: str
     match: str
@@ -87,4 +88,4 @@ def search(cls, observables: list[str]) -> list[tuple[str, "Indicator"]]:
                 except NotImplementedError as error:
                     logging.error(
                         f"Indicator type {indicator.type} has not implemented match(): {error}"
-                    )
+                    )

core/schemas/indicators/forensicartifact.py

@@ -19,7 +19,9 @@ class ForensicArtifact(indicator.Indicator):
     """
 
     _type_filter: ClassVar[str] = indicator.IndicatorType.forensicartifact
-    type: Literal[indicator.IndicatorType.forensicartifact] = indicator.IndicatorType.forensicartifact
+    type: Literal[
+        indicator.IndicatorType.forensicartifact
+    ] = indicator.IndicatorType.forensicartifact
 
     sources: list[dict] = []
     aliases: list[str] = []
@@ -169,7 +171,9 @@ def save_indicators(self, create_links: bool = False):
                     regex_indicator.save()
         if create_links:
             for indicator_obj in indicators:
-                indicator_obj.link_to(self, "indicates", f"Indicates {indicator_obj.name}")
+                indicator_obj.link_to(
+                    self, "indicates", f"Indicates {indicator_obj.name}"
+                )
         return indicators
 
 

core/schemas/indicators/regex.py

@@ -30,4 +30,4 @@ def match(self, value: str) -> indicator.IndicatorMatch | None:
         result = self.compiled_pattern.search(value)
         if result:
             return indicator.IndicatorMatch(name=self.name, match=result.group())
-        return None
+        return None

core/schemas/observable.py

@@ -21,6 +21,7 @@
 class ObservableType(str, Enum):
     ...
 
+
 ObservableTypes = ()
 TYPE_MAPPING = {}
 
@@ -127,6 +128,7 @@ def delete_context(
                 break
         return self.save()
 
+
 def find_type(value: str) -> ObservableType | None:
     for obs_type, obj in TYPE_MAPPING.items():
         if obj.is_valid(value):

core/schemas/observables/certificate.py

@@ -22,7 +22,9 @@ class Certificate(observable.Observable):
         fingerprint: the certificate fingerprint.
     """
 
-    type: Literal[observable.ObservableType.certificate] = observable.ObservableType.certificate
+    type: Literal[
+        observable.ObservableType.certificate
+    ] = observable.ObservableType.certificate
     last_seen: datetime.datetime = Field(default_factory=now)
     first_seen: datetime.datetime = Field(default_factory=now)
     issuer: str | None = None

core/schemas/observables/command_line.py

@@ -4,4 +4,6 @@
 
 
 class CommandLine(observable.Observable):
-    type: Literal[observable.ObservableType.command_line] = observable.ObservableType.command_line
+    type: Literal[
+        observable.ObservableType.command_line
+    ] = observable.ObservableType.command_line

core/schemas/observables/docker_image.py

@@ -4,4 +4,6 @@
 
 
 class DockerImage(observable.Observable):
-    type: Literal[observable.ObservableType.docker_image] = observable.ObservableType.docker_image
+    type: Literal[
+        observable.ObservableType.docker_image
+    ] = observable.ObservableType.docker_image

core/schemas/observables/hostname.py

@@ -6,7 +6,9 @@
 
 
 class Hostname(observable.Observable):
-    type: Literal[observable.ObservableType.hostname] = observable.ObservableType.hostname
+    type: Literal[
+        observable.ObservableType.hostname
+    ] = observable.ObservableType.hostname
 
     @staticmethod
     def is_valid(value: str) -> bool:

core/schemas/observables/mac_address.py

@@ -4,4 +4,6 @@
 
 
 class MacAddress(observable.Observable):
-    type: Literal[observable.ObservableType.mac_address] = observable.ObservableType.mac_address
+    type: Literal[
+        observable.ObservableType.mac_address
+    ] = observable.ObservableType.mac_address

core/schemas/observables/named_pipe.py

@@ -4,4 +4,6 @@
 
 
 class NamedPipe(observable.Observable):
-    type: Literal[observable.ObservableType.named_pipe] = observable.ObservableType.named_pipe
+    type: Literal[
+        observable.ObservableType.named_pipe
+    ] = observable.ObservableType.named_pipe

core/schemas/observables/registry_key.py

@@ -26,7 +26,9 @@ class RegistryKey(observable.Observable):
         path_file: The filesystem path to the file that contains the registry key value.
     """
 
-    type: Literal[observable.ObservableType.registry_key] = observable.ObservableType.registry_key
+    type: Literal[
+        observable.ObservableType.registry_key
+    ] = observable.ObservableType.registry_key
     key: str
     data: bytes
     hive: RegistryHive

core/schemas/observables/user_account.py

@@ -14,7 +14,9 @@ class UserAccount(observable.Observable):
     Value should to be in the form <ACCOUNT_TYPE>:<ACCOUNT_LOGIN>.
     """
 
-    type: Literal[observable.ObservableType.user_account] = observable.ObservableType.user_account
+    type: Literal[
+        observable.ObservableType.user_account
+    ] = observable.ObservableType.user_account
     user_id: str | None = None
     credential: str | None = None
     account_login: str | None = None

core/schemas/observables/user_agent.py

@@ -4,4 +4,6 @@
 
 
 class UserAgent(observable.Observable):
-    type: Literal[observable.ObservableType.user_agent] = observable.ObservableType.user_agent
+    type: Literal[
+        observable.ObservableType.user_agent
+    ] = observable.ObservableType.user_agent

core/web/apiv2/graph.py

@@ -111,7 +111,7 @@ class GraphSearchResponse(BaseModel):
         | entity.EntityTypes
         | indicator.IndicatorTypes
         | tag.Tag
-        | dfiq.DFIQTypes
+        | dfiq.DFIQTypes,
     ]
     paths: list[list[graph.Relationship | graph.TagRelationship]]
     total: int

core/web/apiv2/indicators.py

@@ -79,7 +79,9 @@ async def patch(request: PatchIndicatorRequest, indicator_id) -> IndicatorTypes:
 
     if db_indicator.type == IndicatorType.forensicartifact:
         if db_indicator.pattern != request.indicator.pattern:
-            return forensicartifact.ForensicArtifact.from_yaml_string(request.indicator.pattern)[0]
+            return forensicartifact.ForensicArtifact.from_yaml_string(
+                request.indicator.pattern
+            )[0]
 
     update_data = request.indicator.model_dump(exclude_unset=True)
     updated_indicator = db_indicator.model_copy(update=update_data)

core/web/apiv2/observables.py

@@ -46,6 +46,7 @@ class PatchObservableRequest(BaseModel):
 
     observable: ObservableTypes = Field(discriminator="type")
 
+
 class NewBulkObservableAddRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 

tests/schemas/indicator.py

@@ -38,7 +38,9 @@ def test_filter_entities_different_types(self) -> None:
 
         self.assertEqual(len(all_entities), 1)
         self.assertEqual(len(regex_indicators), 1)
-        self.assertEqual(regex_indicators[0].model_dump_json(), regex_indicator.model_dump_json())
+        self.assertEqual(
+            regex_indicators[0].model_dump_json(), regex_indicator.model_dump_json()
+        )
 
     def test_create_indicator_same_name_diff_types(self) -> None:
         regex1 = regex.Regex(
@@ -242,7 +244,9 @@ def test_forensic_artifacts_parent_extraction(self):
     - blah3
 """
 
-        artifacts = forensicartifact.ForensicArtifact.from_yaml_string(pattern, update_parents=True)
+        artifacts = forensicartifact.ForensicArtifact.from_yaml_string(
+            pattern, update_parents=True
+        )
         self.assertEqual(len(artifacts), 3)
 
         vertices, _, total = artifacts[0].neighbors()