Skip to content

Commit

Permalink
chg: Add more test cases
Browse files Browse the repository at this point in the history
  • Loading branch information
@Rafiot
Rafiot committed Aug 19, 2018
1 parent 2c03fb9 commit 303079a
Show file tree
Hide file tree
Showing 2 changed files with 107 additions and 27 deletions.
31 changes: 27 additions & 4 deletions pymisp/aping.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-

from .exceptions import MISPServerError
from .api import PyMISP, everything_broken
from .api import PyMISP, everything_broken, MISPEvent, MISPAttribute
from typing import TypeVar, Optional, Tuple, List, Dict
from datetime import date, datetime
import json
Expand Down Expand Up @@ -64,16 +64,18 @@ def _check_response(self, response):
return {'errors': [(response.status_code, error_message)]}

# At this point, we had no error.
if logger.isEnabledFor(logging.DEBUG):
logger.debug(response)

try:
response = response.json()
if logger.isEnabledFor(logging.DEBUG):
logger.debug(response)
if response.get('response') is not None:
# Cleanup.
return response.get('response')
return response
except Exception:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(response.text)
return response.text

# TODO: Make that thing async & test it.
Expand Down Expand Up @@ -151,4 +153,25 @@ def search(self, controller: str='events', return_format: str='json',

url = urljoin(self.root_url, f'{controller}/restSearch')
response = self._prepare_request('POST', url, data=json.dumps(query))
return self._check_response(response)
normalized_response = self._check_response(response)
if isinstance(normalized_response, str) or (isinstance(normalized_response, dict) and
normalized_response.get('errors')):
return normalized_response
# The response is in json, we can confert it to a list of pythonic MISP objects
to_return = []
if controller == 'events':
for e in normalized_response:
me = MISPEvent()
me.load(e)
to_return.append(me)
elif controller == 'attributes':
print(normalized_response)
# FIXME: if the query doesn't match, the request returns an empty list, and not a dictionary;
if normalized_response:
for a in normalized_response.get('Attribute'):
ma = MISPAttribute()
ma.from_dict(**a)
to_return.append(ma)
elif controller == 'objects':
raise Exception('Not implemented yet')
return to_return
103 changes: 80 additions & 23 deletions tests/testlive_comprehensive.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -83,16 +83,85 @@ def test_search_value_event(self):
# Delete event
self.admin_misp_connector.delete_event(c_me.id)

def test_search_event_type(self):
me = self.create_event_org_only()
me.add_attribute('ip-src', '8.8.8.8')
second = self.create_event_org_only()
second.add_attribute('ip-dst', '9.9.9.9')
third = self.create_event_org_only()
try:
# Create event
created_event = self.admin_misp_connector.add_event(me)
c_me = MISPEvent()
c_me.load(created_event)
created_event = self.admin_misp_connector.add_event(second)
second_me = MISPEvent()
second_me.load(created_event)
created_event = self.admin_misp_connector.add_event(third)
third_me = MISPEvent()
third_me.load(created_event)
# Search as admin
response = self.admin_misp_connector.search(timestamp=c_me.timestamp.timestamp())
self.assertEqual(len(response), 3)
attrubutes_types_search = self.admin_misp_connector.build_complex_query(or_parameters=['ip-src', 'ip-dst'])
response = self.admin_misp_connector.search(controller='events', timestamp=c_me.timestamp.timestamp(),
type_attribute=attrubutes_types_search)
# print(response)
self.assertEqual(len(response), 2)
finally:
# Delete event
self.admin_misp_connector.delete_event(c_me.id)
self.admin_misp_connector.delete_event(second_me.id)
self.admin_misp_connector.delete_event(third_me.id)

def test_search_attribute_type(self):
me = self.create_event_org_only()
me.add_attribute('ip-src', '8.8.8.8')
second = self.create_event_org_only()
second.add_attribute('ip-dst', '9.9.9.9')
third = self.create_event_org_only()
try:
# Create event
created_event = self.admin_misp_connector.add_event(me)
c_me = MISPEvent()
c_me.load(created_event)
created_event = self.admin_misp_connector.add_event(second)
second_me = MISPEvent()
second_me.load(created_event)
created_event = self.admin_misp_connector.add_event(third)
third_me = MISPEvent()
third_me.load(created_event)
# Search as admin
response = self.admin_misp_connector.search(controller='attributes', timestamp=c_me.timestamp.timestamp())
self.assertEqual(len(response), 5)
attrubutes_types_search = self.admin_misp_connector.build_complex_query(or_parameters=['ip-src', 'ip-dst'])
response = self.admin_misp_connector.search(controller='attributes', timestamp=c_me.timestamp.timestamp(),
type_attribute=attrubutes_types_search)
# print(response)
self.assertEqual(len(response), 2)
finally:
# Delete event
self.admin_misp_connector.delete_event(c_me.id)
self.admin_misp_connector.delete_event(second_me.id)
self.admin_misp_connector.delete_event(third_me.id)

def test_search_value_attribute(self):
me = self.create_event_org_only()
me.add_attribute('text', str(uuid4()))
second = self.create_event_org_only()
second.add_attribute('text', me.attributes[0].value)
try:
# Create event
created_event = self.admin_misp_connector.add_event(me)
c_me = MISPEvent()
c_me.load(created_event)
created_event = self.admin_misp_connector.add_event(second)
second_me = MISPEvent()
second_me.load(created_event)
# Search as admin
response = self.admin_misp_connector.search(controller='attributes', value=me.attributes[0].value)
self.assertEqual(len(response), 1)

# Connect as user
user_misp_connector = ExpandedPyMISP(url, self.test_usr.authkey)
# Search as user
Expand All @@ -101,6 +170,7 @@ def test_search_value_attribute(self):
finally:
# Delete event
self.admin_misp_connector.delete_event(c_me.id)
self.admin_misp_connector.delete_event(second_me.id)

def test_search_tag_event(self):
me = self.create_event_with_tags()
Expand Down Expand Up @@ -132,12 +202,10 @@ def test_search_tag_event_fancy(self):
to_delete.load(created_event)
complex_query = user_misp_connector.build_complex_query(or_parameters=['tlp:white___test'], not_parameters=['tlp:amber___test'])
# Search as user
response = user_misp_connector.search(tags=complex_query)
for e in response:
to_validate = MISPEvent()
to_validate.load(e)
events = user_misp_connector.search(tags=complex_query)
for e in events:
# FIXME Expected event without the tlp:amber attribute, broken for now
for a in to_validate.attributes:
for a in e.attributes:
print([t for t in a.tags if t.name == 'tlp:amber___test'])
# self.assertEqual([t for t in a.tags if t.name == 'tlp:amber___test'], [])
# Delete event
Expand Down Expand Up @@ -165,22 +233,19 @@ def test_search_timestamp(self):
# # Test - last 4 min
response = user_misp_connector.search(timestamp='4m')
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])
received_event = response[0]
self.assertEqual(received_event.timestamp.timestamp(), int(event_creation_timestamp_second.timestamp()))

# # Test 5 sec before timestamp of 2nd event
response = user_misp_connector.search(timestamp=(event_creation_timestamp_second.timestamp()))
# # Test timestamp of 2nd event
response = user_misp_connector.search(timestamp=event_creation_timestamp_second.timestamp())
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])
received_event = response[0]
self.assertEqual(received_event.timestamp.timestamp(), int(event_creation_timestamp_second.timestamp()))

# # Test interval -6 min -> -4 min
response = user_misp_connector.search(timestamp=['6m', '4m'])
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])
received_event = response[0]
self.assertEqual(received_event.timestamp.timestamp(), int(event_creation_timestamp_first.timestamp()))
finally:
# Delete event
Expand Down Expand Up @@ -236,20 +301,14 @@ def test_search_publish_timestamp(self):
# # Test - last 4 min
response = pub_misp_connector.search(publish_timestamp='5s')
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])

# # Test 5 sec before timestamp of 2nd event
response = pub_misp_connector.search(publish_timestamp=(second_to_delete.publish_timestamp.timestamp()))
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])

# # Test interval -6 min -> -4 min
response = pub_misp_connector.search(publish_timestamp=[first_to_delete.publish_timestamp.timestamp() - 5, second_to_delete.publish_timestamp.timestamp() - 5])
self.assertEqual(len(response), 1)
received_event = MISPEvent()
received_event.load(response[0])
finally:
# Delete event
self.admin_misp_connector.delete_event(first_to_delete.id)
Expand All @@ -276,13 +335,11 @@ def test_simple(self):
# Test return content
response = user_misp_connector.search(timestamp=timeframe, metadata=False)
self.assertEqual(len(response), 1)
t = MISPEvent()
t.load(response[0])
t = response[0]
self.assertEqual(len(t.attributes), 1)
response = user_misp_connector.search(timestamp=timeframe, metadata=True)
self.assertEqual(len(response), 1)
t = MISPEvent()
t.load(response[0])
t = response[0]
self.assertEqual(len(t.attributes), 0)
# other things
response = user_misp_connector.search(timestamp=timeframe, published=True)
Expand Down

0 comments on commit 303079a

Please sign in to comment.