MISP
/
PyMISP
mirror of https://github.com/MISP/PyMISP
2
0
Fork 0
Code Issues Releases Wiki Activity

Merge remote-tracking branch 'upstream/master'

pull/312/head
Steve Clement 2018-12-10 10:49:16 +01:00
parent 4eaf958ac2 e0bf5b9d8f
commit 91fe83a475
18 changed files with 999 additions and 221 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
CHANGELOG.txt
View File

@ -2,6 +2,103 @@ Changelog
=========
v2.4.99 (2018-12-06)
--------------------
New
~~~
- Auto generate doc for PyMISPExpanded. [Raphaël Vinot]
- Search_index in ExpandedPyMISP, cleanup, update jupyter. [Raphaël
Vinot]
- Add log search. [Raphaël Vinot]
- Add test for pushing an event to ZMQ. [Raphaël Vinot]
- Change_distribution method. [Raphaël Vinot]
- Add test cases for sightings, cleanup. [Raphaël Vinot]
- [example] Added sighting rest search example. [Sami Mokaddem]
- [sighting] Added support of sighting REST API. [Sami Mokaddem]
- Allow to pass csv to return_format in search. [Raphaël Vinot]
- Page/limit in search. [Raphaël Vinot]
Changes
~~~~~~~
- Bump Changelog. [Raphaël Vinot]
- Bump version. [Raphaël Vinot]
- Bump misp-objects & describeTypes. [Raphaël Vinot]
- Bump Changelog. [Raphaël Vinot]
- Version bump. [Raphaël Vinot]
- Bump misp-objects. [Raphaël Vinot]
- Add test cases for default distribution levels. [Raphaël Vinot]
- Include proposals in attributes search. [Dawid Czarnecki]
Add includeProposals param to the search method
- Bump misp-objects. [Raphaël Vinot]
- Update readme to document testing. [Raphaël Vinot]
- Fixes & update Jupyter. [Raphaël Vinot]
- [tuto] Update search. [Raphaël Vinot]
- Add a script to load the API key from the file system (training VM)
[Raphaël Vinot]
- Bump misp-objects. [Raphaël Vinot]
- Add print in testlive to debug travis. [Raphaël Vinot]
- Bump objects. [Raphaël Vinot]
Fix
~~~
- Auto generate doc for PyMISPExpanded. [Raphaël Vinot]
- Test failing on travis... [Raphaël Vinot]
- Properly handle errors on event creation/update. [Raphaël Vinot]
- Test case. [Raphaël Vinot]
- Do not run the zmq test on travis. [Raphaël Vinot]
- Type of quick_filter. [Raphaël Vinot]
- Quick_filter was broken. [Raphaël Vinot]
- Properly initialize the config when jupyter runs on the VM. [Raphaël
Vinot]
- Travis run. [Raphaël Vinot]
- Readme update + python3 + pep8. [Christophe Vandeplas]
align python path to readme specifying python3
- Feed-generator gitignore. [Christophe Vandeplas]
- Test cases. [Raphaël Vinot]
Other
~~~~~
- Merge branch 'master' of github.com:MISP/PyMISP. [Raphaël Vinot]
- Merge pull request #310 from DragonDev1906/master. [Raphaël Vinot]
Added get_object & get_attribute (by ID)
- Dded get_object & get_attribute. [DragonDev1906]
- Merge pull request #307 from garanews/patch-1. [Raphaël Vinot]
fix for last pymisp version
- Fix for last pymisp version. [garanews]
- Merge branch 'master' of github.com:MISP/PyMISP. [Raphaël Vinot]
- Merge pull request #305 from dawid-
czarnecki/feature/include_proposals. [Raphaël Vinot]
chg: Include proposals in attributes search
- Merge pull request #301 from deralexxx/patch-7. [Raphaël Vinot]
mention virtualenv
- Mention virtualenv. [Alexander J]
mide make sense for people who want to use it with virtualenv
- Merge branch 'master' of github.com:MISP/PyMISP. [Raphaël Vinot]
- Be more precise with the supported time indicators. [Sascha
Rommelfangen]
- Fixed documentation bug. [Sascha Rommelfangen]
- Merge branch 'master' of github.com:MISP/PyMISP. [Raphaël Vinot]
- Merge pull request #295 from 3c7/fix/search_index_date. [Raphaël
Vinot]
Fixes date parameters for search_index() function
- Fixes date parameters for search_index() function. [Nils Kuhnert]
- Merge branch 'sightingAPI' [Raphaël Vinot]
- Merge branch 'master' into sightingAPI. [Raphaël Vinot]
- Merge pull request #285 from juju4/devel. [Raphaël Vinot]
align examples on custom usage of misp_verifycert
- Align examples on custom usage of misp_verifycert. [juju4]
v2.4.96 (2018-10-12)
--------------------
@ -22,6 +119,7 @@ New
Changes
~~~~~~~
- Bump changelog. [Raphaël Vinot]
- Bump version. [Raphaël Vinot]
- Bump misp-objects. [Raphaël Vinot]
- Allow to pass a json string to direct_call. [Raphaël Vinot]
@ -32,6 +130,7 @@ Changes
Fix
~~~
- Test cases sample files. [Raphaël Vinot]
- Prevent checking length on a integer. [Sami Mokaddem]
- Direct call & add example. [Raphaël Vinot]
- Disable test for travis, take 2. [Raphaël Vinot]

32
README.md
View File

@ -26,7 +26,37 @@ pip3 install pymisp
```
git clone https://github.com/MISP/PyMISP.git && cd PyMISP
git submodule update --init
pip3 install -I .
pip3 install -I .[fileobjects,neo,openioc,virustotal]
```
## Installing it with virtualenv
It is recommended to use virtualenv to not polute your OS python envirenment.
```
pip3 install virtualenv
git clone https://github.com/MISP/PyMISP.git && cd PyMISP
python3 -m venv ./
source venv/bin/activate
git submodule update --init
pip3 install -I .[fileobjects,neo,openioc,virustotal]
```
## Running the tests
```bash
pip3 install -U nose pip setuptools coveralls codecov requests-mock
pip3 install git+https://github.com/kbandla/pydeep.git
git clone https://github.com/viper-framework/viper-test-files.git tests/viper-test-files
nosetests-3.4 --with-coverage --cover-package=pymisp,tests --cover-tests tests/test_*.py
```
If you have a MISP instance to test against, you can also run the live ones:
**Note**: You need to update the key in `tests/testlive_comprehensive.py` to the automation key of your admin account.
```bash
nosetests-3.4 --with-coverage --cover-package=pymisp,tests --cover-tests tests/testlive_comprehensive.py
```
## Samples and how to use PyMISP

6
docs/source/modules.rst
View File

@ -14,6 +14,12 @@ PyMISP
.. autoclass:: PyMISP
:members:
PyMISPExpanded (Python 3.6+ only)
---------------------------------
.. autoclass:: PyMISPExpanded
:members:
MISPAbstract
------------

2
docs/tutorial/PyMISP Objects.ipynb
View File

@ -47,7 +47,7 @@
"\n",
"\n",
"```bash\n",
"pip install jupyter\n",
"pip3 install jupyter\n",
"cd docs/tutorial\n",
"jupyter-notebook\n",
"```"

47
docs/tutorial/PyMISP_tutorial.ipynb
View File

@ -100,6 +100,19 @@
"print(\"Event id: %s\" % event.id)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"event = misp_old.new_event(distribution=1,\n",
" threat_level_id=1,\n",
" analysis=1,\n",
" info=\"Event from notebook\")\n",
"print(\"Event id: %s\" % event['Event']['id'])"
]
},
{
"cell_type": "markdown",
"metadata": {},
@ -143,6 +156,17 @@
"print(event)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Fetch by ID\n",
"event = misp_old.get_event(event_id)\n",
"print(event)"
]
},
{
"cell_type": "markdown",
"metadata": {},
@ -297,6 +321,15 @@
" print(event['id'], ':', event['info'])"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"results[0]"
]
},
{
"cell_type": "markdown",
"metadata": {},
@ -387,17 +420,17 @@
"outputs": [],
"source": [
"# The URL of the MISP instance to connect to\n",
"misp_url = 'http://127.0.0.1:8080/'\n",
"#misp_url = 'http://127.0.0.1:8080/'\n",
"# Can be found in the MISP web interface under \n",
"# http://+MISP_URL+/users/view/me -> Authkey\n",
"misp_key = 'fk5BodCZw8owbscW8pQ4ykMASLeJ4NYhuAbshNjo'\n",
"#misp_key = 'BSip0zVadeFDeolkX2g7MHx8mrlr0uE04hh6CQj0'\n",
"# Should PyMISP verify the MISP certificate\n",
"misp_verifycert = False\n",
"#misp_verifycert = False\n",
"\n",
"from pymisp import PyMISP\n",
"\n",
"misp = PyMISP(misp_url, misp_key, misp_verifycert)\n",
"misp.direct_call('attributes/add/2167', {'type': 'ip-dst', 'value': '8.8.8.8'})"
"misp.direct_call('attributes/add/58', {'type': 'ip-dst', 'value': '8.11.8.8'})"
]
},
{
@ -427,12 +460,12 @@
"outputs": [],
"source": [
"# The URL of the MISP instance to connect to\n",
"misp_url = 'http://127.0.0.1:8080/'\n",
"#misp_url = 'http://127.0.0.1:8080/'\n",
"# Can be found in the MISP web interface under \n",
"# http://+MISP_URL+/users/view/me -> Authkey\n",
"misp_key = 'fk5BodCZw8owbscW8pQ4ykMASLeJ4NYhuAbshNjo'\n",
"#misp_key = 'fk5BodCZw8owbscW8pQ4ykMASLeJ4NYhuAbshNjo'\n",
"# Should PyMISP verify the MISP certificate\n",
"misp_verifycert = False\n",
"#misp_verifycert = False\n",
"\n",
"from pymisp import PyMISP\n",
"\n",

447
docs/tutorial/Search-NG.ipynb Normal file
View File

@ -0,0 +1,447 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# The URL of the MISP instance to connect to\n",
"misp_url = 'http://127.0.0.1:8080'\n",
"# Can be found in the MISP web interface under ||\n",
"# http://+MISP_URL+/users/view/me -> Authkey\n",
"misp_key = 'LBelWqKY9SQyG0huZzAMqiEBl6FODxpgRRXMsZFu'\n",
"# Should PyMISP verify the MISP certificate\n",
"misp_verifycert = False"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Getting the API key (automatically generated on the trainig VM)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from pathlib import Path\n",
"\n",
"api_file = Path('apikey')\n",
"if api_file.exists():\n",
" misp_url = 'http://127.0.0.1'\n",
" misp_verifycert = False\n",
" with open(api_file) as f:\n",
" misp_key = f.read().strip()\n",
" print(misp_key)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Initialize PyMISP - NG"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from pymisp import ExpandedPyMISP\n",
"\n",
"misp = ExpandedPyMISP(misp_url, misp_key, misp_verifycert, debug=False)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Index Search (fast, only returns events metadata)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Search unpublished events\n",
"\n",
"**WARNING**: By default, the search query will only return all the events listed on teh index page"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(published=False)\n",
"print(r)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Get the meta data of events"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(eventid=[17217, 1717, 1721, 17218])"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Search Tag & mix with other parameters"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(tags=['tlp:white'], pythonify=True)\n",
"for e in r:\n",
" print(e)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(tag='TODO:VT-ENRICHMENT', published=False)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(tag=['!TODO:VT-ENRICHMENT', 'tlp:white'], published=False) # ! means \"not this tag\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Full text search on event info field"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(eventinfo='circl')"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Search by org"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(org='CIRCL')"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Search updated events"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(timestamp='1h')"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Search full events (Slower, returns full events)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Getting timestamps"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from datetime import datetime, date, timedelta\n",
"from dateutil.parser import parse\n",
"\n",
"int(datetime.now().timestamp())\n",
"\n",
"d = parse('2018-03-24')\n",
"int(d.timestamp())\n",
"\n",
"today = int(datetime.today().timestamp())\n",
"yesterday = int((datetime.today() - timedelta(days=1)).timestamp())\n",
"\n",
"print(today, yesterday)\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"complex_query = misp.build_complex_query(or_parameters=['uibo.lembit@mail.ee', '103.195.185.222'])"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value=complex_query, pythonify=True)\n",
"print(r)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(category='Payload delivery')"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='uibo.lembit@mail.ee', metadata=True, pythonify=True) # no attributes"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(timestamp=['2h', '1h'])"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', enforceWarninglist=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', deleted=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', publish_timestamp=1521846000) # everything published since that timestamp"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', last='1d') # everything published in the last <interval>"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', timestamp=[yesterday, today]) # everything updated since that timestamp"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(value='8.8.8.8', withAttachments=True) # Return attachments"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Search for attributes"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(controller='attributes', value='8.8.8.9')"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(controller='attributes', value='wrapper.no', event_timestamp='5d') # only consider events updated since this timestamp"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"r"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Because reason"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"tag_to_remove = 'foo'\n",
"\n",
"events = misp.search(tags=tag_to_remove, pythonify=True)\n",
"\n",
"for event in events:\n",
" for tag in event.tags:\n",
" if tag.name == tag_to_remove:\n",
" print(f'Got {tag_to_remove} in {event.info}')\n",
" misp.untag(event.uuid, tag_to_remove)\n",
" break\n",
" for attribute in event.attributes:\n",
" for tag in attribute.tags:\n",
" if tag.name == tag_to_remove:\n",
" print(f'Got {tag_to_remove} in {attribute.value}')\n",
" misp.untag(attribute.uuid, tag_to_remove)\n",
" break"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"logs = misp.search_logs(model='Tag', title='tlp:white')\n",
"print(logs)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"logs = misp.search_logs(model='Event', pythonify=True)\n",
"#print(logs)\n",
"for l in logs:\n",
" print(l.title)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"log = misp.search_logs(model='Tag', title=tag_to_remove)[0]\n",
"roles = misp.get_roles_list()\n",
"for r in roles:\n",
" if r['Role']['name'] == 'User':\n",
" new_role = r['Role']['id']\n",
" break\n",
"user = misp.get_user(log['Log']['user_id'])\n",
"user['User']['role_id'] = new_role\n",
"misp.edit_user(user['User']['id'], **user['User'])"
]
}
],
"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.7"
}
},
"nbformat": 4,
"nbformat_minor": 2
}

36
docs/tutorial/Search.ipynb
View File

@ -10,7 +10,7 @@
"misp_url = 'http://127.0.0.1:8080'\n",
"# Can be found in the MISP web interface under \n",
"# http://+MISP_URL+/users/view/me -> Authkey\n",
"misp_key = 'BSip0zVadeFDeolkX2g7MHx8mrlr0uE04hh6CQj0'\n",
"misp_key = 'LBelWqKY9SQyG0huZzAMqiEBl6FODxpgRRXMsZFu'\n",
"# Should PyMISP verify the MISP certificate\n",
"misp_verifycert = False"
]
@ -52,9 +52,9 @@
"metadata": {},
"outputs": [],
"source": [
"from pymisp import ExpandedPyMISP\n",
"from pymisp import PyMISP\n",
"\n",
"misp = ExpandedPyMISP(misp_url, misp_key, misp_verifycert, debug=False)"
"misp = PyMISP(misp_url, misp_key, misp_verifycert, debug=False)"
]
},
{
@ -112,7 +112,9 @@
"metadata": {},
"outputs": [],
"source": [
"r = misp.search_index(tag='TODO:VT-ENRICHMENT')"
"r = misp.search_index(tags=['tlp:white'])\n",
"for e in r:\n",
" print(e)"
]
},
{
@ -347,7 +349,7 @@
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(controller='attributes', values='8.8.8.8')"
"r = misp.search(controller='attributes', value='8.8.8.9')"
]
},
{
@ -356,7 +358,7 @@
"metadata": {},
"outputs": [],
"source": [
"r = misp.search(controller='attributes', values='wrapper.no', event_timestamp='5d') # only consider events updated since this timestamp"
"r = misp.search(controller='attributes', value='wrapper.no', event_timestamp='5d') # only consider events updated since this timestamp"
]
},
{
@ -399,6 +401,28 @@
" break"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"logs = misp.search_logs(model='Tag', title='tlp:white')\n",
"print(logs)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"logs = misp.search_logs(model='Event', pythonify=True)\n",
"#print(logs)\n",
"for l in logs:\n",
" print(l.title)"
]
},
{
"cell_type": "code",
"execution_count": null,

46
docs/tutorial/Usage.ipynb → docs/tutorial/Usage-NG.ipynb
View File

@ -17,13 +17,13 @@
"metadata": {},
"outputs": [],
"source": [
"from pymisp import PyMISP, MISPEvent, MISPAttribute\n",
"from pymisp import ExpandedPyMISP, MISPEvent, MISPAttribute\n",
"\n",
"# The URL of the MISP instance to connect to\n",
"misp_url = 'http://127.0.0.1:8080'\n",
"# Can be found in the MISP web interface under \n",
"# http://+MISP_URL+/users/view/me -> Authkey\n",
"misp_key = 'yB8DMS8LkfYYpcVX8bN2v7xwDZDMp4bpW0sNqNGj'\n",
"misp_key = 'LBelWqKY9SQyG0huZzAMqiEBl6FODxpgRRXMsZFu'\n",
"# Should PyMISP verify the MISP certificate\n",
"misp_verifycert = False"
]
@ -65,7 +65,7 @@
"metadata": {},
"outputs": [],
"source": [
"misp = PyMISP(misp_url, misp_key, misp_verifycert)"
"misp = ExpandedPyMISP(misp_url, misp_key, misp_verifycert)"
]
},
{
@ -87,14 +87,14 @@
},
"outputs": [],
"source": [
"response = misp.search(last='1d')\n",
"response = misp.search(publish_timestamp='2h')\n",
"\n",
"events = []\n",
"for event in response['response']:\n",
" me = MISPEvent()\n",
" me.load(event)\n",
" events.append(me)\n",
"\n",
" \n",
"for e in events:\n",
" print(e)"
]
@ -105,13 +105,7 @@
"metadata": {},
"outputs": [],
"source": [
"response = misp.search(last=['3d', '2d'])\n",
"\n",
"events = []\n",
"for event in response['response']:\n",
" me = MISPEvent()\n",
" me.load(event)\n",
" events.append(me)\n",
"events = misp.search(publish_timestamp=['120m', '100m'], pythonify=True)\n",
"\n",
"for e in events:\n",
" print(e)"
@ -161,22 +155,13 @@
"metadata": {},
"outputs": [],
"source": [
"misp = PyMISP(misp_url, misp_key, misp_verifycert, debug=True)\n",
"\n",
"from datetime import datetime\n",
"ts = int(datetime.now().timestamp())\n",
"\n",
"response = misp.search(timestamp=[ts-3600, ts])\n",
"\n",
"events = []\n",
"for event in response['response']:\n",
" me = MISPEvent()\n",
" me.load(event)\n",
" events.append(me)\n",
"events = misp.search(timestamp=[ts-3600, ts], pythonify=True)\n",
"\n",
"for e in events:\n",
" print(e)\n",
" \n",
"misp = PyMISP(misp_url, misp_key, misp_verifycert) # TODO: remove when fixed"
" print(e)"
]
},
{
@ -194,7 +179,7 @@
"metadata": {},
"outputs": [],
"source": [
"response = misp.search(controller='attributes', last='1h')\n",
"response = misp.search(controller='attributes', publish_timestamp='1h')\n",
"\n",
"attributes = []\n",
"for attribute in response['response']['Attribute']:\n",
@ -212,7 +197,7 @@
"metadata": {},
"outputs": [],
"source": [
"response = misp.search(controller='attributes', last=['2h', '1h'])\n",
"response = misp.search(controller='attributes', publish_timestamp=['2h', '1h'])\n",
"\n",
"attributes = []\n",
"for attribute in response['response']['Attribute']:\n",
@ -360,13 +345,8 @@
"metadata": {},
"outputs": [],
"source": [
"response = misp.search(values=['59.157.4.2', 'hotfixmsupload.com'])\n",
"\n",
"events = []\n",
"for event in response['response']:\n",
" me = MISPEvent()\n",
" me.load(event)\n",
" events.append(me)\n",
"complex_query = misp.build_complex_query(or_parameters=['59.157.4.2', 'hotfixmsupload.com'])\n",
"events = misp.search(value=complex_query, pythonify=True)\n",
"\n",
"for e in events:\n",
" print(e)"

11
examples/add_generic_object.py
View File

@ -20,10 +20,17 @@ if __name__ == '__main__':
args = parser.parse_args()
pymisp = PyMISP(misp_url, misp_key, misp_verifycert)
template = pymisp.get_object_templates_list()
if 'response' in template.keys():
template = template['response']
try:
template_id = [x['ObjectTemplate']['id'] for x in pymisp.get_object_templates_list() if x['ObjectTemplate']['name'] == args.type][0]
template_ids = [x['ObjectTemplate']['id'] for x in template if x['ObjectTemplate']['name'] == args.type]
if len(template_ids) > 0:
template_id = template_ids[0]
else:
raise IndexError
except IndexError:
valid_types = ", ".join([x['ObjectTemplate']['name'] for x in pymisp.get_object_templates_list()])
valid_types = ", ".join([x['ObjectTemplate']['name'] for x in template])
print ("Template for type %s not found! Valid types are: %s" % (args.type, valid_types))
exit()

4
pymisp/__init__.py
View File

@ -1,4 +1,4 @@
__version__ = '2.4.96'
__version__ = '2.4.99'
import logging
import functools
import warnings
@ -35,7 +35,7 @@ try:
from .exceptions import PyMISPError, NewEventError, NewAttributeError, MissingDependency, NoURL, NoKey, InvalidMISPObject, UnknownMISPObjectTemplate, PyMISPInvalidFormat, MISPServerError, PyMISPNotImplementedYet, PyMISPUnexpectedResponse # noqa
from .api import PyMISP # noqa
from .abstract import AbstractMISP, MISPEncode, MISPTag, Distribution, ThreatLevel, Analysis # noqa
from .mispevent import MISPEvent, MISPAttribute, MISPObjectReference, MISPObjectAttribute, MISPObject, MISPUser, MISPOrganisation, MISPSighting # noqa
from .mispevent import MISPEvent, MISPAttribute, MISPObjectReference, MISPObjectAttribute, MISPObject, MISPUser, MISPOrganisation, MISPSighting, MISPLog # noqa
from .tools import AbstractMISPObjectGenerator # noqa
from .tools import Neo4j # noqa
from .tools import stix # noqa

30
pymisp/api.py
View File

@ -336,6 +336,24 @@ class PyMISP(object):
response = self._prepare_request('GET', url)
return self._check_response(response)
def get_object(self, obj_id):
"""Get an object
:param obj_id: Object id to get
"""
url = urljoin(self.root_url, 'objects/view/{}'.format(obj_id))
response = self._prepare_request('GET', url)
return self._check_response(response)
def get_attribute(self, att_id):
"""Get an attribute
:param att_id: Attribute id to get
"""
url = urljoin(self.root_url, 'attributes/view/{}'.format(att_id))
response = self._prepare_request('GET', url)
return self._check_response(response)
def add_event(self, event):
"""Add a new event
@ -1148,6 +1166,7 @@ class PyMISP(object):
:param to_ids: return only the attributes with the to_ids flag set
:param deleted: also return the deleted attributes
:param event_timestamp: the timestamp of the last modification of the event (attributes controller only)). Can be a list (from->to)
:param includeProposals: return shadow attributes if True
:param async_callback: The function to run when results are returned
"""
query = {}
@ -1203,6 +1222,7 @@ class PyMISP(object):
query['metadata'] = kwargs.pop('metadata', None)
if controller == 'attributes':
query['event_timestamp'] = kwargs.pop('event_timestamp', None)
query['includeProposals'] = kwargs.pop('includeProposals', None)
# Cleanup
query = {k: v for k, v in query.items() if v is not None}
@ -1532,7 +1552,7 @@ class PyMISP(object):
"""Get the existing sharing groups"""
url = urljoin(self.root_url, 'sharing_groups.json')
response = self._prepare_request('GET', url)
return self._check_response(response)['response']
return self._check_response(response)
# ############## Users ##################
@ -2262,11 +2282,9 @@ class PyMISP(object):
def get_object_template_id(self, object_uuid):
"""Gets the template ID corresponting the UUID passed as parameter"""
templates = self.get_object_templates_list()
for t in templates:
if t['ObjectTemplate']['uuid'] == object_uuid:
return t['ObjectTemplate']['id']
raise Exception('Unable to find template uuid {} on the MISP instance'.format(object_uuid))
url = urljoin(self.root_url, 'objectTemplates/view/{}'.format(object_uuid))
response = self._prepare_request('GET', url)
return self._check_response(response)
def update_object_templates(self):
url = urljoin(self.root_url, '/objectTemplates/update')

83
pymisp/aping.py
View File

@ -2,7 +2,8 @@
# -*- coding: utf-8 -*-
from .exceptions import MISPServerError, NewEventError, UpdateEventError, UpdateAttributeError, PyMISPNotImplementedYet
from .api import PyMISP, everything_broken, MISPEvent, MISPAttribute, MISPSighting
from .api import PyMISP, everything_broken
from .mispevent import MISPEvent, MISPAttribute, MISPSighting, MISPLog
from typing import TypeVar, Optional, Tuple, List, Dict
from datetime import date, datetime
import json
@ -96,6 +97,8 @@ class ExpandedPyMISP(PyMISP):
created_event = super().add_event(event)
if isinstance(created_event, str):
raise NewEventError(f'Unexpected response from server: {created_event}')
elif 'errors' in created_event:
return created_event
e = MISPEvent()
e.load(created_event)
return e
@ -104,6 +107,8 @@ class ExpandedPyMISP(PyMISP):
updated_event = super().update_event(event.uuid, event)
if isinstance(updated_event, str):
raise UpdateEventError(f'Unexpected response from server: {updated_event}')
elif 'errors' in updated_event:
return updated_event
e = MISPEvent()
e.load(updated_event)
return e
@ -112,6 +117,8 @@ class ExpandedPyMISP(PyMISP):
updated_attribute = super().update_attribute(attribute.uuid, attribute)
if isinstance(updated_attribute, str):
raise UpdateAttributeError(f'Unexpected response from server: {updated_attribute}')
elif 'errors' in updated_attribute:
return updated_attribute
a = MISPAttribute()
a.from_dict(**updated_attribute)
return a
@ -263,7 +270,7 @@ class ExpandedPyMISP(PyMISP):
:param requested_attributes: [CSV only] Select the fields that you wish to include in the CSV export. By setting event level fields additionally, includeContext is not required to get event metadata.
:param include_context: [CSV Only] Include the event data with each attribute.
:param headerless: [CSV Only] The CSV created when this setting is set to true will not contain the header row.
:param pythonify: Returns a list of PyMISP Objects the the plain json output. Warning: it might use a lot of RAM
:param pythonify: Returns a list of PyMISP Objects instead of the plain json output. Warning: it might use a lot of RAM
Deprecated:
@ -346,7 +353,6 @@ class ExpandedPyMISP(PyMISP):
query['requested_attributes'] = requested_attributes
query['includeContext'] = include_context
query['headerless'] = headerless
url = urljoin(self.root_url, f'{controller}/restSearch')
# Remove None values.
# TODO: put that in self._prepare_request
@ -393,7 +399,7 @@ class ExpandedPyMISP(PyMISP):
action: Optional[str]=None, user_id: Optional[int]=None,
change: Optional[str]=None, email: Optional[str]=None,
org: Optional[str]=None, description: Optional[str]=None,
ip: Optional[str]=None):
ip: Optional[str]=None, pythonify: Optional[bool]=False):
'''Search in logs
Note: to run substring queries simply append/prepend/encapsulate the search term with %
@ -411,9 +417,11 @@ class ExpandedPyMISP(PyMISP):
:param org: Organisation of the User doing the action
:param description: Description of the action
:param ip: Origination IP of the User doing the action
:param pythonify: Returns a list of PyMISP Objects instead or the plain json output. Warning: it might use a lot of RAM
'''
query = locals()
query.pop('self')
query.pop('pythonify')
if log_id is not None:
query['id'] = query.pop('log_id')
@ -423,4 +431,69 @@ class ExpandedPyMISP(PyMISP):
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
normalized_response = self._check_response(response)
return normalized_response
if not pythonify:
return normalized_response
to_return = []
for l in normalized_response:
ml = MISPLog()
ml.from_dict(**l['Log'])
to_return.append(ml)
return to_return
def search_index(self, published: Optional[bool]=None, eventid: Optional[SearchType]=None,
tags: Optional[SearchParameterTypes]=None,
date_from: Optional[DateTypes]=None,
date_to: Optional[DateTypes]=None,
eventinfo: Optional[str]=None,
threatlevel: Optional[List[SearchType]]=None,
distribution: Optional[List[SearchType]]=None,
analysis: Optional[List[SearchType]]=None,
org: Optional[SearchParameterTypes]=None,
timestamp: Optional[DateInterval]=None,
pythonify: Optional[bool]=None):
"""Search only at the index level. Using ! in front of a value means NOT (default is OR)
:param published: Set whether published or unpublished events should be returned. Do not set the parameter if you want both.
:param eventid: The events that should be included / excluded from the search
:param tags: Tags to search or to exclude. You can pass a list, or the output of `build_complex_query`
:param date_from: Events with the date set to a date after the one specified. This filter will use the date of the event.
:param date_to: Events with the date set to a date before the one specified. This filter will use the date of the event.
:param eventinfo: Filter on the event's info field.
:param threatlevel: Threat level(s) (1,2,3,4) | list
:param distribution: Distribution level(s) (0,1,2,3) | list
:param analysis: Analysis level(s) (0,1,2) | list
:param org: Search by the creator organisation by supplying the organisation identifier.
:param timestamp: Restrict the results by the timestamp (last edit). Any event with a timestamp newer than the given timestamp will be returned. In case you are dealing with /attributes as scope, the attribute's timestamp will be used for the lookup.
:param pythonify: Returns a list of PyMISP Objects instead or the plain json output. Warning: it might use a lot of RAM
"""
query = locals()
query.pop('self')
query.pop('pythonify')
if query.get('date_from'):
query['datefrom'] = self.make_timestamp(query.pop('date_from'))
if query.get('date_to'):
query['dateuntil'] = self.make_timestamp(query.pop('date_to'))
if query.get('timestamp') is not None:
timestamp = query.pop('timestamp')
if isinstance(timestamp, (list, tuple)):
query['timestamp'] = (self.make_timestamp(timestamp[0]), self.make_timestamp(timestamp[1]))
else:
query['timestamp'] = self.make_timestamp(timestamp)
url = urljoin(self.root_url, 'events/index')
# Remove None values.
# TODO: put that in self._prepare_request
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
normalized_response = self._check_response(response)
if not pythonify:
return normalized_response
to_return = []
for e_meta in normalized_response:
me = MISPEvent()
me.from_dict(**e_meta)
to_return.append(me)
return to_return

288
pymisp/data/describeTypes.json
View File

@ -1,22 +1,22 @@
{
"result": {
"categories": [
"Internal reference",
"Targeting data",
"Antivirus detection",
"Payload delivery",
"Artifacts dropped",
"Payload installation",
"Persistence mechanism",
"Network activity",
"Payload type",
"Attribution",
"External analysis",
"Financial fraud",
"Internal reference",
"Network activity",
"Other",
"Payload delivery",
"Payload installation",
"Payload type",
"Persistence mechanism",
"Person",
"Social network",
"Support Tool",
"Targeting data"
"Social network",
"Person",
"Other"
],
"category_type_mappings": {
"Antivirus detection": [
@ -186,7 +186,9 @@
"attachment",
"comment",
"text",
"x509-fingerprint-md5",
"x509-fingerprint-sha1",
"x509-fingerprint-sha256",
"other",
"hex",
"cookie",
@ -1019,158 +1021,158 @@
}
},
"types": [
"AS",
"aba-rtn",
"attachment",
"authentihash",
"bank-account-nr",
"bic",
"bin",
"boolean",
"bro",
"btc",
"campaign-id",
"campaign-name",
"cc-number",
"comment",
"cookie",
"cortex",
"counter",
"country-of-residence",
"cpe",
"date-of-birth",
"datetime",
"dns-soa-email",
"md5",
"sha1",
"sha256",
"filename",
"pdb",
"filename|md5",
"filename|sha1",
"filename|sha256",
"ip-src",
"ip-dst",
"hostname",
"domain",
"domain|ip",
"email-src",
"email-dst",
"email-subject",
"email-attachment",
"email-body",
"email-dst",
"email-dst-display-name",
"email-header",
"email-message-id",
"email-mime-boundary",
"email-reply-to",
"email-src",
"email-src-display-name",
"email-subject",
"email-thread-index",
"email-x-mailer",
"filename",
"filename|authentihash",
"filename|impfuzzy",
"filename|imphash",
"filename|md5",
"filename|pehash",
"filename|sha1",
"filename|sha224",
"filename|sha256",
"filename|sha384",
"filename|sha512",
"filename|sha512/224",
"filename|sha512/256",
"filename|ssdeep",
"filename|tlsh",
"first-name",
"float",
"frequent-flyer-number",
"gender",
"gene",
"github-organisation",
"github-repository",
"github-username",
"hex",
"hostname",
"hostname|port",
"url",
"http-method",
"iban",
"identity-card-number",
"impfuzzy",
"imphash",
"ip-dst",
"ip-dst|port",
"ip-src",
"ip-src|port",
"issue-date-of-the-visa",
"jabber-id",
"last-name",
"link",
"mac-address",
"mac-eui-64",
"malware-sample",
"malware-type",
"md5",
"middle-name",
"mime-type",
"mobile-application-id",
"mutex",
"named pipe",
"nationality",
"other",
"passenger-name-record-locator-number",
"passport-country",
"passport-expiration",
"passport-number",
"pattern-in-file",
"pattern-in-memory",
"pattern-in-traffic",
"payment-details",
"pdb",
"pehash",
"phone-number",
"place-of-birth",
"place-port-of-clearance",
"place-port-of-onward-foreign-destination",
"place-port-of-original-embarkation",
"port",
"primary-residence",
"prtn",
"redress-number",
"user-agent",
"regkey",
"regkey|value",
"sha1",
"AS",
"snort",
"bro",
"pattern-in-file",
"pattern-in-traffic",
"pattern-in-memory",
"yara",
"stix2-pattern",
"sigma",
"gene",
"mime-type",
"identity-card-number",
"cookie",
"vulnerability",
"attachment",
"malware-sample",
"link",
"comment",
"text",
"hex",
"other",
"named pipe",
"mutex",
"target-user",
"target-email",
"target-machine",
"target-org",
"target-location",
"target-external",
"btc",
"xmr",
"iban",
"bic",
"bank-account-nr",
"aba-rtn",
"bin",
"cc-number",
"prtn",
"phone-number",
"threat-actor",
"campaign-name",
"campaign-id",
"malware-type",
"uri",
"authentihash",
"ssdeep",
"imphash",
"pehash",
"impfuzzy",
"sha224",
"sha256",
"sha384",
"sha512",
"sha512/224",
"sha512/256",
"sigma",
"size-in-bytes",
"snort",
"special-service-request",
"ssdeep",
"stix2-pattern",
"target-email",
"target-external",
"target-location",
"target-machine",
"target-org",
"target-user",
"text",
"threat-actor",
"tlsh",
"travel-details",
"twitter-id",
"uri",
"url",
"user-agent",
"visa-number",
"vulnerability",
"whois-creation-date",
"filename|authentihash",
"filename|ssdeep",
"filename|imphash",
"filename|impfuzzy",
"filename|pehash",
"filename|sha224",
"filename|sha384",
"filename|sha512",
"filename|sha512/224",
"filename|sha512/256",
"filename|tlsh",
"windows-scheduled-task",
"windows-service-name",
"windows-service-displayname",
"whois-registrant-email",
"whois-registrant-phone",
"whois-registrant-name",
"whois-registrant-org",
"whois-registrant-phone",
"whois-registrar",
"windows-scheduled-task",
"windows-service-displayname",
"windows-service-name",
"x509-fingerprint-md5",
"whois-creation-date",
"x509-fingerprint-sha1",
"x509-fingerprint-md5",
"x509-fingerprint-sha256",
"xmr",
"yara"
"dns-soa-email",
"size-in-bytes",
"counter",
"datetime",
"cpe",
"port",
"ip-dst|port",
"ip-src|port",
"hostname|port",
"mac-address",
"mac-eui-64",
"email-dst-display-name",
"email-src-display-name",
"email-header",
"email-reply-to",
"email-x-mailer",
"email-mime-boundary",
"email-thread-index",
"email-message-id",
"github-username",
"github-repository",
"github-organisation",
"jabber-id",
"twitter-id",
"first-name",
"middle-name",
"last-name",
"date-of-birth",
"place-of-birth",
"gender",
"passport-number",
"passport-country",
"passport-expiration",
"redress-number",
"nationality",
"visa-number",
"issue-date-of-the-visa",
"primary-residence",
"country-of-residence",
"special-service-request",
"frequent-flyer-number",
"travel-details",
"payment-details",
"place-port-of-original-embarkation",
"place-port-of-clearance",
"place-port-of-onward-foreign-destination",
"passenger-name-record-locator-number",
"mobile-application-id",
"cortex",
"boolean"
]
}
}
}

2
pymisp/data/misp-objects

@ -1 +1 @@
Subproject commit 6e03108fb104ae90617701aa5d0749cb932c821f
Subproject commit 11a462e79b02428a08b11698d45aa8aa5ab6887d

18
pymisp/mispevent.py
View File

@ -469,8 +469,7 @@ class MISPEvent(AbstractMISP):
'attribute_count' in event.get('Event') and
event.get('Event').get('attribute_count') is None):
event['Event']['attribute_count'] = '0'
e = event.get('Event')
self.from_dict(**e)
self.from_dict(**event['Event'])
if validate:
jsonschema.validate(json.loads(self.to_json()), self.__json_schema)
@ -820,6 +819,15 @@ class MISPFeed(AbstractMISP):
super(MISPFeed, self).__init__()
class MISPLog(AbstractMISP):
def __init__(self):
super(MISPLog, self).__init__()
def __repr__(self):
return '<{self.__class__.__name__}({self.model}, {self.action}, {self.title})'.format(self=self)
class MISPSighting(AbstractMISP):
def __init__(self):
@ -1008,6 +1016,12 @@ class MISPObject(AbstractMISP):
else:
self._known_template = False
if 'distribution' in kwargs and kwargs['distribution'] is not None:
self.distribution = kwargs.pop('distribution')
self.distribution = int(self.distribution)
if self.distribution not in [0, 1, 2, 3, 4, 5]:
raise NewAttributeError('{} is invalid, the distribution has to be in 0, 1, 2, 3, 4, 5'.format(self.distribution))
if kwargs.get('timestamp'):
if sys.version_info >= (3, 3):
self.timestamp = datetime.datetime.fromtimestamp(int(kwargs.pop('timestamp')), datetime.timezone.utc)

6
tests/sharing_groups.json
View File

@ -1,5 +1,4 @@
{
"response": [
[
{
"SharingGroup": {
"id": "1",
@ -96,5 +95,4 @@
],
"editable": true
}
]
}
]

3
tests/test_offline.py
View File

@ -112,7 +112,8 @@ class TestOffline(unittest.TestCase):
self.initURI(m)
pymisp = PyMISP(self.domain, self.key)
sharing_groups = pymisp.get_sharing_groups()
self.assertEqual(sharing_groups[0], self.sharing_groups['response'][0])
print(sharing_groups)
self.assertEqual(sharing_groups['response'][0], self.sharing_groups[0])
def test_auth_error(self, m):
self.initURI(m)

60
tests/testlive_comprehensive.py
View File

@ -3,7 +3,7 @@
import unittest
from pymisp import ExpandedPyMISP, MISPEvent, MISPOrganisation, MISPUser, Distribution, ThreatLevel, Analysis
from pymisp import ExpandedPyMISP, MISPEvent, MISPOrganisation, MISPUser, Distribution, ThreatLevel, Analysis, MISPObject
from datetime import datetime, timedelta, date
from io import BytesIO
@ -11,15 +11,15 @@ import time
try:
from keys import url, key
travis_run = True
except ImportError as e:
print(e)
url = 'http://localhost:8080'
key = 'y0rs3LNOP0Y3v6dfSMMdhxj5Oxx9MfaInpRP2pBC'
key = 'LBelWqKY9SQyG0huZzAMqiEBl6FODxpgRRXMsZFu'
travis_run = False
from uuid import uuid4
travis_run = True
class TestComprehensive(unittest.TestCase):
@ -439,6 +439,51 @@ class TestComprehensive(unittest.TestCase):
self.admin_misp_connector.delete_event(first.id)
self.admin_misp_connector.delete_event(second.id)
def test_default_distribution(self):
'''The default distributions on the VM are This community only for the events and Inherit from event for attr/obj)'''
if travis_run:
return
first = self.create_simple_event()
del first.distribution
o = first.add_object(name='file')
o.add_attribute('filename', value='foo.exe')
try:
# Event create
first = self.user_misp_connector.add_event(first)
self.assertEqual(first.distribution, Distribution.this_community_only.value)
self.assertEqual(first.attributes[0].distribution, Distribution.inherit.value)
self.assertEqual(first.objects[0].distribution, Distribution.inherit.value)
self.assertEqual(first.objects[0].attributes[0].distribution, Distribution.inherit.value)
# Event edit
first.add_attribute('ip-dst', '12.54.76.43')
o = first.add_object(name='file')
o.add_attribute('filename', value='foo2.exe')
first = self.user_misp_connector.update_event(first)
self.assertEqual(first.attributes[1].distribution, Distribution.inherit.value)
self.assertEqual(first.objects[1].distribution, Distribution.inherit.value)
self.assertEqual(first.objects[1].attributes[0].distribution, Distribution.inherit.value)
# Attribute create
attribute = self.user_misp_connector.add_named_attribute(first, 'comment', 'bar')
# FIXME: Add helper that returns a list of MISPAttribute
self.assertEqual(attribute[0]['Attribute']['distribution'], str(Distribution.inherit.value))
# Object - add
o = MISPObject('file')
o.add_attribute('filename', value='blah.exe')
new_obj = self.user_misp_connector.add_object(first.id, o.template_uuid, o)
# FIXME: Add helper that returns a MISPObject
self.assertEqual(new_obj['Object']['distribution'], str(Distribution.inherit.value))
self.assertEqual(new_obj['Object']['Attribute'][0]['distribution'], str(Distribution.inherit.value))
# Object - edit
clean_obj = MISPObject(**new_obj['Object'])
clean_obj.from_dict(**new_obj['Object'])
clean_obj.add_attribute('filename', value='blah.exe')
new_obj = self.user_misp_connector.edit_object(clean_obj)
for a in new_obj['Object']['Attribute']:
self.assertEqual(a['distribution'], str(Distribution.inherit.value))
finally:
# Delete event
self.admin_misp_connector.delete_event(first.id)
def test_simple_event(self):
'''Search a bunch of parameters:
* Value not existing
@ -517,7 +562,7 @@ class TestComprehensive(unittest.TestCase):
# quickfilter
events = self.user_misp_connector.search(timestamp=timeframe,
quickfilter='%bar%', pythonify=True)
quickfilter='%foo blah%', pythonify=True)
# FIXME: should return one event
# print(events)
# self.assertEqual(len(events), 1)
@ -920,8 +965,9 @@ class TestComprehensive(unittest.TestCase):
first = self.create_simple_event()
try:
first = self.user_misp_connector.add_event(first)
r = self.admin_misp_connector.pushEventToZMQ(first.id)
self.assertEqual(r['message'], 'Event published to ZMQ')
if not travis_run:
r = self.admin_misp_connector.pushEventToZMQ(first.id)
self.assertEqual(r['message'], 'Event published to ZMQ')
finally:
# Delete event
self.admin_misp_connector.delete_event(first.id)