mirror of https://github.com/MISP/MISP
chg: [test] Add snort attribute to test
parent
ba1387edb5
commit
58928a4497
|
@ -10,9 +10,10 @@ if (!isset($argv[2])) {
|
||||||
if (!in_array($argv[1], ['modify', 'replace'], true)) {
|
if (!in_array($argv[1], ['modify', 'replace'], true)) {
|
||||||
fail(1, "Invalid argument '{$argv[1]}', it must be 'modify' or 'replace'.");
|
fail(1, "Invalid argument '{$argv[1]}', it must be 'modify' or 'replace'.");
|
||||||
}
|
}
|
||||||
$newConfig = json_decode($argv[2], true);
|
try {
|
||||||
if ($newConfig === null) {
|
$newConfig = json_decode($argv[2], true, JSON_THROW_ON_ERROR);
|
||||||
fail(2, "Could not decode new config, it is not JSON: " . json_last_error_msg());
|
} catch (Exception $e) {
|
||||||
|
fail(2, "Could not decode new config, it is not JSON: " . $e->getMessage());
|
||||||
}
|
}
|
||||||
if (!is_array($newConfig)) {
|
if (!is_array($newConfig)) {
|
||||||
fail(2, "Provided new config is not array, `" . gettype($newConfig) . "` given.");
|
fail(2, "Provided new config is not array, `" . gettype($newConfig) . "` given.");
|
||||||
|
@ -41,4 +42,4 @@ if ($argv[1] === 'modify') {
|
||||||
file_put_contents($configFile, "<?php\n\$config = " . var_export($merged, true) . ';', LOCK_EX);
|
file_put_contents($configFile, "<?php\n\$config = " . var_export($merged, true) . ';', LOCK_EX);
|
||||||
|
|
||||||
// Returns config file before modification
|
// Returns config file before modification
|
||||||
echo json_encode($config);
|
echo json_encode($config, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
||||||
|
|
|
@ -2,6 +2,7 @@
|
||||||
import os
|
import os
|
||||||
import json
|
import json
|
||||||
import uuid
|
import uuid
|
||||||
|
import logging
|
||||||
import inspect
|
import inspect
|
||||||
import subprocess
|
import subprocess
|
||||||
import unittest
|
import unittest
|
||||||
|
@ -11,21 +12,19 @@ from xml.etree import ElementTree as ET
|
||||||
from io import BytesIO
|
from io import BytesIO
|
||||||
import urllib3 # type: ignore
|
import urllib3 # type: ignore
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
from typing import Union
|
||||||
import logging
|
|
||||||
logging.disable(logging.CRITICAL)
|
|
||||||
logger = logging.getLogger('pymisp')
|
|
||||||
|
|
||||||
|
|
||||||
from pymisp import PyMISP, MISPOrganisation, MISPUser, MISPRole, MISPSharingGroup, MISPEvent, MISPLog, MISPSighting, Distribution, ThreatLevel, Analysis, MISPEventReport, MISPServerError
|
from pymisp import PyMISP, MISPOrganisation, MISPUser, MISPRole, MISPSharingGroup, MISPEvent, MISPLog, MISPSighting, Distribution, ThreatLevel, Analysis, MISPEventReport, MISPServerError
|
||||||
from pymisp.tools import DomainIPObject
|
from pymisp.tools import DomainIPObject
|
||||||
|
from pymisp.api import get_uuid_or_id_from_abstract_misp
|
||||||
|
|
||||||
|
logging.disable(logging.CRITICAL)
|
||||||
|
logger = logging.getLogger('pymisp')
|
||||||
|
urllib3.disable_warnings()
|
||||||
|
|
||||||
# Load access information for env variables
|
# Load access information for env variables
|
||||||
url = "http://" + os.environ["HOST"]
|
url = "http://" + os.environ["HOST"]
|
||||||
key = os.environ["AUTH"]
|
key = os.environ["AUTH"]
|
||||||
|
|
||||||
urllib3.disable_warnings()
|
|
||||||
|
|
||||||
|
|
||||||
def create_simple_event() -> MISPEvent:
|
def create_simple_event() -> MISPEvent:
|
||||||
caller_name = inspect.stack()[1].function
|
caller_name = inspect.stack()[1].function
|
||||||
|
@ -52,13 +51,19 @@ def request(pymisp: PyMISP, request_type: str, url: str, data: dict = {}) -> dic
|
||||||
return pymisp._check_json_response(response)
|
return pymisp._check_json_response(response)
|
||||||
|
|
||||||
|
|
||||||
|
def publish_immediately(pymisp: PyMISP, event: Union[MISPEvent, int, str, uuid.UUID], with_email: bool = False):
|
||||||
|
event_id = get_uuid_or_id_from_abstract_misp(event)
|
||||||
|
action = "alert" if with_email else "publish"
|
||||||
|
return check_response(request(pymisp, 'POST', f'events/{action}/{event_id}/disable_background_processing:1'))
|
||||||
|
|
||||||
|
|
||||||
class MISPSetting:
|
class MISPSetting:
|
||||||
def __init__(self, admin_connector: PyMISP, new_setting: dict):
|
def __init__(self, admin_connector: PyMISP, new_setting: dict):
|
||||||
self.admin_connector = admin_connector
|
self.admin_connector = admin_connector
|
||||||
self.new_setting = new_setting
|
self.new_setting = new_setting
|
||||||
|
|
||||||
def __enter__(self):
|
def __enter__(self):
|
||||||
self.original = self.__run("modify", json.dumps(self.new_setting))
|
self.original = self.__run("modify", json.dumps(self.new_setting).encode("utf-8"))
|
||||||
# Try to reset config cache
|
# Try to reset config cache
|
||||||
self.admin_connector.get_server_setting("MISP.live")
|
self.admin_connector.get_server_setting("MISP.live")
|
||||||
|
|
||||||
|
@ -68,12 +73,12 @@ class MISPSetting:
|
||||||
self.admin_connector.get_server_setting("MISP.live")
|
self.admin_connector.get_server_setting("MISP.live")
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def __run(command: str, data: str) -> str:
|
def __run(command: str, data: bytes) -> bytes:
|
||||||
dir_path = os.path.dirname(os.path.realpath(__file__))
|
dir_path = os.path.dirname(os.path.realpath(__file__))
|
||||||
r = subprocess.run(["php", dir_path + "/modify_config.php", command, data], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
r = subprocess.run(["php", dir_path + "/modify_config.php", command, data], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
if r.returncode != 0:
|
if r.returncode != 0:
|
||||||
raise Exception([r.returncode, r.stdout, r.stderr])
|
raise Exception([r.returncode, r.stdout, r.stderr])
|
||||||
return r.stdout.decode("utf-8")
|
return r.stdout
|
||||||
|
|
||||||
|
|
||||||
class TestComprehensive(unittest.TestCase):
|
class TestComprehensive(unittest.TestCase):
|
||||||
|
@ -465,8 +470,6 @@ class TestComprehensive(unittest.TestCase):
|
||||||
check_response(self.admin_misp_connector.delete_event(event))
|
check_response(self.admin_misp_connector.delete_event(event))
|
||||||
|
|
||||||
def test_publish_alert_filter(self):
|
def test_publish_alert_filter(self):
|
||||||
check_response(self.admin_misp_connector.set_server_setting('MISP.background_jobs', 0, force=True))
|
|
||||||
|
|
||||||
first = create_simple_event()
|
first = create_simple_event()
|
||||||
first.add_tag('test_publish_filter')
|
first.add_tag('test_publish_filter')
|
||||||
first.threat_level_id = ThreatLevel.medium
|
first.threat_level_id = ThreatLevel.medium
|
||||||
|
@ -499,7 +502,7 @@ class TestComprehensive(unittest.TestCase):
|
||||||
|
|
||||||
# Publish events
|
# Publish events
|
||||||
for event in (first, second, third, four):
|
for event in (first, second, third, four):
|
||||||
check_response(self.admin_misp_connector.publish(event, alert=True))
|
publish_immediately(self.admin_misp_connector, event, with_email=True)
|
||||||
|
|
||||||
# Email notification should be send just to first event
|
# Email notification should be send just to first event
|
||||||
mail_logs = self.admin_misp_connector.search_logs(model='User', action='email')
|
mail_logs = self.admin_misp_connector.search_logs(model='User', action='email')
|
||||||
|
@ -516,8 +519,6 @@ class TestComprehensive(unittest.TestCase):
|
||||||
check_response(self.admin_misp_connector.update_user(self.admin_misp_connector._current_user))
|
check_response(self.admin_misp_connector.update_user(self.admin_misp_connector._current_user))
|
||||||
# Delete filter
|
# Delete filter
|
||||||
self.admin_misp_connector.delete_user_setting('publish_alert_filter')
|
self.admin_misp_connector.delete_user_setting('publish_alert_filter')
|
||||||
# Reenable background jobs
|
|
||||||
check_response(self.admin_misp_connector.set_server_setting('MISP.background_jobs', 1, force=True))
|
|
||||||
# Delete events
|
# Delete events
|
||||||
for event in (first, second, third, four):
|
for event in (first, second, third, four):
|
||||||
check_response(self.admin_misp_connector.delete_event(event))
|
check_response(self.admin_misp_connector.delete_event(event))
|
||||||
|
@ -923,11 +924,12 @@ class TestComprehensive(unittest.TestCase):
|
||||||
def test_search_snort_suricata(self):
|
def test_search_snort_suricata(self):
|
||||||
event = create_simple_event()
|
event = create_simple_event()
|
||||||
event.add_attribute('ip-src', '8.8.8.8', to_ids=True)
|
event.add_attribute('ip-src', '8.8.8.8', to_ids=True)
|
||||||
|
event.add_attribute('snort', 'alert tcp 192.168.1.0/24 any -> 131.171.127.1 25 (content: "hacking"; msg: "malicious packet"; sid:2000001;)', to_ids=True)
|
||||||
event = self.user_misp_connector.add_event(event)
|
event = self.user_misp_connector.add_event(event)
|
||||||
check_response(event)
|
check_response(event)
|
||||||
|
|
||||||
self.admin_misp_connector.publish(event, alert=False)
|
publish_immediately(self.admin_misp_connector, event)
|
||||||
time.sleep(6)
|
|
||||||
snort = self._search_event({'returnFormat': 'snort', 'eventid': event.id})
|
snort = self._search_event({'returnFormat': 'snort', 'eventid': event.id})
|
||||||
self.assertIsInstance(snort, str)
|
self.assertIsInstance(snort, str)
|
||||||
self.assertIn('8.8.8.8', snort)
|
self.assertIn('8.8.8.8', snort)
|
||||||
|
|
Loading…
Reference in New Issue