First ASTGrep service version with debfuscation attempts

ASTGrep/requirements.txt

@@ -1,3 +1,4 @@
 assemblyline-v4-service
 ast-grep-cli
-pylspclient
+pylspclient
+cryptography

ASTGrep/rules/extended/encrypted-js.yml

@@ -1,30 +1,47 @@
 # yaml-language-server: $schema=https://raw.githubusercontent.com/ast-grep/ast-grep/main/schemas/rule.json
-id: obfuscation-encrypted-js
+id: obfuscation-encrypted-js-aes
 language: JavaScript
 message: The code was obfuscated using AES encryption
+severity: error
+metadata:
+  category: obfuscation
+  extended-obfuscation: yes
+  deobfuscate: |
+    {
+      "type": "extract",
+      "steps": [
+        {"func": "encode", "source": "IV", "encoding": "hex"},
+        {"func": "encode", "source": "SALT", "encoding": "hex"},
+        {"func": "encode", "source": "AUTH_TAG", "encoding": "hex"},
+        {"func": "encode", "source": "DATA", "encoding": "hex"},
+        {"func": "scrypt", "output": "KEY"},
+        {"func": "slice", "source": "KEY", "output": "KEY"},
+        {"func": "aes", "source": "DATA"}
+      ]
+    }
 rule:
   pattern:
-    context: const $DECR = $DECRYPT('$DATA', '$AUTH', '$SALT', '$IV', "$MASTER")
+    context: const $DECR = $DECRYPT('$DATA', '$AUTH_TAG', '$SALT', '$IV', "$MASTER_KEY")
     strictness: signature
   # kind: lexical_declaration
 
   follows:
     stopBy: end
     kind: function_declaration
     pattern:
-      context: function $DECRYPT(encdata, $AUTH_TAG_PARAM, $SALT_PARAM, $IV_PARAM, $MASTER_KEY) { $$$ }
+      context: function $DECRYPT(encdata, $AUTH_TAG_PARAM, $SALT_PARAM, $IV_PARAM, $MASTER_PARAM) { $$$ }
       strictness: signature
     has:
       stopBy: end
       all:
         - pattern:
-            context: "const $KEY_2 = crypto.scryptSync($MASTER_KEY, Buffer.from($SALT_PARAM, 'hex'), 64, { N: 16384, r: 8, p: 1 }).slice(0, 32)"
+            context: "const $KEY_2 = crypto.scryptSync($MASTER_PARAM, Buffer.from($SALT_PARAM, 'hex'), $KEY_LENGTH, { N: $OPT_N, r: $OPT_R, p: $OPT_P }).slice($SLICE_START, $SLICE_END)"
             strictness: signature
             # selector: lexical_declaration
             stopBy: end
             preceeds:
               pattern:
-                context: const $DECIPHER = crypto.createDecipheriv('aes-256-gcm', $KEY, Buffer.from($IV_PARAM, 'hex'));
+                context: const $DECIPHER = crypto.createDecipheriv('$AES_MODE', $KEY, Buffer.from($IV_PARAM, 'hex'));
                 strictness: signature
                 stopBy: end
                 preceeds:

ASTGrep/rules/extended/fernet.yml

@@ -1,15 +1,34 @@
 # yaml-language-server: $schema=https://raw.githubusercontent.com/ast-grep/ast-grep/main/schemas/rule.json
 
-id: obfuscation-fernet
-message: Executing code dynamically $CODE
+id: obfuscation-python-fernet
+message: Contains code obfuscated with Fernet encryption
 severity: error # error, warning, info, hint
 language: Python
 rule:
-  pattern: Fernet($PASS).decrypta($DATA)
+  any:
+    - pattern:
+        context: Fernet("$FERNET_KEY").decrypt("$FERNET_SOURCE")
+        strictness: signature
+    - pattern:
+        context: $X.decrypt("$FERNET_SOURCE")
+        selector: expression_statement
+        strictness: signature
+      follows:
+        stopBy: end
+        pattern:
+          context: $X = Fernet("$FERNET_KEY")
+          selector: expression_statement
+          strictness: signature
+    # TODO: other variations
 metadata:
   some: metadata
   extended-obfuscation: yes
-note: test
-
-# utils: Extract repeated rule as local utility here.
-# note: Add detailed explanation for the rule.
+  deobfuscate: |
+    {
+      "type": "extract",
+      "steps": [
+        {"func": "encode", "source": "FERNET_KEY", "encoding": "utf-8"},
+        {"func": "encode", "source": "FERNET_SOURCE", "encoding": "utf-8"},
+        {"func": "fernet"}
+      ]
+    }

ASTGrep/rules/sgconfig.yml

@@ -1,6 +1,6 @@
 ruleDirs:
 - ./extended
-- ./detection
+# - ./detection
 # testConfigs:
 # - testDir: ./rule-tests
 # utilDirs:

ASTGrep/service/al_run.py

@@ -9,7 +9,7 @@
 
 import yaml
 from assemblyline.common.exceptions import RecoverableError
-from assemblyline_v4_service.common.base import UPDATES_DIR, ServiceBase
+from assemblyline_v4_service.common.base import ServiceBase
 from assemblyline_v4_service.common.request import ServiceRequest
 from assemblyline_v4_service.common.result import (
     Result,
@@ -19,6 +19,8 @@
 )
 
 from .controller import (
+    LANGUAGE_TO_EXT,
+    ASTGrepDeobfuscationController,
     ASTGrepLSPController,
     ASTGrepScanController,
     UnsupportedLanguageError,
@@ -27,7 +29,7 @@
 
 configure_yaml()
 
-RULES_DIR = os.path.join(UPDATES_DIR, "sg_rules")
+# RULES_DIR = os.path.join(UPDATES_DIR, "sg_rules")
 
 SEVERITY_TO_HEURISTIC = {
     "INFO": 3,
@@ -48,8 +50,22 @@
 USE_LSP = True
 MAX_LINE_SIZE = 5000
 
+RULES_DIR = [
+    "./rules/extended",
+    # "./rules/detection",
+]
+
 
 class AssemblylineService(ServiceBase):
+    def _read_rules(self):
+        for rule_path in RULES_DIR:
+            for root, _, files in os.walk(rule_path):
+                for file in files:
+                    if file.endswith(".yml") or file.endswith(".yaml"):
+                        with open(os.path.join(root, file), "r") as f:
+                            data = yaml.safe_load(f)
+                        metadata = data.get("metadata", {})
+                        self.metadata_cache[data.get("id")] = metadata
 
     def __init__(self, config=None):
         super().__init__(config)
@@ -62,6 +78,8 @@ def __init__(self, config=None):
             self._astgrep = ASTGrepLSPController(self.log, RULES_DIR)
         else:
             self._astgrep = ASTGrepScanController(self.log, RULES_DIR)
+        self._deobfuscator = ASTGrepDeobfuscationController(self.log, RULES_DIR)
+        self._read_rules()
 
     def start(self):
         self.log.info(f"{self.service_attributes.name} service started")
@@ -72,9 +90,7 @@ def _load_rules(self) -> None:
         # signature client doesn't support joining to a yaml, so we need to recreate it using our delimiter
         os.makedirs(RULES_DIR, exist_ok=True)
         new_rules_dir = tempfile.TemporaryDirectory(prefix="sg_rules_", dir=RULES_DIR)
-        new_deobfuscation_rules_dir = tempfile.TemporaryDirectory(
-            prefix="sg_rules_", dir=RULES_DIR
-        )
+        new_deobfuscation_rules_dir = tempfile.TemporaryDirectory(prefix="sg_rules_", dir=RULES_DIR)
         deobfuscation_files = []
         metadata = {}
         files = []
@@ -199,11 +215,12 @@ def _process_results(self, results: list[dict]) -> Iterable[ResultMultiSection]:
             heuristic = SEVERITY_TO_HEURISTIC.get(str(severity).upper(), 0)
 
             # TODO: Support for attribution
-            metadata = {} # self.metadata_cache.get(rule_id, {})
+            metadata = self.metadata_cache.get(rule_id, {})
             title = metadata.get("title", metadata.get("name", message[:100]))
             attack_id = metadata.get("attack_id")
 
-            is_deobfuscation = metadata.get("deobfuscation-trigger", False)
+            is_deobfuscation = metadata.get("extended-obfuscation", False)
+            self.log.debug("Is deobfuscation: %s", is_deobfuscation)
             self._should_deobfuscate = self._should_deobfuscate or is_deobfuscation
 
             section = ResultTextSection(
@@ -253,7 +270,26 @@ def execute(self, request: ServiceRequest) -> None:
                 json.dump(self._astgrep.last_results, f, indent=2)
             request.add_supplementary(f.name, "astgrep_raw_results.json", "AST-Grep Results")
 
-        # if self._should_deobfuscate:
+        if self._should_deobfuscate:
+            result_no = 1
+            for deobf_result in self._deobfuscator.deobfuscate_file(
+                request.file_path, request.file_type
+            ):
+                path = f"{self.working_directory}/_deobfuscated_code_{result_no}.{LANGUAGE_TO_EXT[request.file_type]}"
+                with open(path, "w+") as f:
+                    f.write(deobf_result)
+                request.add_extracted(
+                    path,
+                    f"_deobfuscated_code_{result_no}.{LANGUAGE_TO_EXT[request.file_type]}",
+                    "Deobfuscated file",
+                )
+            deobf_section = ResultTextSection("Obfuscation found")
+            deobf_section.add_line(
+                "Obfuscation was detected in the file. Extracted deobfuscated code."
+            )
+            deobf_section.set_heuristic(4)
+            result.add_section(deobf_section)
+
         #     deobfuscated_path = self._deobfuscator.deobfuscate_file(
         #         request.file_path, request.file_type
         #     )