Fix formatting/whitespace

salt/sensoroni/files/analyzers/urlscan/urlscan/requirements.txt

@@ -1,2 +0,0 @@
-requests>=2.27.1
-pyyaml>=6.0

salt/sensoroni/files/analyzers/urlscan/urlscan/urlscan.json

@@ -1,7 +0,0 @@
-{
-  "name": "Urlscan",
-  "version": "0.1",
-  "author": "Security Onion Solutions",
-  "description": "This analyzer submits a URL to Urlscan for context around an observable.",
-  "supportedTypes" :  ["url"]
-}

salt/sensoroni/files/analyzers/urlscan/urlscan/urlscan.py

@@ -1,91 +0,0 @@
-import json
-import requests
-import helpers
-import sys
-import os
-import argparse
-import time
-
-
-def checkConfigRequirements(conf):
-    if "enabled" in conf:
-        if "api_key" not in conf or len(conf['api_key']) == 0:
-            sys.exit(126)
-        else:
-            return True
-    else:
-        sys.exit(126)
-
-
-def buildReq(conf, artifact_type, artifact_value):
-    headers = {"API-Key": conf["api_key"]}
-    url = conf['base_url'] + 'scan/'
-    visibility = conf['visibility']
-    data = {"url": artifact_value, "visibility": visibility}
-    return url, headers, data
-
-
-def getReport(conf, report_url):
-    report = requests.request('GET', report_url)
-    timeout = conf.get('timeout', 300)
-    counter = 0
-    while report.status_code == 404:
-        time.sleep(2)
-        counter += 2
-        if counter >= timeout:
-            break
-        report = requests.request('GET', report_url)
-    return report
-
-
-def sendReq(url, headers, data):
-    submission = requests.request('POST', url=url, headers=headers, data=data).json()
-    report_url = submission['api']
-    return report_url
-
-
-def prepareResults(raw):
-    if raw and "verdicts" in raw:
-        if raw["verdicts"]["overall"]["malicious"] is True:
-            status = "threat"
-            summary = "malicious"
-        elif raw["verdicts"]["overall"]["score"] > 0:
-            status = "caution"
-            summary = "suspicious"
-        else:
-            status = "info"
-            summary = "Scan complete."
-    else:
-        status = "caution"
-        summary = "internal_failure"
-
-    results = {'response': raw, 'status': status, 'summary': summary}
-    return results
-
-
-def analyze(conf, input):
-    checkConfigRequirements(conf)
-    meta = helpers.loadMetadata(__file__)
-    data = helpers.parseArtifact(input)
-    helpers.checkSupportedType(meta, data["artifactType"])
-    request = buildReq(conf, data["artifactType"], data["value"])
-    report_url = sendReq(request[0], request[1], request[2])
-    time.sleep(10)
-    report = getReport(conf, report_url)
-    return prepareResults(report.json())
-
-
-def main():
-    dir = os.path.dirname(os.path.realpath(__file__))
-    parser = argparse.ArgumentParser(description='Search Alienvault OTX for a given artifact')
-    parser.add_argument('artifact', help='the artifact represented in JSON format')
-    parser.add_argument('-c', '--config', metavar="CONFIG_FILE", default=dir + "/urlscan.yaml", help='optional config file to use instead of the default config file')
-
-    args = parser.parse_args()
-    if args.artifact:
-        results = analyze(helpers.loadConfig(args.config), args.artifact)
-        print(json.dumps(results))
-
-
-if __name__ == "__main__":
-    main()

salt/sensoroni/files/analyzers/urlscan/urlscan/urlscan.yaml

@@ -1,5 +0,0 @@
-base_url: https://urlscan.io/api/v1/
-api_key: "{{ salt['pillar.get']('sensoroni:analyzers:urlscan:api_key', '') }}"
-enabled: "{{ salt['pillar.get']('sensoroni:analyzers:urlscan:enabled', 'False') }}"
-visibility: "{{ salt['pillar.get']('sensoroni:analyzers:urlscan:visibility', 'public') }}"
-timeout: "{{ salt['pillar.get']('sensoroni:analyzers:urlscan:visibility', '180') }}"

salt/sensoroni/files/analyzers/urlscan/urlscan/urlscan_test.py

@@ -1,121 +0,0 @@
-from io import StringIO
-import sys
-from unittest.mock import patch, MagicMock, PropertyMock, call
-from urlscan import urlscan
-import unittest
-
-
-class TestUrlScanMethods(unittest.TestCase):
-
-    def test_main_missing_input(self):
-        with patch('sys.exit', new=MagicMock()) as sysmock:
-            with patch('sys.stderr', new=StringIO()) as mock_stderr:
-                sys.argv = ["cmd"]
-                urlscan.main()
-                self.assertEqual(mock_stderr.getvalue(), "usage: cmd [-h] [-c CONFIG_FILE] artifact\ncmd: error: the following arguments are required: artifact\n")
-                sysmock.assert_called_once_with(2)
-
-    def test_main_success(self):
-        output = {"foo": "bar"}
-        with patch('sys.stdout', new=StringIO()) as mock_stdout:
-            with patch('urlscan.urlscan.analyze', new=MagicMock(return_value=output)) as mock:
-                sys.argv = ["cmd", "input"]
-                urlscan.main()
-                expected = '{"foo": "bar"}\n'
-                self.assertEqual(mock_stdout.getvalue(), expected)
-                mock.assert_called_once()
-
-    def test_checkConfigRequirements_notEnabled(self):
-        conf = {"not_a_key": "abcd12345"}
-        with self.assertRaises(SystemExit) as cm:
-            urlscan.checkConfigRequirements(conf)
-        self.assertEqual(cm.exception.code, 126)
-
-    def test_checkConfigRequirements_noApikey(self):
-        conf = {"enabled": True, "not_a_key": "abcd12345"}
-        with self.assertRaises(SystemExit) as cm:
-            urlscan.checkConfigRequirements(conf)
-        self.assertEqual(cm.exception.code, 126)
-
-    def test_checkConfigRequirements_Exist(self):
-        conf = {"enabled": True, "api_key": "abcd12345"}
-        config_exists = urlscan.checkConfigRequirements(conf)
-        self.assertTrue(config_exists)
-
-    def test_buildReq(self):
-        conf = {'base_url': 'https://myurl/api/v1/', 'api_key': 'abcd12345', 'visibility': 'public'}
-        artifact_type = "url"
-        artifact_value = "https://abc.com"
-        result = urlscan.buildReq(conf, artifact_type, artifact_value)
-        self.assertEqual("https://myurl/api/v1/scan/", result[0])
-        self.assertEqual({'API-Key': 'abcd12345'}, result[1])
-
-    def test_sendReq(self):
-        with patch('requests.request', new=MagicMock(return_value=MagicMock())) as mock:
-            headers = {"API-Key": "abcd1234"}
-            data = {"url": "https://urlscan.io", "visibility": "public"}
-            response = urlscan.sendReq("https://myurl", headers=headers, data=data)
-            mock.assert_called_once_with("POST", url="https://myurl", headers={"API-Key": "abcd1234"}, data={"url": "https://urlscan.io", "visibility": "public"})
-            self.assertIsNotNone(response)
-
-    def test_getReport_noRetry(self):
-        output_report = MagicMock()
-        type(output_report).status_code = PropertyMock(return_value=404)
-        output_report_body = {"requests": "body"}
-        output_report.json.return_value = output_report_body
-        with patch('requests.request', new=MagicMock(return_value=output_report)) as mock:    
-            result = urlscan.getReport({'timeout': 0}, "https://abc.com/report")
-            self.assertEqual(404, result.status_code)
-            mock.assert_called_once()
-    
-    def test_getReport_withRetry(self):
-        output_report = MagicMock()
-        type(output_report).status_code = PropertyMock(return_value=404)
-        output_report_body = {"requests": "body"}
-        output_report.json.return_value = output_report_body
-        with patch('requests.request', new=MagicMock(return_value=output_report)) as mock:
-            result = urlscan.getReport({'timeout': 3}, "https://abc.com/report")
-            self.assertEqual(404, result.status_code)
-            mock.assert_has_calls([call('GET', 'https://abc.com/report'),call('GET', 'https://abc.com/report')])       
-    
-    def test_prepareResults_sus(self):
-        raw = {"requests": [{"request": {"requestId": "1"}}], "verdicts": {"overall": {"score": 50, "malicious": False, "hasVerdicts": False}}}
-        results = urlscan.prepareResults(raw)
-        self.assertEqual(results["response"], raw)
-        self.assertEqual(results["summary"], "suspicious")
-        self.assertEqual(results["status"], "caution")
-
-    def test_prepareResults_mal(self):
-        raw = {"requests": [{"request": {"requestId": "2"}}], "verdicts": {"overall": {"score": 100, "malicious": True, "hasVerdicts": False}}}
-        results = urlscan.prepareResults(raw)
-        self.assertEqual(results["response"], raw)
-        self.assertEqual(results["summary"], "malicious")
-        self.assertEqual(results["status"], "threat")
-
-    def test_prepareResults_info(self):
-        raw = {"requests": [{"request": {"requestId": "3"}}], "verdicts": {"overall": {"score": 0, "malicious": False, "hasVerdicts": False}}}
-        results = urlscan.prepareResults(raw)
-        self.assertEqual(results["response"], raw)
-        self.assertEqual(results["summary"], "Scan complete.")
-        self.assertEqual(results["status"], "info")
-
-    def test_prepareResults_error(self):
-        raw = {}
-        results = urlscan.prepareResults(raw)
-        self.assertEqual(results["response"], raw)
-        self.assertEqual(results["summary"], "internal_failure")
-        self.assertEqual(results["status"], "caution")
-
-    def test_analyze(self):
-        output_req = "https://myurl/report" 
-        output_report = MagicMock()
-        output_report_body = {"requests": [{"request": {"requestId": "3"}}], "verdicts": {"overall": {"score": 0, "malicious": False, "hasVerdicts": False}}}
-        output_report.json.return_value = output_report_body 
-        artifactInput = '{"value":"https://abc.com","artifactType":"url"}'
-        conf = {'enabled': True, 'base_url': 'https://myurl/api/v1/', 'api_key': 'abcd12345', 'visibility': 'public'}
-        with patch('urlscan.urlscan.sendReq', new=MagicMock(return_value=output_req)) as mock_req:
-            with patch('urlscan.urlscan.getReport', new=MagicMock(return_value=output_report)) as mock_report:
-                results = urlscan.analyze(conf, artifactInput)
-                self.assertEqual(results["summary"], "Scan complete.")
-                mock_req.assert_called_once()
-                mock_report.assert_called_once()