\" # specify desired directory name\r\n",
"new_path = adls_path + dir_name\r\n",
"csl_beacon_pd = csl_beacon_df.coalesce(1).write.format(\"json\").save(new_path)\r\n"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "markdown",
"source": [
"## Stop Spark Session"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"%synapse stop"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "markdown",
"source": [
"## Export results from ADLS to local filesystem\r\n"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"def initialize_storage_account(storage_account_name, storage_account_key):\r\n",
"\r\n",
" try:\r\n",
" global service_client\r\n",
"\r\n",
" service_client = DataLakeServiceClient(\r\n",
" account_url=\"{}://{}.dfs.core.windows.net\".format(\r\n",
" \"https\", storage_account_name\r\n",
" ),\r\n",
" credential=storage_account_key,\r\n",
" )\r\n",
"\r\n",
" except Exception as e:\r\n",
" print(e)\r\n",
"\r\n",
"\r\n",
"def list_directory_contents(container_name, input_path, file_type):\r\n",
" try:\r\n",
" file_system_client = service_client.get_file_system_client(\r\n",
" file_system=container_name\r\n",
" )\r\n",
"\r\n",
" paths = file_system_client.get_paths(path=input_path)\r\n",
"\r\n",
" pathlist = []\r\n",
"\r\n",
" for path in paths:\r\n",
" pathlist.append(path.name) if path.name.endswith(file_type) else pathlist\r\n",
"\r\n",
" return pathlist\r\n",
"\r\n",
" except Exception as e:\r\n",
" print(e)\r\n",
"\r\n",
"\r\n",
"def download_file_from_directory(container_name, input_path, input_file):\r\n",
" try:\r\n",
" file_system_client = service_client.get_file_system_client(\r\n",
" file_system=container_name\r\n",
" )\r\n",
"\r\n",
" directory_client = file_system_client.get_directory_client(input_path)\r\n",
"\r\n",
" local_file = open(\"output.json\", \"wb\")\r\n",
"\r\n",
" file_client = directory_client.get_file_client(input_file)\r\n",
"\r\n",
" download = file_client.download_file()\r\n",
"\r\n",
" downloaded_bytes = download.readall()\r\n",
"\r\n",
" local_file.write(downloaded_bytes)\r\n",
"\r\n",
" local_file.close()\r\n",
"\r\n",
" except Exception as e:\r\n",
" print(e)\r\n",
"\r\n",
"\r\n",
"def json_normalize(input_file, output_file):\r\n",
" nwbeaconList = []\r\n",
" with open(input_file) as f:\r\n",
" for jsonObj in f:\r\n",
" nwbeaconDict = json.loads(jsonObj)\r\n",
" nwbeaconList.append(nwbeaconDict)\r\n",
"\r\n",
" with open(output_file, \"w\") as write_file:\r\n",
" json.dump(nwbeaconList, write_file)"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632987153746
}
}
},
{
"cell_type": "markdown",
"source": [
"### Download the files from ADLS\r\n",
"\r\n",
"In below sections, we will provide input details about ADLS account ad then use functions to connect , list contents and download results locally.\r\n",
"\r\n",
"If you need help in locating input details, follow below steps\r\n",
"- Go to the https://web.azuresynapse.net and sign in to your workspace.\r\n",
"- In Synapse Studio, click Data, select the Linked tab, and select the container under Azure Data Lake Storage Gen2.\r\n",
"- Navigate to folder from the container, right click and select Properies.\r\n",
"- Copy ABFSS path , extact the details and map to the input fields\r\n",
"\r\n",
"\r\n",
"You can check [View account access keys](https://docs.microsoft.com/azure/storage/common/storage-account-keys-manage?tabs=azure-portal#view-account-access-keys) doc to find and retrieve your storage account keys for ADLS account.\r\n",
"\r\n",
"\r\n",
"Warning: If you are storing secrets such as storage account keys in the notebook you should
\r\n",
"probably opt to store either into msticpyconfig file on the compute instance or use\r\n",
"Azure Key Vault to store the secrets.
\r\n",
"Read more about using KeyVault\r\n",
"in the MSTICPY docs\r\n",
"
"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"# Primary storage info\r\n",
"account_name = \"\" # fill in your primary account name\r\n",
"container_name = \"\" # fill in your container name\r\n",
"subscription_id = \"\"\r\n",
"resource_group = \"\" # fill in your resource gropup for ADLS account\r\n",
"workspace_name = \"\" # fill in your la workspace name\r\n",
"input_path = f\"WorkspaceResourceId=/subscriptions/{subscription_id}/resourcegroups/{resource_group}/providers/microsoft.operationalinsights/workspaces/\"\r\n",
"\r\n",
"adls_path = f\"abfss://{container_name}@{account_name}.dfs.core.windows.net/{input_path}/{workspace_name}\"\r\n",
"\r\n",
"dir_name = \"/\" #Replace the dirname previously specified to store results from spark\r\n",
"account_key = \"\" # Replace your storage account key"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1633376635135
}
}
},
{
"cell_type": "code",
"source": [
"new_path = input_path + dir_name\r\n",
"\r\n",
"initialize_storage_account(account_name, account_key)\r\n",
"pathlist = list_directory_contents(container_name, new_path, \"json\")\r\n",
"input_file = pathlist[0].split(\"/\")[-1]\r\n",
"download_file_from_directory(container_name, new_path, input_file)\r\n",
"\r\n",
"json_normalize(\"output.json\", \"out_normalized.json\")"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632987683046
}
}
},
{
"cell_type": "markdown",
"source": [
"### Display results"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"df = pd.read_json('out_normalized.json')\r\n",
"\r\n",
"df.head()"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632987687222
}
}
},
{
"cell_type": "markdown",
"source": [
"## Enrich results \r\n",
"In this section, we will enrich entities retrieved from network beaconing behavior such as IP information.\r\n",
"Types of Enrichment which will beneficial in perfoming investigation will be IP Geolcation , Whois Registrar information and ThreatIntel lookups.\r\n",
"\r\n",
"For first time users, please refer `Getting Started Guide For Microsoft Sentinel ML Notebooks` and section [Create your configuration file](https://docs.microsoft.com/azure/sentinel/notebook-get-started#create-your-configuration-file) to create your `mstipyconfig`. "
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "markdown",
"source": [
"### IP Geolocation Enrichment\r\n",
"In this step, we will use msticpy geolocation capabilities using maxmind database. You will need maxmind API key to download the database.\r\n",
"\r\n",
"Note:\r\n",
"You may see the GeoLite driver downloading its database the first time you run this.\r\n",
"
\r\n",
"
\r\n",
"\r\n",
" Learn more about MSTICPy GeoIP providers...
\r\n",
" \r\n",
" MSTICPy GeoIP Providers\r\n",
"
\r\n",
" \r\n",
"
"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"from msticpy.sectools.geoip import GeoLiteLookup\r\n",
"\r\n",
"iplocation = GeoLiteLookup()\r\n",
"\r\n",
"df = iplocation.df_lookup_ip(df, column=\"DestinationIP\")\r\n",
"df.head()"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632990019315
}
}
},
{
"cell_type": "markdown",
"source": [
"### Whois registration enrichment\r\n",
"In this step, we can perform whois lokup on all public destination ips and populate additional information such as ASN. You can use this output to further filter known ASNs from the results."
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"num_ips = len(df[\"DestinationIP\"].unique())\r\n",
"print(f\"Performing WhoIs lookups for {num_ips} IPs \", end=\"\")\r\n",
"df[\"DestASN\"] = df.apply(lambda x: get_whois_info(x.DestinationIP, True), axis=1)\r\n",
"df[\"DestASNFull\"] = df.apply(lambda x: x.DestASN[1], axis=1)\r\n",
"df[\"DestASN\"] = df.apply(lambda x: x.DestASN[0], axis=1)\r\n",
"\r\n",
"#Display results\r\n",
"df.head()"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632989270369
}
}
},
{
"cell_type": "markdown",
"source": [
"### ThreatIntel Enrichment\r\n",
"\r\n",
"In this step, we can perform threatintel lookup using msticpy and open source TI providers such as IBM Xforce, VirusTotal, Greynoise etc. \r\n",
"Below example shows performing lookup on single IP as well as bulk lookup on all ips using IBM Xforce TI Provider. \r\n",
"
You will need to register with IBM Xforce and enter API keys into `mstipyconfig.yaml`\r\n",
"\r\n",
"\r\n",
" Learn more...
\r\n",
" \r\n",
"
\r\n",
" \r\n",
" - More details are shown in the A Tour of Cybersec notebook features notebook
\r\n",
" - Threat Intel Lookups in MSTICPy
\r\n",
" - To learn more about adding TI sources, see the TI Provider setup in the A Getting Started Guide For Microsoft Sentinel ML Notebooks notebook\r\n",
"
\r\n",
" \r\n",
"
\r\n",
"\r\n",
"\r\n"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"ti_lookup = TILookup()\r\n",
"# Perform lookup on single IOC\r\n",
"result = ti_lookup.lookup_ioc(observable=\"52.183.120.194\", providers=[\"XForce\"])\r\n",
"ti_lookup.result_to_df(result)"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632990440057
}
}
},
{
"cell_type": "code",
"source": [
"# Flattening all the desnation IPs into comma separated list\r\n",
"ip_list = df['DestinationIP'].astype(str).values.flatten().tolist()\r\n",
"\r\n",
"# Perform bulk lookup on all IPs with specified providers\r\n",
"ti_resp = ti_lookup.lookup_iocs(data=ip_list, providers=[\"AzSTI\", \"XForce\"])\r\n",
"select_ti = browse_results(ti_resp, severities=['high','warning'])\r\n",
"select_ti"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
},
"gather": {
"logged": 1632991107856
}
}
},
{
"cell_type": "markdown",
"source": [
"## Visualization\r\n",
"MSTICpy also includes a feature to allow you to map locations, this can be particularily useful when looking at the distribution of remote network connections or other events. Below we plot the locations of destination IPs observed in our results."
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "code",
"source": [
"from msticpy.nbtools import entityschema\r\n",
"from msticpy.sectools.ip_utils import convert_to_ip_entities\r\n",
"from msticpy.nbtools.foliummap import FoliumMap, get_map_center\r\n",
"\r\n",
"# Convert our IP addresses in string format into an ip address entity\r\n",
"ip_entity = entityschema.IpAddress()\r\n",
"ip_list = [convert_to_ip_entities(i)[0] for i in df['DestinationIP'].head(10)]\r\n",
" \r\n",
"# Get center location of all IP locaitons to center the map on\r\n",
"location = get_map_center(ip_list)\r\n",
"logon_map = FoliumMap(location=location, zoom_start=4)\r\n",
"\r\n",
"# Add location markers to our map and dsiplay it\r\n",
"if len(ip_list) > 0:\r\n",
" logon_map.add_ip_cluster(ip_entities=ip_list)\r\n",
"display(logon_map.folium_map)"
],
"outputs": [],
"execution_count": null,
"metadata": {
"jupyter": {
"source_hidden": false,
"outputs_hidden": false
},
"nteract": {
"transient": {
"deleting": false
}
}
}
},
{
"cell_type": "markdown",
"source": [
"## Conclusion\r\n",
"\r\n",
"We originally started our hunting on very large datasets of firewall logs. Due to the sheer scale of data, we leveraged spark to load the data. \r\n",
"
We then performed baselining on historical data and use it to further filter current day dataset. In the next step we performed various data transformation by using spark features such as paritioning, windowing, ranking datatset to find outbound network beaconing like behavior.\r\n",
"
In order to analyze this data further, we enrich IP entities from result dataset with additional information such as Geolocation, whois registration and threat intel lookups. \r\n",
"\r\n",
"Analysts can perform further investigation on selected IP addresses from enrichment results by correlating various data sources available. \r\n",
"You can then create incidents in Microsoft Sentinel and track investigation in it.\r\n"
],
"metadata": {
"nteract": {
"transient": {
"deleting": false
}
}
}
}
],
"metadata": {
"kernelspec": {
"name": "python38-azureml",
"language": "python",
"display_name": "Python 3.8 - AzureML"
},
"language_info": {
"name": "python",
"version": "3.8.1",
"mimetype": "text/x-python",
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"pygments_lexer": "ipython3",
"nbconvert_exporter": "python",
"file_extension": ".py"
},
"kernel_info": {
"name": "python38-azureml"
},
"microsoft": {
"host": {
"AzureML": {
"notebookHasBeenCompleted": true
}
}
},
"nteract": {
"version": "nteract-front-end@1.0.0"
}
},
"nbformat": 4,
"nbformat_minor": 0
}