STIX 2.0 Training.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TAXII 2.0 Tutorial\n",
    "\n",
    "Connect to the UberTAXII server, list API roots, and pull some STIX data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# A \"Server\" object pulls general information about the server, and gives you a list of API roots.\n",
    "# Connect to it by providing the URL, username, and password.\n",
    "\n",
    "from taxii2client import Server\n",
    "server = Server('https://ubertaxii.com/taxii/', user='ubertaxii', password='certain_players_want_a_pony')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print some basic information, available without making additional requests.\n",
    "\n",
    "print(\"Title:\\t\\t\" + server.title)\n",
    "print(\"Description:\\t\" + server.description)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Iterate through the API roots. Note that addtional requests are made automatically to get the title and description.\n",
    "\n",
    "for api_root in server.api_roots:\n",
    "    print(api_root.title)\n",
    "    print(api_root.description)\n",
    "    print(api_root.url)\n",
    "    print(\"\\n\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the list of collections for an API root (using MITRE's)\n",
    "\n",
    "api_root = server.api_roots[3]\n",
    "for collection in api_root.collections:\n",
    "    print(collection.title)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# STIX 2.0 Tutorial\n",
    "\n",
    "Starting with our reference to a TAXII server, do some stuff with STIX data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from stix2 import TAXIICollectionSource, Filter\n",
    "\n",
    "taxii_source = TAXIICollectionSource(api_root.collections[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = taxii_source.query(Filter(\"type\", \"=\", \"intrusion-set\"))\n",
    "print(\"Got {} results\".format(len(results)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "apt30 = results[10]\n",
    "print(apt30.name) # Note that it's a python-stix2 object already, no need to parse the JSON"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# What malware and tools does that intrusion set use?\n",
    "\n",
    "# Find relationships where the source is that intrusion set\n",
    "all_relationships = taxii_source.query(Filter(\"type\", \"=\", \"relationship\"))\n",
    "relationships = [r for r in all_relationships if r.source_ref == apt30.id]\n",
    "print(\"Found {} relationships\".format(len(relationships)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for rel in relationships:\n",
    "    target = taxii_source.get(rel.target_ref)\n",
    "    print(\"{} {} {} ({})\".format(apt30.name, rel.relationship_type, target.name, target.type))\n",
    "\n",
    "# Gotcha! Note that this does not optimize the number of calls.\n",
    "# You still need to know when a TAXII request is going to be made and when the result is available locally.\n",
    "# Also, watch for n+1 queries (like here) and try to optimize if possible"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Make it Easy: Environments\n",
    "Use environments to add default sources, sinks, etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from stix2 import Environment, MemoryStore\n",
    "\n",
    "taxiisource = TAXIICollectionSource(api_root.collections[0])\n",
    "env = Environment(source=taxiisource)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "env.get(apt30.id).name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Try the other direction\n",
    "rels = env.query(Filter(\"source_ref\", \"=\", apt30.id))\n",
    "print(len(rels))\n",
    "\n",
    "# Note the delay: not all filters happen server-side,\n",
    "# it sometimes transparently just pulls all data and filters at the client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "malware = env.get(rels[3].target_ref)\n",
    "malware.name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sharing back: Creating STIX\n",
    "\n",
    "You can also use the enviornment to set defaults so you don't have to duplicate adding standard markings, etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from stix2 import Indicator, Relationship, ObjectFactory, CompositeDataSource\n",
    "\n",
    "factory = ObjectFactory(created_by_ref=\"identity--311b2d2d-f010-5473-83ec-1edf84858f4c\")\n",
    "source = CompositeDataSource()\n",
    "memorystore = MemoryStore()\n",
    "source.add_data_sources([memorystore.source, taxiisource])\n",
    "env2 = Environment(factory=factory, source=source, sink=memorystore)\n",
    "\n",
    "ind = factory.create(Indicator,\n",
    "                     labels=[\"malicious-activity\"],\n",
    "                     pattern=\"[file:hashes.md5 = 'd41d8cd98f00b204e9800998ecf8427e']\")\n",
    "rel = factory.create(Relationship, source_ref=ind.id, target_ref=malware.id, relationship_type=\"indicates\")\n",
    "print(rel)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Push to our data source\n",
    "env2.add(ind)\n",
    "env2.add(rel)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Is it still there?\n",
    "print(env2.get(rel.id))\n",
    "\n",
    "# NOTE: This could have included a push to TAXII with the TAXII source/sink"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Something quick? Use the workbench\n",
    "\n",
    "The workbench layer is essentially a default environment and some helpers to automatically work with it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Jupyter notebook thing...restart our kernel\n",
    "from IPython.core.display import HTML\n",
    "HTML(\"<script>Jupyter.notebook.kernel.restart()</script>\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import the workbench utilities\n",
    "from stix2.workbench import *\n",
    "from taxii2client import Collection\n",
    "\n",
    "# ATT&CK\n",
    "collection = Collection(\"https://ubertaxii.com/taxii/mitre/collections/b8291031-66d3-4402-ad0e-0c2e6f0df3a2/\", user=\"ubertaxii\", password=\"certain_players_want_a_pony\")\n",
    "tc_source = TAXIICollectionSource(collection)\n",
    "add_data_source(tc_source)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = attack_patterns()\n",
    "print(len(response))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ap = response[127]\n",
    "print(\"Attack Pattern: \" + ap.name)\n",
    "\n",
    "def not_me(item):\n",
    "    if item.source_ref != ap.id:\n",
    "        return item.source_ref\n",
    "    return item.target_ref\n",
    "    \n",
    "\n",
    "for rel in ap.relationships():\n",
    "    print(\"Related to: {} ({})\".format(get(not_me(rel)).name), rel.relationship_type)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}