|
| 1 | ++++ |
| 2 | +title = "Creating Enrichments" |
| 3 | +description = "Enrich entity data using SCOT with Airflow" |
| 4 | +weight = 3 |
| 5 | ++++ |
| 6 | + |
| 7 | +## SCOT enrichments |
| 8 | +SCOT entity enrichments provide the capability to incorporate data enrichment or other actions on flaired entities. These actions occur automatically when an entity is flaired. Actions can be anything from gathering data associated with the entity for display within the entity's flair pane in SCOT, or other types of action that can be taken outside of SCOT. SCOT works with the open-source workflow management platform, **Apache Airflow**, to launch actions based on the entity's type. |
| 9 | + |
| 10 | +## The enrichment workflow |
| 11 | +- In the SCOT API configuration file, specify the following required fields: |
| 12 | + - enrichmentApiJobEndpoint (must be in this format WITH PLACEHOLDER INCLUDED): "/api/v1/dags/scot_entity_[ENTITY_TYPE_PLACEHOLDER]_enrichment/dagRuns" |
| 13 | + - enrichmentHost (your deployed airflow instance url): "https://airflow.example.com" |
| 14 | + - enrichmentUsername: "scot4-enrichment-account-name" |
| 15 | + - enrichmentPassword: "scot4-enrichment-account-password" |
| 16 | + - enrichmentTypes (semicolon-delimited list of entity types): "ipaddr;domain;email;etc" |
| 17 | +- In Airflow, jobs are defined as Python scripts called a DAG. Each DAG you create corresponds to the workflow of enrichment(s) you want performed on a given entity type. |
| 18 | +- when an entity is newly created and flaired, SCOT automatically checks the entity type and, if it matches any of the types specified in the enrichment configuration, it will send a request to your Airflow server with the entity information and a callback URL to capture results |
| 19 | +- Airflow uses the endpoint specified in the request (scot4_entity_[ENTITY]_enrichment) to match against the names of the DAGs you've created (NOTE: the name you give your created DAGs must match this same scheme) and will trigger the DAG. |
| 20 | +- Once the DAG completes, Airflow will make a POST request back to SCOT's /entity/enrichment endpoint, including any collected data as directed by the DAG (exmaple below). |
| 21 | +- SCOT will add the enrichment data to the entity. |
| 22 | +- In SCOT, you can click on the flaired entity to open the flair pane. In the flair pane, you will see the enrichment name(s) appear as tabs. Click on each enrichment tab to see the latest enrichment data for this entity. From this view you can also view previous enrichment data results, and now send a new enrichment request for potentially updated data. |
| 23 | + |
| 24 | +## Deploying Apache Airflow |
| 25 | +Directions for deploying an Airflow instance can be found at https://airflow.apache.org/ |
| 26 | + |
| 27 | +## Creating a new DAG in Airflow |
| 28 | +You can review the docs for making DAG workflows at https://airflow.apache.org/docs/apache-airflow/stable/index.html. |
| 29 | +SCOT enrichments are designed such that one DAG corresponds to one entity type. Within that one DAG, you can have one or more tasks (python functions with an @Task decorator) defined for that enrichment type, for example: |
| 30 | + |
| 31 | +A DAG to enrich an IPv4 entity can have: |
| 32 | +1. A task to collect enrichment data from an external 3rd party source |
| 33 | +2. A task to collect enrichment data from an internal source |
| 34 | +3. A task to add the IP to an internal blocklist if the enrichment data meets certain criteria |
| 35 | +4. A task that returns new entity class IDs to the /entity/{entity_id}/entity_class endpoint in SCOT based on enrichment data |
| 36 | + |
| 37 | +You are limited only by what you can do programmatically in python and using APIs |
| 38 | + |
| 39 | +## Sample DAG |
| 40 | +Below is a simple DAG workflow you can use as a starting place. It's made for enrichment of an IPv4 address (but could be modified for any entity type). Based on how you set up your Airflow instance and define your DAG endpoints, you should name this script to match the endpoint. In this case, where we define the endpoint (see workflow above) as `/api/v1/dags/scot_entity_[ENTITY_TYPE_PLACEHOLDER]_enrichment/dagRuns`, and assuming you have an entity type named `ipaddr`, your DAG should be named: `scot_entity_ipaddr_enrichment` |
| 41 | + |
| 42 | +``` |
| 43 | +import os |
| 44 | +import json |
| 45 | +import pendulum |
| 46 | +
|
| 47 | +from airflow.utils.task_group import TaskGroup |
| 48 | +from airflow.decorators import task, dag, task_group |
| 49 | +from airflow.models import Variable |
| 50 | +from airflow.operators.bash import BashOperator |
| 51 | +from airflow.operators.python import get_current_context |
| 52 | +from airflow.models.log import Log |
| 53 | +from airflow.utils.db import create_session |
| 54 | +from airflow.models import Variable |
| 55 | +
|
| 56 | +@dag( |
| 57 | + schedule_interval=None, |
| 58 | + start_date=pendulum.datetime(2021, 1, 1, tz="UTC"), |
| 59 | + catchup=False, |
| 60 | + tags=['author:user', 'scot4_enrichment'], |
| 61 | + doc_md=open(f"{os.path.dirname(os.path.realpath(__file__))}/README.md").read(), |
| 62 | + params={'entity_id': None, 'entity_value': '192.168.1.1', 'callback_url':None} |
| 63 | +) |
| 64 | +def scot4_entity_ipaddr_enrichment(): |
| 65 | + @task |
| 66 | + def task1_enrichment(): |
| 67 | + import pandas as pd |
| 68 | + # get the IP address from the Airflow context |
| 69 | + context = get_current_context() |
| 70 | + ip_addr = context['params'].get('entity_value') |
| 71 | + # query enrichment data for IP from 3rd party API service |
| 72 | + enr_data = get_data_from_api(ip_addr) |
| 73 | + # convert any results to markdown so SCOT can parse it |
| 74 | + markdown = pd.DataFrame.from_records([enr_data]).T.to_markdown() |
| 75 | + # use this schema for SCOT parsing (title value = tab name in the flair pane) |
| 76 | + enrichment_data = { 'title': 'Data Enrichment Summary', |
| 77 | + 'enrichment_class': 'markdown', |
| 78 | + 'description': 'Summary from API being used', |
| 79 | + 'data': {'markdown': markdown,} |
| 80 | + } |
| 81 | + return {'entity_class_ids': entity_class_ids, 'enrichment_data': [enrichment_data]} |
| 82 | +
|
| 83 | +
|
| 84 | + @task |
| 85 | + def add_enrichment_to_scot(results): |
| 86 | + import requests |
| 87 | +
|
| 88 | + context = get_current_context() |
| 89 | + callback_url = context['params'].get('callback_url') |
| 90 | + api_key = Variable.get('scot4-instance-api-key') |
| 91 | +
|
| 92 | + if callback_url is not None and results.get('enrichment_data') is not None and len(results['enrichment_data']) > 0: |
| 93 | + for enrichment_data in results['enrichment_data']: |
| 94 | + context = get_current_context() |
| 95 | + entity_id = context['params'].get('entity_id') |
| 96 | + url = f"{callback_url}/entity/{entity_id}/enrichment" |
| 97 | + res = requests.post(url, data=json.dumps(enrichment_data), headers= |
| 98 | + 'Content-Type':'application/json', |
| 99 | + 'Authorization': f'apikey {api_key}', |
| 100 | + }) |
| 101 | + if not res.ok: |
| 102 | + raise Exception(f"Request to {url} failed: {res.status_code} {res.reason}: {res.text}") |
| 103 | +
|
| 104 | +
|
| 105 | + task1_results = task1_enrichment() |
| 106 | + add_enrichment_to_scot(task1_results) |
| 107 | +
|
| 108 | +dag = scot4_entity_ipaddr_enrichment() |
| 109 | +``` |
0 commit comments