Skip to content

Feature/enrich threatactor #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bilgehanertan merged 3 commits into from
Oct 16, 2025
Merged

Feature/enrich threatactor #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a0dcf82
add: relation for actor/country
bilgehanertan Oct 16, 2025
937d0ce
add context information
bilgehanertan Oct 16, 2025
8987fc1
add version
bilgehanertan Oct 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "python-catalyst"
version = "0.1.5"
version = "0.1.6"
description = "Python client for the PRODAFT CATALYST API"
readme = "README.md"
license = { file = "LICENSE" }
Expand Down
2 changes: 1 addition & 1 deletion python_catalyst/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""PRODAFT CATALYST API client package."""

__version__ = "0.1.5"
__version__ = "0.1.6"

from .client import CatalystClient
from .enums import ObservableType, PostCategory, TLPLevel
Expand Down
43 changes: 34 additions & 9 deletions python_catalyst/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,7 +365,7 @@ def _process_entity(
collected_object_refs: List,
entity_mappings: Dict,
external_reference: stix2.ExternalReference = None,
) -> None:
) -> Tuple[List[Dict], List, Dict]:
"""
Process a single entity and add it to the report.

Expand Down Expand Up @@ -398,6 +398,8 @@ def _process_entity(
if self.logger:
self.logger.debug(f"Added {entity_type}: {entity_value}")

return related_objects, collected_object_refs, entity_mappings

def _process_entities(
self,
entities: List[Dict],
Expand All @@ -407,7 +409,7 @@ def _process_entities(
collected_object_refs: List,
entity_mappings: Dict,
external_reference: stix2.ExternalReference = None,
) -> None:
) -> Tuple[List[Dict], List, Dict]:
"""
Process a list of entities of the same type.

Expand All @@ -421,7 +423,11 @@ def _process_entities(
external_reference: Optional reference to the report
"""
for entity in entities:
self._process_entity(
(
related_objects,
collected_object_refs,
entity_mappings,
) = self._process_entity(
entity,
entity_type,
converter_method,
Expand All @@ -431,14 +437,16 @@ def _process_entities(
external_reference,
)

return related_objects, collected_object_refs, entity_mappings

def _process_threat_actor(
self,
threat_actor: Dict,
related_objects: List,
collected_object_refs: List,
entity_mappings: Dict,
external_reference: stix2.ExternalReference = None,
) -> None:
) -> Tuple[List[Dict], List, Dict]:
"""
Process a threat actor entity with detailed information.

Expand All @@ -461,14 +469,14 @@ def _process_threat_actor(
f"Retrieved detailed information for threat actor: {entity_value}"
)

ta_object = self.converter.create_detailed_threat_actor(
ta_object, bundle = self.converter.create_detailed_threat_actor(
detailed_threat_actor,
context,
report_reference=external_reference,
)

is_abstract = detailed_threat_actor.get("is_abstract", False)
related_objects.append(ta_object)
related_objects.extend(bundle)
collected_object_refs.append(ta_object.id)

if is_abstract:
Expand All @@ -482,6 +490,8 @@ def _process_threat_actor(
if self.logger:
self.logger.debug(f"Added threat actor: {entity_value}")

return related_objects, collected_object_refs, entity_mappings

except Exception as e:
if self.logger:
self.logger.warning(
Expand All @@ -501,6 +511,8 @@ def _process_threat_actor(
if self.logger:
self.logger.debug(f"Added threat actor: {entity_value}")

return related_objects, collected_object_refs, entity_mappings

def create_report_from_member_content_with_references(
self, content: Dict
) -> Tuple[Dict, List[Dict]]:
Expand Down Expand Up @@ -528,6 +540,7 @@ def create_report_from_member_content_with_references(
content_id = content.get("id")
slug = content.get("slug", "") # noqa: F841
tlp = content.get("tlp", TLPLevel.CLEAR.value)
topics = content.get("topics", [])
self.converter = self.get_stix_converter(tlp)

if published_on:
Expand All @@ -547,6 +560,9 @@ def create_report_from_member_content_with_references(
labels.append(content["category"])
if content.get("sub_category") and content["sub_category"].get("name"):
labels.append(content["sub_category"]["name"])
if len(topics) > 0:
for topic in topics:
labels.append(topic["name"])

report_id = (
f"report--{str(uuid.uuid5(uuid.NAMESPACE_URL, f'catalyst-{content_id}'))}"
Expand Down Expand Up @@ -606,14 +622,15 @@ def create_report_from_member_content_with_references(
entity_id = observable.get("id")
entity_value = observable.get("value")
entity_type = observable.get("type")

entity_context = observable.get("context", "")
if entity_id and entity_value and entity_type:
observable_data = {
"id": entity_id,
"value": entity_value,
"type": entity_type,
"post_id": content_id,
"tlp_marking": content_marking,
"context": entity_context,
}

(
Expand Down Expand Up @@ -657,7 +674,11 @@ def create_report_from_member_content_with_references(
f"Skipping threat actor {threat_actor.get('value')} because user is not authenticated... This will be implemented in the future."
)
continue
self._process_threat_actor(
(
related_objects,
collected_object_refs,
entity_mappings,
) = self._process_threat_actor(
threat_actor,
related_objects,
collected_object_refs,
Expand All @@ -684,7 +705,11 @@ def create_report_from_member_content_with_references(
converter_method = processor
mapping_type = entity_type

self._process_entities(
(
related_objects,
collected_object_refs,
entity_mappings,
) = self._process_entities(
all_entities.get(entity_type, []),
mapping_type,
converter_method,
Expand Down
48 changes: 45 additions & 3 deletions python_catalyst/stix_converter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1096,11 +1096,15 @@ def create_indicator_from_observable(
marking_ref = tlp_marking.id if tlp_marking else self.tlp_marking.id

created_by_ref = self.get_created_by_ref()
description = f"Indicator for {observable_type}: {value}"
if "context" in observable_data:
ctx = observable_data["context"]
description = f"{description}\n\n{ctx}"

return stix2.Indicator(
id=indicator_id,
name=indicator_name,
description=f"Indicator for {observable_type}: {value}",
description=description,
pattern=pattern,
pattern_type="stix",
created_by_ref=created_by_ref,
Expand Down Expand Up @@ -1210,6 +1214,7 @@ def create_detailed_threat_actor(
Returns:
STIX ThreatActor or IntrusionSet object with full details included
"""
bundle = []
external_references = []
if report_reference:
external_references = [report_reference]
Expand Down Expand Up @@ -1291,7 +1296,7 @@ def create_detailed_threat_actor(

if is_abstract:
# Create an Intrusion Set for abstract entities
return stix2.IntrusionSet(
actor = stix2.IntrusionSet(
id=IntrusionSet.generate_id(entity_value),
name=entity_value,
description=description,
Expand All @@ -1302,12 +1307,13 @@ def create_detailed_threat_actor(
external_references if external_references else None
),
custom_properties=custom_properties,
allow_custom=True,
)
else:
# Create a Threat Actor with the appropriate type
actor_type = "threat-actor-group" if is_group else "threat-actor-individual"
custom_properties["x_opencti_type"] = actor_type
return stix2.ThreatActor(
actor = stix2.ThreatActor(
id=ThreatActor.generate_id(entity_value, actor_type),
name=entity_value,
description=description,
Expand All @@ -1318,4 +1324,40 @@ def create_detailed_threat_actor(
external_references if external_references else None
),
custom_properties=custom_properties,
allow_custom=True,
)
bundle.append(actor)

suspected_origins = threat_actor_data.get("suspected_origins", [])
if isinstance(suspected_origins, list):
for origin in suspected_origins:
cname = origin.get("name")
ccode = origin.get("code")
if not cname:
continue

loc = stix2.Location(
name=cname,
country=cname,
custom_properties=(
{
"x_country_code": (
ccode.upper() if isinstance(ccode, str) else None
)
}
if ccode
else None
),
allow_custom=True,
)
rel_type = "originates-from" if is_abstract else "located-at"
rel = stix2.Relationship(
relationship_type=rel_type,
source_ref=actor.id,
target_ref=loc.id,
)

bundle.append(loc)
bundle.append(rel)

return actor, bundle
2 changes: 1 addition & 1 deletion setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

setup(
name="python-catalyst",
version="0.1.5",
version="0.1.6",
description="Python client for the PRODAFT CATALYST API",
long_description=long_description,
long_description_content_type="text/markdown",
Expand Down