notebooks/CyberML - Anomalous Access Detection.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# CyberML - Anomalous Access Detection\n",
    "\n",
    "Here we demonstrate a novel CyberML model which can learn user access patterns and then automatically detect anomalous user access based on learned behavior.\n",
    "The model internally uses Collaborative Filtering for Implicit Feedback as published here: http://yifanhu.net/PUB/cf.pdf\n",
    "and is based on Apache Spark's implementation of this: https://spark.apache.org/docs/2.2.0/ml-collaborative-filtering.html.\n",
    "\n",
    "This notebook demonstrates a usage example of Anomalous Resource Access model.\n",
    "All the model requires is a dataset in which there are 'users' which access 'resources'.\n",
    "The model is based on Collaborative Filtering and it uses Machine Learning to learn access patterns of users and resources.\n",
    "When a user accesses a resource which is outside of the user's learned profile then this access recieves a high anomaly score.\n",
    "\n",
    "In this notebook we provide a usage example and a synthetic dataset in which there are 3 departments:\n",
    "(1) Finance, (2) HR and (3) Engineering.\n",
    "In the training data users access only a subset of resources from their own departments.\n",
    "To evaluate the model we use two datasets.\n",
    "The first contains access patterns unseen during training in which users access resources within their departments (again, resources they didn't access during training but within their department).\n",
    "The latter contains users accessing resources from outside their department.\n",
    "We then use the model to assign anomaly scores expecting that the first get low anomaly scores and the latter recieve high anomaly scores.\n",
    "This is what this example demonstrates.\n",
    "\n",
    "Note: the data does NOT contain information about departments, this information is implictly learned by the model by analyzing the access patterns."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create an Azure Databricks cluster and install the following libs\n",
    "\n",
    "1. In Cluster Libraries install from library source Maven:\n",
    "Coordinates: com.microsoft.ml.spark:mmlspark_2.11:1.0.0-rc3\n",
    "Repository: https://mmlspark.azureedge.net/maven\n",
    "\n",
    "2. In Cluster Libraries install from PyPI the library called plotly"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup & Initialization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# this is used to produce the synthetic dataset for this test\n",
    "from mmlspark.cyber.dataset import DataFactory\n",
    "\n",
    "# the access anomalies model generator\n",
    "from mmlspark.cyber.anomaly.collaborative_filtering import AccessAnomaly\n",
    "\n",
    "from pyspark.sql import functions as f, types as t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sparkContext.setCheckpointDir('dbfs:/checkpoint_path/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Loadup datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "factory = DataFactory(\n",
    "  num_hr_users = 25,\n",
    "  num_hr_resources = 50,\n",
    "  num_fin_users = 35,\n",
    "  num_fin_resources = 75,\n",
    "  num_eng_users = 15,\n",
    "  num_eng_resources = 25,\n",
    "  single_component = True\n",
    ")\n",
    "\n",
    "training_pdf = factory.create_clustered_training_data(ratio=0.4)\n",
    "\n",
    "# a tenant id is used when independant datasets originate from different tenants, in this example we set all tenants-ids to the same value\n",
    "training_df = spark.createDataFrame(training_pdf).withColumn('tenant_id', f.lit(0))\n",
    "ingroup_df = spark.createDataFrame(factory.create_clustered_intra_test_data(training_pdf)).withColumn('tenant_id', f.lit(0))\n",
    "outgroup_df = spark.createDataFrame(factory.create_clustered_inter_test_data()).withColumn('tenant_id', f.lit(0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(training_df.count())\n",
    "print(ingroup_df.count())\n",
    "print(outgroup_df.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model setup & training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "access_anomaly = AccessAnomaly(\n",
    "  tenantCol='tenant_id',\n",
    "  userCol='user',\n",
    "  resCol='res',\n",
    "  likelihoodCol='likelihood',\n",
    "  maxIter=1000\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = access_anomaly.fit(training_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Apply model & show result stats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ingroup_scored_df = model.transform(ingroup_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ingroup_scored_df.agg(\n",
    "  f.min('anomaly_score').alias('min_anomaly_score'),\n",
    "  f.max('anomaly_score').alias('max_anomaly_score'),\n",
    "  f.mean('anomaly_score').alias('mean_anomaly_score'),\n",
    "  f.stddev('anomaly_score').alias('stddev_anomaly_score'),\n",
    ").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "outgroup_scored_df = model.transform(outgroup_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "outgroup_scored_df.agg(\n",
    "  f.min('anomaly_score').alias('min_anomaly_score'),\n",
    "  f.max('anomaly_score').alias('max_anomaly_score'),\n",
    "  f.mean('anomaly_score').alias('mean_anomaly_score'),\n",
    "  f.stddev('anomaly_score').alias('stddev_anomaly_score'),\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Examine results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Select a subset of results to send to Log Analytics\n",
    "#\n",
    "\n",
    "full_res_df = outgroup_scored_df.orderBy(f.desc('anomaly_score')).cache()\n",
    "\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "w = Window.partitionBy(\n",
    "                  'tenant_id',\n",
    "                  'user',\n",
    "                  'res'  \n",
    "                ).orderBy(\n",
    "                  f.desc('anomaly_score')\n",
    "                )\n",
    "\n",
    "# select values above threshold\n",
    "results_above_threshold = full_res_df.filter(full_res_df.anomaly_score > 1.0)\n",
    "\n",
    "# get distinct resource/user and corresponding timestamp and highest score\n",
    "results_to_la = results_above_threshold.withColumn(\n",
    "                  'index', f.row_number().over(w)\n",
    "                  ).orderBy(\n",
    "                    f.desc('anomaly_score')\n",
    "                  ).select(\n",
    "                    'tenant_id',\n",
    "                    f.col('user'),\n",
    "                    f.col('res'),\n",
    "                    'anomaly_score'\n",
    "                  ).where(\n",
    "                    'index == 1'\n",
    "                  ).limit(100).cache()\n",
    "\n",
    "# add a fake timestamp to the results\n",
    "results_to_la = results_to_la.withColumn('timestamp', f.current_timestamp())\n",
    "  \n",
    "display(results_to_la)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Display all resource accesses by users with highest anomalous score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from plotly import __version__\n",
    "from plotly.offline import download_plotlyjs, init_notebook_mode, plot, iplot, offline\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "print (__version__) # requires version >= 1.9.0\n",
    "\n",
    "# run plotly in offline mode\n",
    "offline.init_notebook_mode()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Find all server accesses of users with high predicted scores\n",
    "# For display, limit to top 25 results\n",
    "results_to_display = results_to_la.orderBy(\n",
    "                  f.desc('anomaly_score')\n",
    "                ).limit(25).cache()\n",
    "interesting_records = full_res_df.join(results_to_display, ['user'], 'left_semi')\n",
    "non_anomalous_records = interesting_records.join(results_to_display, ['user', 'res'], 'left_anti')\n",
    "\n",
    "top_non_anomalous_records = non_anomalous_records.groupBy(\n",
    "                          'tenant_id',\n",
    "                          'user', \n",
    "                          'res'\n",
    "                        ).agg(\n",
    "                          f.count('*').alias('count'),\n",
    "                        ).select(\n",
    "                          f.col('tenant_id'),\n",
    "                          f.col('user'),\n",
    "                          f.col('res'),\n",
    "                          'count'\n",
    "                        )\n",
    "\n",
    "#pick only a subset of non-anomalous record for UI\n",
    "w = Window.partitionBy(\n",
    "                  'tenant_id',\n",
    "                  'user',\n",
    "                ).orderBy(\n",
    "                  f.desc('count')\n",
    "                )\n",
    "\n",
    "# pick top non-anomalous set\n",
    "top_non_anomalous_accesses = top_non_anomalous_records.withColumn(\n",
    "                  'index', f.row_number().over(w)\n",
    "                  ).orderBy(\n",
    "                    f.desc('count')\n",
    "                  ).select(\n",
    "                    'tenant_id',\n",
    "                    f.col('user'),\n",
    "                    f.col('res'),\n",
    "                    f.col('count')\n",
    "                  ).where(\n",
    "                    'index in (1,2,3,4,5)'\n",
    "                  ).limit(25)\n",
    "\n",
    "# add back anomalous record\n",
    "fileShare_accesses = (top_non_anomalous_accesses\n",
    "                          .select('user', 'res', 'count')\n",
    "                          .union(results_to_display.select('user', 'res', f.lit(1).alias('count'))).cache())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get unique users and file shares\n",
    "high_scores_df = fileShare_accesses.toPandas()\n",
    "unique_arr = np.append(high_scores_df.user.unique(), high_scores_df.res.unique())\n",
    "\n",
    "unique_df = pd.DataFrame(data = unique_arr, columns = ['name'])\n",
    "unique_df['index'] = range(0, len(unique_df.index))\n",
    "\n",
    "# create index for source & target and color for the normal accesses\n",
    "normal_line_color = 'rgba(211, 211, 211, 0.8)'\n",
    "anomolous_color = 'red'\n",
    "x = pd.merge(high_scores_df, unique_df, how='left', left_on='user', right_on='name').drop(['name'], axis=1).rename(columns={'index' : 'userIndex'})\n",
    "all_access_index_df = pd.merge(x, unique_df, how='left', left_on='res', right_on='name').drop(['name'], axis=1).rename(columns={'index' : 'resIndex'})\n",
    "all_access_index_df['color'] = normal_line_color\n",
    "\n",
    "# results_to_display index, color and \n",
    "y = results_to_display.toPandas().drop(['tenant_id', 'timestamp', 'anomaly_score'], axis=1)\n",
    "y = pd.merge(y, unique_df, how='left', left_on='user', right_on='name').drop(['name'], axis=1).rename(columns={'index' : 'userIndex'})\n",
    "high_scores_index_df = pd.merge(y, unique_df, how='left', left_on='res', right_on='name').drop(['name'], axis=1).rename(columns={'index' : 'resIndex'})\n",
    "high_scores_index_df['count'] = 1\n",
    "high_scores_index_df['color'] = anomolous_color\n",
    "\n",
    "# substract 1 for the red entries in all_access df\n",
    "hsi_df = high_scores_index_df[['user','res', 'count']].rename(columns={'count' : 'hsiCount'})\n",
    "all_access_updated_count_df = pd.merge(all_access_index_df, hsi_df, how='left', left_on=['user', 'res'], right_on=['user', 'res'])\n",
    "all_access_updated_count_df['count'] = np.where(all_access_updated_count_df['hsiCount']==1, all_access_updated_count_df['count'] - 1, all_access_updated_count_df['count'])\n",
    "all_access_updated_count_df = all_access_updated_count_df.loc[all_access_updated_count_df['count'] > 0]\n",
    "all_access_updated_count_df = all_access_updated_count_df[['user','res', 'count', 'userIndex', 'resIndex', 'color']]\n",
    "\n",
    "# combine the two tables\n",
    "frames = [all_access_updated_count_df, high_scores_index_df]\n",
    "display_df = pd.concat(frames, sort=True)\n",
    "# display_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_trace = dict(\n",
    "    type='sankey',\n",
    "    domain = dict(\n",
    "      x =  [0,1],\n",
    "      y =  [0,1]\n",
    "    ),\n",
    "    orientation = \"h\",\n",
    "    valueformat = \".0f\",\n",
    "    node = dict(\n",
    "      pad = 10,\n",
    "      thickness = 30,\n",
    "      line = dict(\n",
    "        color = \"black\",\n",
    "        width = 0\n",
    "      ),\n",
    "      label = unique_df['name'].dropna(axis=0, how='any')\n",
    "    ),\n",
    "    link = dict(\n",
    "      source = display_df['userIndex'].dropna(axis=0, how='any'),\n",
    "      target = display_df['resIndex'].dropna(axis=0, how='any'),\n",
    "      value = display_df['count'].dropna(axis=0, how='any'),\n",
    "      color = display_df['color'].dropna(axis=0, how='any'),\n",
    "  )\n",
    ")\n",
    "\n",
    "layout =  dict(\n",
    "    title = \"All resources accessed by users with highest anomalous scores\",\n",
    "    height = 772,\n",
    "    font = dict(\n",
    "      size = 10\n",
    "    ),    \n",
    ")\n",
    "\n",
    "fig = dict(data=[data_trace], layout=layout)\n",
    "\n",
    "p = plot(fig, output_type='div')\n",
    "\n",
    "displayHTML(p)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookName": "byoml_demo",
   "notebookOrigID": 659272127833844,
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}