Merge pull request #1 from CybercentreCanada/assemblyline/issues/181

Move url_analysis from NetRep → URLCreator

.vscode/settings.json

@@ -1,32 +1,33 @@
 {
-    "editor.codeActionsOnSave": {
-        "source.organizeImports": true,
-    },
-    "editor.formatOnSave": true,
-    "editor.rulers": [
-        120
-    ],
-    "editor.tabSize": 4,
-    "editor.wordWrap": "wordWrapColumn",
-    "editor.wordWrapColumn": 120,
-    "files.insertFinalNewline": true,
-    "files.trimFinalNewlines": true,
-    "files.trimTrailingWhitespace": true,
-    "isort.args": [
-        "-l",
-        "120",
-        "--profile=black",
-        // "--src=${workspaceFolder}"
-    ],
-    "[python]": {
-        "editor.defaultFormatter": "ms-python.black-formatter"
-    },
-    "black-formatter.args": [
-        "--line-length=120"
-    ],
-    "flake8.args": [
-        "--max-line-length=120",
-        //Added the ignore of E203 for now : https://github.com/PyCQA/pycodestyle/issues/373
-        "--ignore=E203,W503"
-    ],
+  "editor.codeActionsOnSave": {
+    "source.organizeImports": true
+  },
+  "editor.formatOnSave": true,
+  "editor.rulers": [120],
+  "editor.tabSize": 4,
+  "editor.wordWrap": "wordWrapColumn",
+  "editor.wordWrapColumn": 120,
+  "files.insertFinalNewline": true,
+  "files.trimFinalNewlines": true,
+  "files.trimTrailingWhitespace": true,
+  "isort.args": [
+    "-l",
+    "120",
+    "--profile=black"
+    // "--src=${workspaceFolder}"
+  ],
+  "[python]": {
+    "editor.defaultFormatter": "ms-python.black-formatter"
+  },
+  "black-formatter.args": ["--line-length=120"],
+  "flake8.args": [
+    "--max-line-length=120",
+    //Added the ignore of E203 for now : https://github.com/PyCQA/pycodestyle/issues/373
+    "--ignore=E203,W503"
+  ],
+  "python.testing.pytestArgs": [
+    "tests"
+  ],
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true
 }

Dockerfile

@@ -3,6 +3,10 @@ FROM cccs/assemblyline-v4-service-base:$branch
 
 ENV SERVICE_PATH urlcreator.URLCreator
 
+# Install python dependencies
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir --user --requirement requirements.txt && rm -rf ~/.cache/pip
+
 # Switch to assemblyline user
 USER assemblyline
 

requirements.txt

@@ -0,0 +1 @@
+multidecoder==1.0.0

service_manifest.yml

@@ -56,6 +56,12 @@ heuristics:
       linux: 0
       recon: 0
 
+  - heur_id: 4
+    name: Phishing
+    score: 500
+    filetype: .*
+    description: This IOC seems to be phishing-related
+
 docker_config:
   image: ${REGISTRY}cccs/assemblyline-service-urlcreator:$SERVICE_TAG
   cpu_cores: 1

tests/test_url_analysis.py

@@ -0,0 +1,72 @@
+from utils.network import url_analysis
+
+
+def test_unicode_characters():
+    # Shouldn't crash due to non-ASCII characters being in the URL
+    url = "https://hello:world@écoute.com/"
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs == {"uri": ["https://écoute.com/"], "domain": ["écoute.com"], "ip": []}
+    assert '"OBFUSCATION": "Embedded credentials"' in res_section.body
+
+
+def test_embedded_base64():
+    url = "https://somedomain.com/some/path?u=a1aHR0cHM6Ly9iYWQuY29t#dGVzdEBleGFtcGxlLmNvbQ=="
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs == {"uri": ["https://bad.com"], "domain": ["bad.com"], "ip": []}
+    assert res_section.tags == {
+        # Encoded URL in query
+        "network.static.uri": ["https://bad.com"],
+        # Domain from encoded URL
+        "network.static.domain": ["bad.com"],
+        # Encoded email in fragment
+        "network.email.address": ["[email protected]"],
+    }
+
+    # Handle garbage base64 strings, this shouldn't generate any results
+    url = "https://somedomain.com/some/path?u=2F4wOWl6vSIfij9tBVr7MyOThiV1"
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs == {"uri": [], "domain": [], "ip": []}
+    assert not res_section.body
+
+
+def test_safelinks():
+    # Ref: https://learn.microsoft.com/en-us/microsoft-365/security/office-365-security/safe-links-about?view=o365-worldwide
+    url = "https://safelinks.com/?url=https%3A%2F%2Fhelloworld%2Ecom%2Fbad%7C01%7Ctest%40example%2Ecom"
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs == {
+        # URL to be redirected to
+        "uri": ["https://helloworld.com/bad"],
+        "domain": ["helloworld.com"],
+        "ip": [],
+    }
+    assert res_section.tags == {
+        # URL to be redirected to
+        "network.static.uri": ["https://helloworld.com/bad"],
+        "network.static.domain": ["helloworld.com"],
+        # Recipient email address
+        "network.email.address": ["[email protected]"],
+    }
+
+
+def test_hexed_ip():
+    # Ref: https://www.darkreading.com/cloud/shellbot-cracks-linux-ssh-servers-debuts-new-evasion-tactic
+    url = "http://0x7f000001"
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs["ip"] == ["127.0.0.1"]
+    assert res_section.tags == {"network.static.ip": ["127.0.0.1"]}
+
+
+def test_phishing():
+    # Ref: https://www.vmray.com/cyber-security-blog/detection-signature-updates-2/#elementor-toc__heading-anchor-9
+    url = "https://[email protected]@[email protected]@bad.com/malicious.zip?evil=true"
+    res_section, network_iocs = url_analysis(url)
+    # Should reveal the true target URL for reputation checking
+    assert network_iocs["uri"] == ["https://bad.com/malicious.zip?evil=true"]
+    assert network_iocs["domain"] == ["bad.com"]
+    assert '"OBFUSCATION": "URL masquerade"' in res_section.body
+
+    url = "https://[email protected]/"
+    res_section, network_iocs = url_analysis(url)
+    assert network_iocs["uri"] == ["https://bad.com/"]
+    assert network_iocs["domain"] == ["bad.com"]
+    assert '"OBFUSCATION": "Embedded credentials"' in res_section.body

urlcreator.py

@@ -5,9 +5,11 @@
 from assemblyline.odm.base import IP_ONLY_REGEX, IPV4_ONLY_REGEX
 from assemblyline_v4_service.common.base import ServiceBase
 from assemblyline_v4_service.common.request import ServiceRequest
-from assemblyline_v4_service.common.result import Heuristic, Result, ResultTableSection, ResultTextSection, TableRow
+from assemblyline_v4_service.common.result import Heuristic, Result, ResultSection, ResultTableSection, ResultTextSection, TableRow
 from assemblyline_v4_service.common.task import MaxExtractedExceeded
 
+from utils.network import url_analysis
+
 # Threshold to trigger heuristic regarding high port usage in URI
 HIGH_PORT_MINIMUM = 1024
 
@@ -55,12 +57,18 @@ def execute(self, request: ServiceRequest) -> None:
         high_port_table = ResultTableSection("High Port Usage", heuristic=Heuristic(2))
         tool_table = ResultTableSection("Discovery Tool Found in URI Path", heuristic=Heuristic(3))
         max_extracted_section = ResultTextSection("Too many URI files to be created")
+        url_analysis_section = ResultSection("URL Analysis")
 
         for tag_value, tag_score in urls:
             # Analyse the URL for the possibility of it being a something we should download
             parsed_url = urlparse(tag_value)
             interesting_features = []
 
+            # Look for data that might be embedded in URLs
+            analysis_table, _ = url_analysis(tag_value)
+            if analysis_table.body:
+                url_analysis_section.add_subsection(analysis_table)
+
             if tag_score >= minimum_maliciousness:
                 scoring_uri.add_row(TableRow(dict(URL=tag_value, Score=tag_score)))
                 scoring_uri.add_tag("network.static.uri", tag_value)
@@ -132,3 +140,5 @@ def execute(self, request: ServiceRequest) -> None:
             request.result.add_section(tool_table)
         if max_extracted_section.body:
             request.result.add_section(max_extracted_section)
+        if url_analysis_section.body:
+            request.result.add_section(url_analysis_section)