Merge pull request #95 from ninoseki/add-whitelist

feat: add MISP-warninglist compatible whitelisting feature

Author: cmmorrow

tests/fixtures/whitelist.json

@@ -0,0 +1,8 @@
+{
+  "name": "test",
+  "version": 20200203,
+  "description": "test",
+  "matching_attributes": ["hostname", "domain"],
+  "type": "hostname",
+  "list": ["00-tv.com", "000webhost.com", "000webhostapp.com"]
+}

tests/test_whitelist.py

@@ -0,0 +1,15 @@
+import unittest
+from pathlib import Path
+
+from threatingestor.whitelist import Whitelist
+
+
+class TestState(unittest.TestCase):
+    def setUp(self):
+        parent = Path(__file__).parent.absolute()
+        path = parent / "fixtures/whitelist.json"
+        self.whitelist = Whitelist(paths=[str(path)])
+
+    def test_contains(self):
+        self.assertTrue(self.whitelist.contains("00-tv.com"))
+        self.assertFalse(self.whitelist.contains("example.com"))

threatingestor/__init__.py

@@ -2,7 +2,6 @@
 import time
 import collections
 
-
 from loguru import logger
 import statsd
 try:
@@ -16,6 +15,7 @@
 import threatingestor.config
 import threatingestor.state
 import threatingestor.exceptions
+import threatingestor.whitelist
 
 
 class Ingestor:
@@ -81,11 +81,21 @@ def __init__(self, config_file):
             self.operators = {name: operator(**kwargs)
                               for name, operator, kwargs in self.config.operators()}
 
+            logger.debug("Initializing whitelists")
+            self.whitelist = threatingestor.whitelist.Whitelist(self.config.whitelists())
+
         except (TypeError, ConnectionError, threatingestor.exceptions.PluginError):
             logger.warning("Twitter config format has recently changed. See https://github.com/InQuest/ThreatIngestor/releases/tag/v1.0.0b5")
             logger.exception("Error initializing plugins")
             sys.exit(1)
 
+    def _is_whitelisted(self, artifact) -> bool:
+        if self.whitelist.contains(str(artifact)):
+            logger.debug(
+                f"Reject {str(artifact)} from further processing because it is whitelisted."
+            )
+            return True
+        return False
 
     def run(self):
         """Run once, or forever, depending on config."""
@@ -118,6 +128,13 @@ def run_once(self):
             # Save the source state.
             self.statedb.save_state(source, saved_state)
 
+            # Reject whitelisted artifacts
+            artifacts = [
+                artifact
+                for artifact in artifacts
+                if not self._is_whitelisted(artifact)
+            ]
+
             # Process artifacts with each operator.
             for operator in self.operators:
                 logger.debug(f"Processing {len(artifacts)} artifacts from source '{source}' with operator '{operator}'")

threatingestor/config.py

@@ -162,3 +162,7 @@ def operators(self):
 
         logger.debug(f"Found {len(operators)} total operators")
         return operators
+
+    def whitelists(self):
+        """Returns whitelist list."""
+        return self.config.get("whitelists", [])

threatingestor/whitelist.py

@@ -0,0 +1,31 @@
+import json
+from pathlib import Path
+from typing import List, Set
+
+from loguru import logger
+
+class Whitelist:
+    """Base class for Whitelist plugin."""
+
+    def __init__(self, paths: List[str]):
+        self.paths = paths
+        self.values: Set[str] = set()
+        self._load_paths()
+
+    def contains(self, value: str) -> bool:
+        return value in self.values
+
+    def _load_path(self, path: str):
+        if Path(path).is_file():
+            try:
+                with open(path) as f:
+                    data = json.load(f)
+                    list_ = data.get("list", [])
+                    self.values.update(list_)
+            except json.decoder.JSONDecodeError:
+                logger.exception(f"Couldn't load {path} as a JSON file")
+
+
+    def _load_paths(self):
+        for path in self.paths:
+            self._load_path(path)