Merge pull request #2 from NMD03/tidal

Tidal
pull/941/head
Niclas Dauster 2024-02-26 09:42:02 +01:00 committed by GitHub
commit 8ad3460282
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 757 additions and 0 deletions

View File

View File

@ -0,0 +1,15 @@
import requests
class TidalAPI:
def __init__(self):
self.base_url = 'https://app-api.tidalcyber.com/api/v1/'
def get_data(self, endpoint):
url = self.base_url + endpoint
try:
response = requests.get(url)
return response.json()
except Exception as e:
print(f'Error: {e}')
return None

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal Campaigns",
"namespace": "tidal",
"description": "Tidal Campaigns Galaxy",
"type": "campaigns",
"uuid": "43a8fce6-08d3-46c2-957d-53606efe2c48"
},
"cluster": {
"authors": "Tidal",
"category": "Campaigns",
"description": "Tidal Campaigns Cluster",
"name": "Tidal Campaigns",
"source": "Tidal",
"type": "campaigns"
}
}

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal Groups",
"namespace": "tidal",
"description": "Tidal Groups Galaxy",
"type": "groups",
"uuid": "41c3e5c0-de5c-4edb-b48b-48cd8e7519e6"
},
"cluster": {
"authors": "Tidal",
"category": "Threat Groups",
"description": "Tidal Threat Groups Cluster",
"name": "Tidal Threat Groups",
"source": "Tidal",
"type": "groups"
}
}

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal References",
"namespace": "tidal",
"description": "Tidal References Galaxy",
"type": "references",
"uuid": "43a8fce6-08d3-46c2-957d-53606efe2c48"
},
"cluster": {
"authors": "Tidal",
"category": "References",
"description": "Tidal References Cluster",
"name": "Tidal References",
"source": "Tidal",
"type": "references"
}
}

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal Software",
"namespace": "tidal",
"description": "Tidal Software Galaxy",
"type": "software",
"uuid": "38d62d8b-4c49-489a-9bc4-8e294c4f04f7"
},
"cluster": {
"authors": "Tidal",
"category": "Software",
"description": "Tidal Software Cluster",
"name": "Tidal Software",
"source": "Tidal",
"type": "software"
}
}

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal Tactic",
"namespace": "tidal",
"description": "Tidal Tactic Galaxy",
"type": "tactic",
"uuid": "43a8fce6-08d3-46c2-957d-53606efe2c48"
},
"cluster": {
"authors": "Tidal",
"category": "Tactic",
"description": "Tidal Tactic Cluster",
"name": "Tidal Tactic",
"source": "Tidal",
"type": "tactic"
}
}

View File

@ -0,0 +1,17 @@
{
"galaxy": {
"name": "Tidal Technique",
"namespace": "tidal",
"description": "Tidal Technique Galaxy",
"type": "technique",
"uuid": "43a8fce6-08d3-46c2-957d-53606efe2c48"
},
"cluster": {
"authors": "Tidal",
"category": "Technique",
"description": "Tidal Technique Cluster",
"name": "Tidal Technique",
"source": "Tidal",
"type": "technique"
}
}

103
tools/tidal-api/main.py Normal file
View File

@ -0,0 +1,103 @@
from api.api import TidalAPI
from models.galaxy import Galaxy
from models.cluster import (
GroupCluster,
SoftwareCluster,
CampaignsCluster,
TechniqueCluster,
TacticCluster,
ReferencesCluster,
)
import argparse
import json
import os
CONFIG = "./config"
GALAXY_PATH = "../../galaxies"
CLUSTER_PATH = "../../clusters"
def create_galaxy(endpoint: str, version: int, extended_relations: bool = False):
api = TidalAPI()
data = api.get_data(endpoint)
with open(f"{CONFIG}/{endpoint}.json", "r") as file:
config = json.load(file)
galaxy = Galaxy(**config["galaxy"], version=version)
galaxy.save_to_file(f"{GALAXY_PATH}/tidal-{endpoint}.json")
match endpoint:
case "groups":
cluster = GroupCluster(**config["cluster"], uuid=galaxy.uuid, enrichment=extended_relations)
cluster.add_values(data)
case "software":
cluster = SoftwareCluster(**config["cluster"], uuid=galaxy.uuid, enrichment=extended_relations)
cluster.add_values(data)
case "campaigns":
cluster = CampaignsCluster(**config["cluster"], uuid=galaxy.uuid)
cluster.add_values(data)
case "technique":
cluster = TechniqueCluster(**config["cluster"], uuid=galaxy.uuid)
cluster.add_values(data)
case "tactic":
cluster = TacticCluster(**config["cluster"], uuid=galaxy.uuid)
cluster.add_values(data)
case "references":
cluster = ReferencesCluster(**config["cluster"], uuid=galaxy.uuid)
cluster.add_values(data)
case _:
print("Error: Invalid endpoint")
return
cluster.save_to_file(f"{CLUSTER_PATH}/tidal-{endpoint}.json")
print(f"Galaxy tidal-{endpoint} created")
def main(args, galaxies):
if args.all:
for galaxy in galaxies:
create_galaxy(galaxy, args.version, args.extended_relations)
else:
create_galaxy(args.type, args.version, args.extended_relations)
if __name__ == "__main__":
galaxies = []
for f in os.listdir(CONFIG):
if f.endswith(".json"):
galaxies.append(f.split(".")[0])
parser = argparse.ArgumentParser(
description="Create galaxy and cluster json files from Tidal API"
)
parser.add_argument(
"-a",
"--all",
action="store_true",
help="Create all galaxies and clusters",
)
parser.add_argument(
"--type",
choices=galaxies,
help="The type of the file to create",
)
parser.add_argument(
"-v",
"--version",
type=int,
required=True,
help="The version of the galaxy",
)
parser.add_argument(
"--extended-relations",
action="store_true",
help="Create extended relations in the cluster",
)
parser.set_defaults(func=main)
args = parser.parse_args()
if hasattr(args, "func"):
args.func(args, galaxies=galaxies)
else:
parser.print_help()

View File

View File

@ -0,0 +1,522 @@
from dataclasses import dataclass, field, asdict
from typing import Type
import json
@dataclass
class Meta:
pass
@dataclass
class GroupsMeta(Meta):
source: str = None
group_attack_id: str = None
country: str = None
observed_countries: list = None
observed_motivations: list = None
target_categories: list = None
tags: list = None
owner: str = None
@dataclass
class AssociatedGroupsMeta(Meta):
id: str = None
owner_id: str = None
owner: str = None
@dataclass
class SoftwareMeta(Meta):
source: str = None
type: str = None
software_attack_id: str = None
platforms: list = None
tags: list = None
owner: str = None
@dataclass
class AssociatedSoftwareMeta(Meta):
id: str = None
owner_id: str = None
owner: str = None
@dataclass
class TechniqueMeta(Meta):
source: str = None
platforms: list = None
tags: list = None
owner: str = None
@dataclass
class SubTechniqueMeta(Meta):
source: str = None
technique_attack_id: str = None
@dataclass
class TacticMeta(Meta):
source: str = None
tactic_attack_id: str = None
ordinal_position: int = None
tags: list = None
owner: str = None
@dataclass
class ReferencesMeta(Meta):
source: str = None
refs: list = None
title: str = None
author: str = None
date_accessed: str = None
date_published: str = None
owner: str = None
@dataclass
class CampaignsMeta(Meta):
source: str = None
campaign_attack_id: str = None
first_seen: str = None
last_seen: str = None
tags: list = None
owner: str = None
@dataclass
class ClusterValue:
description: str = ""
meta: Meta = field(default_factory=Meta)
related: list = field(default_factory=list)
uuid: str = ""
value: str = ""
def return_value(self):
value_dict = asdict(self)
value_dict["meta"] = {
k: v for k, v in asdict(self.meta).items() if v is not None
}
return value_dict
class Cluster:
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
):
self.authors = authors
self.category = category
self.description = description
self.name = name
self.source = source
self.type = type
self.uuid = uuid
self.values = []
self.CLUSTER_PATH = "../../clusters"
def add_values(self, data: dict, meta_class: Type[Meta]):
pass
def save_to_file(self, path):
with open(path, "w") as file:
file.write(json.dumps(self.__dict__(), indent=4))
def __str__(self) -> str:
return f"Cluster: {self.name} - {self.type} - {self.uuid}"
def __dict__(self) -> dict:
return {
"authors": self.authors,
"category": self.category,
"description": self.description,
"name": self.name,
"source": self.source,
"type": self.type,
"uuid": self.uuid,
"values": self.values,
}
def _get_relation_from_mitre_id(
self, mitre_id: str, cluster: str, meta_key: str, array: bool = False
):
with open(f"{self.CLUSTER_PATH}/{cluster}.json", "r") as file:
mitre = json.load(file)
for entry in mitre["values"]:
try:
if array:
for id in entry["meta"][meta_key]:
if id == mitre_id:
return entry["uuid"]
else:
if entry["meta"][meta_key] == mitre_id:
return entry["uuid"]
except KeyError:
continue
return None
class GroupCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
enrichment: bool = False,
):
super().__init__(authors, category, description, name, source, type, uuid)
self.enrichment = enrichment
def add_values(self, data):
for entry in data["data"]:
meta = GroupsMeta(
source=entry.get("source"),
group_attack_id=entry.get("group_attack_id"),
country=(
entry.get("country")[0].get("country_code")
if entry.get("country")
else None
),
observed_countries=[
x.get("country_code") for x in entry.get("observed_country")
],
observed_motivations=[
x.get("name") for x in entry.get("observed_motivation")
],
target_categories=[x.get("name") for x in entry.get("observed_sector")],
tags=[x.get("tag") for x in entry.get("tags")],
owner=entry.get("owner_name"),
)
related = []
if self.enrichment:
related_cluster = self._get_relation_from_mitre_id(
entry.get("group_attack_id"), "threat-actor", "synonyms", True
)
if related_cluster:
related.append(
{
"dest-uuid": related_cluster,
"type": "similar",
}
)
for associated_group in entry.get("associated_groups"):
associated_meta = AssociatedGroupsMeta(
id=associated_group.get("id"),
owner_id=associated_group.get("owner_id"),
owner=associated_group.get("owner_name"),
)
associated_related = []
associated_related.append(
{
"dest-uuid": entry.get("id"),
"type": "similar",
}
)
value = ClusterValue(
description=associated_group.get("description"),
meta=associated_meta,
related=associated_related,
uuid=associated_group.get("associated_group_id"),
value=associated_group.get("name"),
)
self.values.append(value.return_value())
related.append(
{
"dest-uuid": associated_group.get("associated_group_id"),
"type": "similar",
}
)
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=related,
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())
class SoftwareCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
enrichment: bool = False,
):
super().__init__(authors, category, description, name, source, type, uuid)
self.enrichment = enrichment
def add_values(self, data):
for entry in data["data"]:
meta = SoftwareMeta(
source=entry.get("source"),
type=entry.get("type"),
software_attack_id=entry.get("software_attack_id"),
platforms=[x.get("name") for x in entry.get("platforms")],
tags=[x.get("tag") for x in entry.get("tags")],
owner=entry.get("owner_name"),
)
related = []
for relation in entry.get("groups"):
related.append(
{
"dest-uuid": relation.get("group_id"),
"type": "used-by",
}
)
if self.enrichment:
related_cluster = self._get_relation_from_mitre_id(
entry.get("software_attack_id"), "mitre-tool", "external_id"
)
if related_cluster:
related.append(
{
"dest-uuid": related_cluster,
"type": "similar",
}
)
related_cluster = self._get_relation_from_mitre_id(
entry.get("software_attack_id"), "mitre-malware", "external_id"
)
if related_cluster:
related.append(
{
"dest-uuid": related_cluster,
"type": "similar",
}
)
for associated_software in entry.get("associated_software"):
associated_meta = AssociatedSoftwareMeta(
id=associated_software.get("id"),
owner_id=associated_software.get("owner_id"),
owner=associated_software.get("owner_name"),
)
associated_related = []
associated_related.append(
{
"dest-uuid": entry.get("id"),
"type": "similar",
}
)
value = ClusterValue(
description=associated_software.get("description"),
meta=associated_meta,
related=associated_related,
uuid=associated_software.get("associated_software_id"),
value=associated_software.get("name"),
)
self.values.append(value.return_value())
related.append(
{
"dest-uuid": associated_software.get("associated_software_id"),
"type": "similar",
}
)
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=related,
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())
class TechniqueCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
):
super().__init__(authors, category, description, name, source, type, uuid)
def add_values(self, data):
for entry in data["data"]:
meta = TechniqueMeta(
source=entry.get("source"),
platforms=[x.get("name") for x in entry.get("platforms")],
tags=[x.get("tag") for x in entry.get("tags")],
owner=entry.get("owner_name"),
)
related = []
for relation in entry.get("tactic"):
related.append(
{
"dest-uuid": relation.get("tactic_id"),
"type": "uses",
}
)
for sub_technique in entry.get("sub_technique"):
sub_meta = SubTechniqueMeta(
source=sub_technique.get("source"),
technique_attack_id=sub_technique.get("technique_attack_id"),
)
sub_related = []
for relation in sub_technique.get("tactic"):
sub_related.append(
{
"dest-uuid": relation.get("tactic_id"),
"type": "uses",
}
)
sub_value = ClusterValue(
description=sub_technique.get("description"),
meta=sub_meta,
related=sub_related,
uuid=sub_technique.get("id"),
value=sub_technique.get("name"),
)
self.values.append(sub_value.return_value())
related.append(
{
"dest-uuid": sub_technique.get("id"),
"type": "similar",
}
)
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=related,
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())
class TacticCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
):
super().__init__(authors, category, description, name, source, type, uuid)
def add_values(self, data):
for entry in data["data"]:
meta = TacticMeta(
source=entry.get("source"),
tactic_attack_id=entry.get("tactic_attack_id"),
ordinal_position=entry.get("ordinal_position"),
tags=[x.get("tag") for x in entry.get("tags")],
owner=entry.get("owner_name"),
)
related = []
for relation in entry.get("techniques"):
related.append(
{
"dest-uuid": relation.get("technique_id"),
"type": "uses",
}
)
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=related,
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())
class ReferencesCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
):
super().__init__(authors, category, description, name, source, type, uuid)
def add_values(self, data):
for entry in data["data"]:
meta = ReferencesMeta(
source=entry.get("source"),
refs=[entry.get("url")],
title=entry.get("title"),
author=entry.get("author"),
date_accessed=entry.get("date_accessed"),
date_published=entry.get("date_published"),
owner=entry.get("owner_name"),
)
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=[],
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())
class CampaignsCluster(Cluster):
def __init__(
self,
authors: str,
category: str,
description: str,
name: str,
source: str,
type: str,
uuid: str,
):
super().__init__(authors, category, description, name, source, type, uuid)
def add_values(self, data):
for entry in data["data"]:
meta = CampaignsMeta(
source=entry.get("source"),
campaign_attack_id=entry.get("campaign_attack_id"),
first_seen=entry.get("first_seen"),
last_seen=entry.get("last_seen"),
tags=[x.get("tag") for x in entry.get("tags")],
owner=entry.get("owner_name"),
)
related = []
value = ClusterValue(
description=entry.get("description"),
meta=meta,
related=related,
uuid=entry.get("id"),
value=entry.get("name"),
)
self.values.append(value.return_value())

View File

@ -0,0 +1,15 @@
import json
from dataclasses import dataclass, asdict
@dataclass
class Galaxy():
description: str
name: str
namespace: str
type: str
uuid: str
version: str
def save_to_file(self, path: str):
with open(path, "w") as file:
file.write(json.dumps(asdict(self), indent=4))