mirror of
https://github.com/MISP/misp-galaxy.git
synced 2024-11-26 08:47:18 +00:00
refactor [tool] code
This commit is contained in:
parent
9467e101bf
commit
35b8192208
14 changed files with 444 additions and 504 deletions
|
@ -1,84 +0,0 @@
|
|||
# Tidal Cyber API
|
||||
|
||||
This is a tool generating MISP galaxies and clusters from Tidal Cyber API.
|
||||
|
||||
## Endpoints
|
||||
https://app-api.tidalcyber.com/api/v1/technique
|
||||
|
||||
https://app-api.tidalcyber.com/api/v1/references
|
||||
|
||||
https://app-api.tidalcyber.com/api/v1/tactic
|
||||
|
||||
https://app-api.tidalcyber.com/api/v1/campaigns/
|
||||
|
||||
https://app-api.tidalcyber.com/api/v1/software/
|
||||
|
||||
https://app-api.tidalcyber.com/api/v1/groups/
|
||||
|
||||
## Configuration
|
||||
The configuration file is located in `config.json` and maps the fields of the Tidal API to the Galaxy and Cluster fields. It consists of the following sections:
|
||||
- `UUID`: The UUID of the galaxy to be created
|
||||
- `GALAXY_CONFIGS`: The configuration of the galaxies to be created in the `galaxies` folder of the MISP-galaxy repository
|
||||
- `name`: The name of the galaxy
|
||||
- `namespace`: The namespace of the galaxy
|
||||
- `description`: The description of the galaxy
|
||||
- `type`: The type of the galaxy
|
||||
- `uuid`: The UUID of the galaxy (will be inserted from the `UUID` section)
|
||||
- `CLUSTER_CONFIGS`: The configuration of the clusters to be created in the `clusters` folder of the MISP-galaxy repository
|
||||
- `authors`: The authors of the cluster
|
||||
- `category`: The category of the cluster
|
||||
- `description`: The description of the cluster
|
||||
- `name`: The name of the cluster
|
||||
- `source`: The source of the cluster
|
||||
- `type`: The type of the cluster
|
||||
- `uuid`: The UUID of the cluster (will be inserted from the `UUID` section)
|
||||
- `values`: The values of the cluster (will be inserted from the `VALUE_FIELDS` section)
|
||||
- `VALUE_FIELDS`: Defines the mapping of the fields in the Tidal Cyber API to the fields in the MISP cluster values array
|
||||
- `description`: The description of the cluster value
|
||||
- `meta`: The metadata of the cluster value
|
||||
- `related`: The related cluster values of the cluster value (you can define a `type` for each relation type in the config which will not be mapped to a field of the API)
|
||||
- `uuid`: The UUID of the cluster value
|
||||
- `value`: The value of the cluster value
|
||||
>Note: The fields `meta` can be formatted as the format of the data the API provides sometimes does not match the format defined by the [MISP galaxy format](https://www.misp-standard.org/rfc/misp-standard-galaxy-format.html#name-conventions-and-terminology). You can configure this using an extraction configuration.
|
||||
|
||||
### Extraction Configuration
|
||||
The extraction configuration is a dictionary that maps the fields of the Tidal Cyber API to the fields of the MISP galaxy. It can be used to extract data stored in an object in the API response. The extraction configuration looks like this:
|
||||
```json
|
||||
{
|
||||
"extract": "<mode>",
|
||||
"key": "<key>",
|
||||
"subkey": "<subkey>"
|
||||
}
|
||||
```
|
||||
**Extract modes**:
|
||||
|
||||
- `single`: Extracts a single value from the API response
|
||||
- `multiple`: Extracts multiple values from the API response
|
||||
- `reverse`: Gets the value of the key and writes it into an array (no subkey needed)
|
||||
|
||||
### "Private" Relations
|
||||
The Tidal Cyber API provides relations between different objects. Some of these relations point to objects that are not part of the galaxies created based on the API response nor are they part of the MISP galaxy. These relations can be marked as `private` in the config file. For example:
|
||||
```json
|
||||
"related": {
|
||||
"tactic": {
|
||||
"mode": "public",
|
||||
"dest-uuid": "tactic_id",
|
||||
"type": "uses"
|
||||
},
|
||||
"sub_technique": {
|
||||
"mode": "private",
|
||||
"dest-uuid": "id",
|
||||
"type": "sub-technique-of"
|
||||
}
|
||||
},
|
||||
```
|
||||
|
||||
## Usage
|
||||
```bash
|
||||
python3 main.py create-galaxy -v <version> --type <galaxy_to_create>
|
||||
```
|
||||
To build all galaxies and clusters, run the following command:
|
||||
|
||||
```bash
|
||||
python3 main.py create-galaxy -v <version> --all
|
||||
```
|
|
@ -1,271 +0,0 @@
|
|||
{
|
||||
"UUIDS": {
|
||||
"software": "38d62d8b-4c49-489a-9bc4-8e294c4f04f7",
|
||||
"groups": "41c3e5c0-de5c-4edb-b48b-48cd8e7519e6",
|
||||
"campaigns": "43a8fce6-08d3-46c2-957d-53606efe2c48",
|
||||
"technique": "3e28b683-8159-4398-8729-7248eac9aa45",
|
||||
"tactic": "16a396e2-a4a9-4dfd-be0a-6ba75fb5382c",
|
||||
"references": "cf5e180f-26e9-42b2-b10c-c9e55f448750"
|
||||
},
|
||||
"GALAXY_CONFIGS": {
|
||||
"software": {
|
||||
"name": "Tidal Software",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal Software Galaxy",
|
||||
"type": "software",
|
||||
"uuid": ""
|
||||
},
|
||||
"groups": {
|
||||
"name": "Tidal Groups",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal Groups Galaxy",
|
||||
"type": "groups",
|
||||
"uuid": ""
|
||||
},
|
||||
"campaigns": {
|
||||
"name": "Tidal Campaigns",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal Campaigns Galaxy",
|
||||
"type": "campaigns",
|
||||
"uuid": ""
|
||||
},
|
||||
"technique": {
|
||||
"name": "Tidal Technique",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal Technique Galaxy",
|
||||
"type": "technique",
|
||||
"uuid": ""
|
||||
},
|
||||
"tactic": {
|
||||
"name": "Tidal Tactic",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal Tactic Galaxy",
|
||||
"type": "tactics",
|
||||
"uuid": ""
|
||||
},
|
||||
"references": {
|
||||
"name": "Tidal References",
|
||||
"namespace": "tidal",
|
||||
"description": "Tidal References Galaxy",
|
||||
"type": "references",
|
||||
"uuid": ""
|
||||
}
|
||||
},
|
||||
"CLUSTER_CONFIGS": {
|
||||
"software": {
|
||||
"authors": "Tidal",
|
||||
"category": "Software",
|
||||
"description": "Tidal Software Cluster",
|
||||
"name": "Tidal Software",
|
||||
"source": "Tidal",
|
||||
"type": "software",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
},
|
||||
"groups": {
|
||||
"authors": "Tidal",
|
||||
"category": "Threat Groups",
|
||||
"description": "Tidal Threat Groups Cluster",
|
||||
"name": "Tidal Threat Groups",
|
||||
"source": "Tidal",
|
||||
"type": "groups",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
},
|
||||
"campaigns": {
|
||||
"authors": "Tidal",
|
||||
"category": "Campaigns",
|
||||
"description": "Tidal Campaigns Cluster",
|
||||
"name": "Tidal Campaigns",
|
||||
"source": "Tidal",
|
||||
"type": "campaigns",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
},
|
||||
"technique": {
|
||||
"authors": "Tidal",
|
||||
"category": "Techniques",
|
||||
"description": "Tidal Techniques Cluster",
|
||||
"name": "Tidal Techniques",
|
||||
"source": "Tidal",
|
||||
"type": "technique",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
},
|
||||
"tactic": {
|
||||
"authors": "Tidal",
|
||||
"category": "Tactics",
|
||||
"description": "Tidal Tactics Cluster",
|
||||
"name": "Tidal Tactics",
|
||||
"source": "Tidal",
|
||||
"type": "tactic",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
},
|
||||
"references": {
|
||||
"authors": "Tidal",
|
||||
"category": "References",
|
||||
"description": "Tidal References Cluster",
|
||||
"name": "Tidal References",
|
||||
"source": "Tidal",
|
||||
"type": "references",
|
||||
"uuid": "",
|
||||
"values": []
|
||||
}
|
||||
},
|
||||
"VALUE_FIELDS": {
|
||||
"software": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"type": "type",
|
||||
"software-attack-id": "software_attack_id",
|
||||
"platforms": {
|
||||
"extract": "multiple",
|
||||
"key": "platforms",
|
||||
"subkey": "name"
|
||||
},
|
||||
"tags": {
|
||||
"extract": "multiple",
|
||||
"key": "tags",
|
||||
"subkey": "tag"
|
||||
},
|
||||
"owner": "owner_name"
|
||||
},
|
||||
"related": {
|
||||
"groups": {
|
||||
"mode": "private",
|
||||
"dest-uuid": "group_id",
|
||||
"type": "used-by"
|
||||
},
|
||||
"associated_software": {
|
||||
"mode": "private",
|
||||
"dest-uuid": "id",
|
||||
"type": "related-to"
|
||||
}
|
||||
},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
},
|
||||
"groups": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"group-attack-id": "group_attack_id",
|
||||
"country": {
|
||||
"extract": "single",
|
||||
"key": "country",
|
||||
"subkey": "country_code"
|
||||
},
|
||||
"observed_country": {
|
||||
"extract": "multiple",
|
||||
"key": "observed_country",
|
||||
"subkey": "country_code"
|
||||
},
|
||||
"observed_motivation": {
|
||||
"extract": "multiple",
|
||||
"key": "observed_motivation",
|
||||
"subkey": "name"
|
||||
},
|
||||
"target-category": {
|
||||
"extract": "multiple",
|
||||
"key": "observed_sector",
|
||||
"subkey": "name"
|
||||
},
|
||||
"tags": "tags",
|
||||
"owner": "owner_name"
|
||||
},
|
||||
"related": {
|
||||
"associated_groups": {
|
||||
"mode": "private",
|
||||
"dest-uuid": "id",
|
||||
"type": "related-to"
|
||||
}
|
||||
},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
},
|
||||
"campaigns": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"campaign-attack-id": "campaign_attack_id",
|
||||
"first_seen": "first_seen",
|
||||
"last_seen": "last_seen",
|
||||
"tags": "tags",
|
||||
"owner": "owner_name"
|
||||
},
|
||||
"related": {},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
},
|
||||
"technique": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"platforms": {
|
||||
"extract": "multiple",
|
||||
"key": "platforms",
|
||||
"subkey": "name"
|
||||
},
|
||||
"tags": {
|
||||
"extract": "multiple",
|
||||
"key": "tags",
|
||||
"subkey": "tag"
|
||||
},
|
||||
"owner": "owner_name"
|
||||
},
|
||||
"related": {
|
||||
"tactic": {
|
||||
"mode": "public",
|
||||
"dest-uuid": "tactic_id",
|
||||
"type": "uses"
|
||||
},
|
||||
"sub_technique": {
|
||||
"mode": "private",
|
||||
"dest-uuid": "id",
|
||||
"type": "sub-technique-of"
|
||||
}
|
||||
},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
},
|
||||
"tactic": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"tactic-attack-id": "tactic_attack_id",
|
||||
"ordinal_position": "ordinal_position",
|
||||
"tags": "tags",
|
||||
"owner": "owner_name"
|
||||
},
|
||||
"related": {
|
||||
"techniques": {
|
||||
"mode": "public",
|
||||
"dest-uuid": "technique_id",
|
||||
"type": "uses"
|
||||
}
|
||||
},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
},
|
||||
"references": {
|
||||
"description": "description",
|
||||
"meta": {
|
||||
"source": "source",
|
||||
"refs": {
|
||||
"extract": "reverse",
|
||||
"key": "url"
|
||||
},
|
||||
"owner": "owner_name",
|
||||
"title": "title",
|
||||
"author": "authors",
|
||||
"date_accessed": "date_accessed",
|
||||
"date_published": "date_published"
|
||||
},
|
||||
"related": {},
|
||||
"uuid": "id",
|
||||
"value": "name"
|
||||
}
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/campaigns.json
Normal file
17
tools/tidal-api/config/campaigns.json
Normal 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"
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/groups.json
Normal file
17
tools/tidal-api/config/groups.json
Normal 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"
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/references.json
Normal file
17
tools/tidal-api/config/references.json
Normal 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"
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/software.json
Normal file
17
tools/tidal-api/config/software.json
Normal 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"
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/tactic.json
Normal file
17
tools/tidal-api/config/tactic.json
Normal 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"
|
||||
}
|
||||
}
|
17
tools/tidal-api/config/technique.json
Normal file
17
tools/tidal-api/config/technique.json
Normal 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"
|
||||
}
|
||||
}
|
|
@ -1,134 +1,89 @@
|
|||
from api.api import TidalAPI
|
||||
from models.galaxy import Galaxy
|
||||
from models.cluster import Cluster
|
||||
from utils.extractor import extract_links
|
||||
from utils.config import load_config
|
||||
from models.cluster import GroupCluster, SoftwareCluster, CampaignsCluster, TechniqueCluster, TacticCluster, ReferencesCluster
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
|
||||
CLUSTER_PATH = "../../clusters/"
|
||||
GALAXY_PATH = "../../galaxies/"
|
||||
CONFIG = "./config"
|
||||
GALAXY_PATH = "../../galaxies"
|
||||
CLUSTER_PATH = "../../clusters"
|
||||
|
||||
config = load_config("./config.json")
|
||||
|
||||
UUIDS = config["UUIDS"]
|
||||
GALAXY_CONFIGS = config["GALAXY_CONFIGS"]
|
||||
CLUSTER_CONFIGS = config["CLUSTER_CONFIGS"]
|
||||
VALUE_FIELDS = config["VALUE_FIELDS"]
|
||||
|
||||
|
||||
def create_cluster_values(data, cluster, add_private):
|
||||
value_fields = VALUE_FIELDS[cluster.internal_type]
|
||||
for entry in data["data"]:
|
||||
values = {}
|
||||
for key, value in value_fields.items():
|
||||
match key:
|
||||
case "description":
|
||||
values[value] = entry.get(key)
|
||||
case "meta":
|
||||
metadata = create_metadata(entry, value)
|
||||
values["meta"] = metadata
|
||||
case "related":
|
||||
relations = create_relations(entry, value, add_private)
|
||||
values["related"] = relations
|
||||
case "uuid":
|
||||
values[key] = entry.get(value)
|
||||
case "value":
|
||||
values[key] = entry.get(value)
|
||||
case _:
|
||||
print(
|
||||
f"Error: Invalid configuration for {key} in {cluster.internal_type} value fields."
|
||||
)
|
||||
cluster.add_value(values)
|
||||
|
||||
|
||||
def create_metadata(data, format):
|
||||
metadata = {}
|
||||
for meta_key, meta_value in format.items():
|
||||
if isinstance(meta_value, dict):
|
||||
if meta_value.get("extract") == "single" and data.get(meta_value["key"]):
|
||||
metadata[meta_key] = data.get(meta_value["key"])[0].get(
|
||||
meta_value["subkey"]
|
||||
)
|
||||
elif meta_value.get("extract") == "multiple" and data.get(
|
||||
meta_value["key"]
|
||||
):
|
||||
metadata[meta_key] = [
|
||||
entry.get(meta_value["subkey"])
|
||||
for entry in data.get(meta_value["key"])
|
||||
]
|
||||
elif meta_value.get("extract") == "reverse" and data.get(meta_value["key"]):
|
||||
metadata[meta_key] = [data.get(meta_value["key"])]
|
||||
elif data.get(meta_value):
|
||||
metadata[meta_key] = data.get(meta_value)
|
||||
return metadata
|
||||
|
||||
|
||||
def create_relations(data, format, add_private):
|
||||
relations = []
|
||||
for i in range(len(list(format))):
|
||||
for relation in data[list(format)[i]]:
|
||||
if not add_private and list(format.values())[i].get("mode") == "private":
|
||||
continue
|
||||
relation_entry = {}
|
||||
for relation_key, relation_value in list(format.values())[i].items():
|
||||
if relation_key != "type":
|
||||
if relation_key == "mode":
|
||||
continue
|
||||
relation_entry[relation_key] = relation.get(relation_value)
|
||||
else:
|
||||
relation_entry[relation_key] = relation_value
|
||||
relations.append(relation_entry)
|
||||
return relations
|
||||
|
||||
|
||||
def create_galaxy_and_cluster(galaxy_type, version, add_private=False):
|
||||
def create_galaxy(endpoint: str, version: int):
|
||||
api = TidalAPI()
|
||||
galaxy = Galaxy(**GALAXY_CONFIGS[galaxy_type], version=version)
|
||||
galaxy.save_to_file(f"{GALAXY_PATH}/tidal-{galaxy_type}.json")
|
||||
data = api.get_data(endpoint)
|
||||
with open(f"{CONFIG}/{endpoint}.json", "r") as file:
|
||||
config = json.load(file)
|
||||
|
||||
cluster = Cluster(**CLUSTER_CONFIGS[galaxy_type], internal_type=galaxy_type)
|
||||
data = api.get_data(galaxy_type)
|
||||
create_cluster_values(data, cluster, add_private)
|
||||
cluster.save_to_file(f"{CLUSTER_PATH}/tidal-{galaxy_type}.json")
|
||||
galaxy = Galaxy(**config["galaxy"], version=version)
|
||||
galaxy.save_to_file(f"{GALAXY_PATH}/tidal-{endpoint}.json")
|
||||
|
||||
print(f"Galaxy tidal-{galaxy_type} created")
|
||||
match endpoint:
|
||||
case "groups":
|
||||
cluster = GroupCluster(**config["cluster"], uuid=galaxy.uuid)
|
||||
cluster.add_values(data)
|
||||
case "software":
|
||||
cluster = SoftwareCluster(**config["cluster"], uuid=galaxy.uuid)
|
||||
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 create_galaxy(args):
|
||||
def main(args, galaxies):
|
||||
if args.all:
|
||||
for galaxy_type in GALAXY_CONFIGS:
|
||||
create_galaxy_and_cluster(galaxy_type, args.version, args.addprivate)
|
||||
for galaxy in galaxies:
|
||||
create_galaxy(galaxy, args.version)
|
||||
else:
|
||||
create_galaxy_and_cluster(args.type, args.version, args.addprivate)
|
||||
create_galaxy(args.type, args.version)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Create a galaxy and cluster for Tidal API"
|
||||
)
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
|
||||
galaxy_parser = subparsers.add_parser(
|
||||
"create_galaxy", help="Create a galaxy from the Tidal API"
|
||||
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"
|
||||
)
|
||||
galaxy_parser.add_argument(
|
||||
parser.add_argument(
|
||||
"--all",
|
||||
action="store_true",
|
||||
help="Create all galaxies and clusters",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--type",
|
||||
choices=list(GALAXY_CONFIGS.keys()) + ["all"],
|
||||
help="The type of the galaxy",
|
||||
choices=galaxies,
|
||||
help="The type of the file to create",
|
||||
)
|
||||
galaxy_parser.add_argument(
|
||||
"-v", "--version", type=int, required=True, help="The version of the galaxy"
|
||||
parser.add_argument(
|
||||
"-v",
|
||||
"--version",
|
||||
type=int,
|
||||
required=True,
|
||||
help="The version of the galaxy",
|
||||
)
|
||||
galaxy_parser.add_argument(
|
||||
"--all", action="store_true", help="Flag to create all predefined galaxy types"
|
||||
)
|
||||
galaxy_parser.add_argument(
|
||||
"--addprivate", action="store_true", help="Flag to add private relations"
|
||||
)
|
||||
galaxy_parser.set_defaults(func=create_galaxy)
|
||||
parser.set_defaults(func=main)
|
||||
|
||||
args = parser.parse_args()
|
||||
if hasattr(args, "func"):
|
||||
args.func(args)
|
||||
args.func(args, galaxies=galaxies)
|
||||
else:
|
||||
parser.print_help()
|
|
@ -1,7 +1,79 @@
|
|||
from dataclasses import dataclass, field, asdict
|
||||
import json
|
||||
|
||||
@dataclass
|
||||
class Meta:
|
||||
pass
|
||||
|
||||
@dataclass
|
||||
class GroupsMeta():
|
||||
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 SoftwareMeta():
|
||||
source: str = None
|
||||
type: str = None
|
||||
software_attack_id: str = None
|
||||
platforms: list = None
|
||||
tags: list = None
|
||||
owner: str = None
|
||||
|
||||
@dataclass
|
||||
class TechniqueMeta():
|
||||
source: str = None
|
||||
platforms: list = None
|
||||
tags: list = None
|
||||
owner: str = None
|
||||
|
||||
@dataclass
|
||||
class TacticMeta():
|
||||
source: str = None
|
||||
tactic_attack_id: str = None
|
||||
ordinal_position: int = None
|
||||
tags: list = None
|
||||
owner: str = None
|
||||
|
||||
@dataclass
|
||||
class ReferencesMeta():
|
||||
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():
|
||||
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, values: list, internal_type: str):
|
||||
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
|
||||
|
@ -9,15 +81,200 @@ class Cluster():
|
|||
self.source = source
|
||||
self.type = type
|
||||
self.uuid = uuid
|
||||
self.values = values
|
||||
self.internal_type = internal_type
|
||||
self.values = []
|
||||
|
||||
def add_value(self, value):
|
||||
self.values.append(value)
|
||||
def add_values(self):
|
||||
print("This method should be implemented in the child class")
|
||||
|
||||
def save_to_file(self, path):
|
||||
with open(path, "w") as file:
|
||||
file.write(json.dumps(self.__dict__, indent=4))
|
||||
file.write(json.dumps(self.__dict__(), indent=4))
|
||||
|
||||
def get_config(self):
|
||||
return self.__dict__
|
||||
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,
|
||||
}
|
||||
|
||||
class GroupCluster(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 = 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 = []
|
||||
for relation in entry.get("associated_groups"):
|
||||
related.append({
|
||||
"dest-uuid": relation.get("id"),
|
||||
"type": "related-to",
|
||||
}
|
||||
)
|
||||
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):
|
||||
super().__init__(authors, category, description, name, source, type, uuid)
|
||||
|
||||
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",
|
||||
}
|
||||
)
|
||||
for relation in entry.get("associated_software"):
|
||||
related.append({
|
||||
"dest-uuid": relation.get("id"),
|
||||
"type": "related-to",
|
||||
}
|
||||
)
|
||||
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",
|
||||
}
|
||||
)
|
||||
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())
|
|
@ -1,14 +1,15 @@
|
|||
import json
|
||||
from dataclasses import dataclass, asdict
|
||||
|
||||
@dataclass
|
||||
class Galaxy():
|
||||
def __init__(self, description, name, namespace, type, uuid, version):
|
||||
self.description = description
|
||||
self.name = name
|
||||
self.namespace = namespace
|
||||
self.type = type
|
||||
self.uuid = uuid
|
||||
self.version = version
|
||||
description: str
|
||||
name: str
|
||||
namespace: str
|
||||
type: str
|
||||
uuid: str
|
||||
version: str
|
||||
|
||||
def save_to_file(self, path):
|
||||
def save_to_file(self, path: str):
|
||||
with open(path, "w") as file:
|
||||
file.write(json.dumps(self.__dict__, indent=4))
|
||||
file.write(json.dumps(asdict(self), indent=4))
|
|
@ -1,14 +0,0 @@
|
|||
import json
|
||||
|
||||
def load_config(file_path):
|
||||
with open(file_path, 'r') as file:
|
||||
config = json.load(file)
|
||||
return link_uuids(config)
|
||||
|
||||
def link_uuids(config):
|
||||
uuids = config["UUIDS"]
|
||||
for key, galaxy_config in config["GALAXY_CONFIGS"].items():
|
||||
galaxy_config["uuid"] = uuids[key]
|
||||
for key, cluster_config in config["CLUSTER_CONFIGS"].items():
|
||||
cluster_config["uuid"] = uuids[key]
|
||||
return config
|
|
@ -1,6 +0,0 @@
|
|||
import re
|
||||
|
||||
def extract_links(text):
|
||||
links = re.findall(r'\[([^\]]+)\]\((https?://[^\s\)]+)\)', text)
|
||||
urls = set([url for text, url in links])
|
||||
return urls
|
Loading…
Reference in a new issue