Added a new AI-based implementation for malware detection in Android applications

.gitignore

@@ -79,4 +79,5 @@ mobsf/secret
 mobsf/StaticAnalyzer/migrations
 mobsf/MobSF/windows_vm_priv_key.asc
 mobsf/setup_done.txt
+mobsf/StaticAnalyzer/tools/IA_model/**
 TODO.md

mobsf/MobSF/settings.py

@@ -538,3 +538,7 @@
     # CORELLIUM_PROJECT_ID is optional, MobSF will use any available project id
     # ===============================================
     # ^CONFIG-END^: Do not edit this line
+
+
+# IA Settings
+IA_DANGER_PERCENTAGE = os.getenv('MOBSF_IA_DANGER_PERCENTAGE', 60)

mobsf/StaticAnalyzer/models.py

@@ -83,6 +83,7 @@ class Meta:
     NETWORK_SECURITY = models.TextField(default=[])
     SECRETS = models.TextField(default=[])
     SBOM = models.TextField(default={})
+    IA_MALWARE_PERCENTAGE = models.TextField(default=[])
 
 
 class StaticAnalyzerIOS(models.Model):

mobsf/StaticAnalyzer/tests.py

@@ -20,6 +20,7 @@
 def static_analysis_test():
     """Test Static Analyzer."""
     logger.info('Running Static Analyzer Unit test')
+    auth = api_key(settings.MOBSF_HOME)
     try:
         uploaded = []
         logger.info('Running Upload Test')
@@ -32,9 +33,13 @@ def static_analysis_test():
                 continue
             fpath = os.path.join(apk_dir, filename)
             with open(fpath, 'rb') as file_pointer:
+                print(f"[+] Archivo: {fpath}")
+                print("[*]¨Auth: {}".format(auth))
                 response = http_client.post(
-                    '/upload/',
-                    {'file': file_pointer})
+                    '/api/v1/upload',
+                    {'file': file_pointer},
+                    HTTP_AUTHORIZATION=auth)
+
                 obj = json.loads(response.content.decode('utf-8'))
                 if response.status_code == 200 and obj['status'] == 'success':
                     logger.info('[OK] Upload OK: %s', filename)

mobsf/StaticAnalyzer/views/android/apk.py

@@ -83,6 +83,7 @@
     Permissions,
     has_permission,
 )
+from mobsf.StaticAnalyzer.views.common.IA_function import validate_malware_ia
 
 logger = logging.getLogger(__name__)
 
@@ -220,6 +221,9 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
         code_an_dic['domains'] = MalwareDomainCheck().scan(
             checksum,
             code_an_dic['urls_list'])
+
+        ia_analisis = validate_malware_ia(man_data_dic['perm'])
+
         context = save_get_ctx(
             app_dic,
             man_data_dic,
@@ -230,6 +234,7 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
             apkid_results,
             trackers,
             rescan,
+            ia_analisis,
         )
         if queue:
             return mark_task_completed(

mobsf/StaticAnalyzer/views/android/db_interaction.py

@@ -36,6 +36,7 @@ def get_context_from_db_entry(db_entry: QuerySet) -> dict:
         manifest_analysis = process_suppression_manifest(
             python_list(db_entry[0].MANIFEST_ANALYSIS),
             package)
+        ia_malware_percentage = python_dict(db_entry[0].IA_MALWARE_PERCENTAGE)
         context = {
             'version': settings.MOBSF_VER,
             'title': 'Static Analysis',
@@ -89,7 +90,10 @@ def get_context_from_db_entry(db_entry: QuerySet) -> dict:
             'secrets': python_list(db_entry[0].SECRETS),
             'logs': get_scan_logs(db_entry[0].MD5),
             'sbom': python_dict(db_entry[0].SBOM),
+            'IA_MALWARE_PERCENTAGE': ia_malware_percentage * 100,
+            'IA_DANGER_PERCENTAGE': getattr(settings, 'IA_DANGER_PERCENTAGE', 20),
         }
+
         return context
     except Exception:
         msg = 'Fetching data from the DB failed.'
@@ -104,7 +108,8 @@ def get_context_from_analysis(app_dic,
                               cert_dic,
                               bin_anal,
                               apk_id,
-                              trackers) -> dict:
+                              trackers,
+                              ia_analisis) -> dict:
     """Get the context for APK/ZIP from analysis results."""
     try:
         package = man_data_dic['packagename']
@@ -114,6 +119,12 @@ def get_context_from_analysis(app_dic,
         manifest_analysis = process_suppression_manifest(
             man_an_dic['manifest_anal'],
             package)
+
+        if ia_analisis:
+            ia_malware_percentage  = float(ia_analisis.get('IA_MALWARE_PERCENTAGE', 0))
+        else: 
+            ia_malware_percentage  = 0 
+
         context = {
             'title': 'Static Analysis',
             'version': settings.MOBSF_VER,
@@ -164,7 +175,10 @@ def get_context_from_analysis(app_dic,
             'secrets': code_an_dic['secrets'],
             'logs': get_scan_logs(app_dic['md5']),
             'sbom': code_an_dic['sbom'],
+            'IA_MALWARE_PERCENTAGE': ia_malware_percentage * 100,
+            'IA_DANGER_PERCENTAGE': getattr(settings, 'IA_DANGER_PERCENTAGE', 0),
         }
+
         return context
     except Exception as exp:
         msg = 'Rendering to Template failed.'
@@ -180,7 +194,8 @@ def save_or_update(update_type,
                    cert_dic,
                    bin_anal,
                    apk_id,
-                   trackers) -> None:
+                   trackers,
+                   ia_analisis) -> None:
     """Save/Update an APK/ZIP DB entry."""
     try:
         values = {
@@ -231,6 +246,11 @@ def save_or_update(update_type,
             'SECRETS': code_an_dic['secrets'],
             'SBOM': code_an_dic['sbom'],
         }
+        if ia_analisis:
+            values['IA_MALWARE_PERCENTAGE'] = ia_analisis.get('IA_MALWARE_PERCENTAGE', '0')
+        else:
+            values['IA_MALWARE_PERCENTAGE'] = 0 
+
         if update_type == 'save':
             db_entry = StaticAnalyzerAndroid.objects.filter(
                 MD5=app_dic['md5'])
@@ -257,7 +277,7 @@ def save_or_update(update_type,
         append_scan_status(app_dic['md5'], msg, repr(exp))
 
 
-def save_get_ctx(app, man, m_anal, code, cert, elf, apkid, trk, rscn):
+def save_get_ctx(app, man, m_anal, code, cert, elf, apkid, trk, rscn, ia_analisis=None):
     # SAVE TO DB
     if rscn:
         msg = 'Updating Database...'
@@ -280,6 +300,7 @@ def save_get_ctx(app, man, m_anal, code, cert, elf, apkid, trk, rscn):
         elf,
         apkid,
         trk,
+        ia_analisis,
     )
     return get_context_from_analysis(
         app,
@@ -290,4 +311,5 @@ def save_get_ctx(app, man, m_anal, code, cert, elf, apkid, trk, rscn):
         elf,
         apkid,
         trk,
+        ia_analisis,
     )

mobsf/StaticAnalyzer/views/common/IA_function.py

@@ -0,0 +1,50 @@
+from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+import torch
+import torch.nn.functional as functional
+import os 
+
+
+def array_to_str(dataset):
+
+    if isinstance(dataset, (list, tuple)):
+        cadena = ''
+        for x in dataset:
+            cadena += x + ','
+    else:
+        cadena = dataset
+    cadena.strip()
+    cadena = cadena[:-1]
+    if len(cadena) == 0:
+        cadena = 'Sin_permisos_definidos'
+
+    return cadena
+
+
+def validate_malware_ia(data):
+
+    permission = list(data.keys()) if data.keys() else []
+    permission = array_to_str(permission)
+
+    model_path = os.path.join("mobsf", "StaticAnalyzer", "tools", "IA_model", "NyerAndroidMalware")
+    #model_path = 'mobsf\\StaticAnalyzer\\tools\\IA_model\\NyerAndroidMalware'
+
+    model = AutoModelForSequenceClassification.from_pretrained(model_path)
+    tokenizer = AutoTokenizer.from_pretrained(model_path)
+
+    inputs = tokenizer(permission, return_tensors='pt', truncation=True, padding=True)
+
+
+    with torch.no_grad():
+        outputs = model(**inputs)
+        logits = outputs.logits
+
+    probs = functional.softmax(logits, dim=1)
+
+
+    prob_benign, prob_malware = probs[0].tolist()
+
+    print(f'Probabilidad Benigno: {prob_benign * 100:.2f}%')
+    print(f'Probabilidad Malware: {prob_malware * 100:.2f}%')
+
+    return {'IA_MALWARE_PERCENTAGE': prob_malware}

mobsf/templates/static_analysis/android_binary_analysis.html

@@ -433,6 +433,8 @@ <h5 class="card-title"></h5>
                       <span class="badge bg-success">VirusTotal Detection</span> <strong>{{ virus_total.positives }}/{{ virus_total.total }}</strong><br/>
                     {% endif %}
                   {% endif %}
+                  <span class="badge {% if IA_MALWARE_PERCENTAGE >= IA_DANGER_PERCENTAGE %}bg-danger{% else %}bg-success{% endif %}">IA MALWARE PERCENTAGE</span>
+                  {{ IA_MALWARE_PERCENTAGE | floatformat:2 }}%<br/>
                 </p>
                 {% if app_type not in 'so' %}
                   <p> <a href="{% url 'appsec_dashboard' checksum=md5 %}" class="btn btn-outline-primary btn-sm" role="button"><i class="fas fa-user-shield"></i> MobSF Scorecard</a></p>