Merge pull request #49 from ajinabraham/semgrep_bump

Bump semgrep to 1.86.0

Author: ajinabraham

.github/workflows/python_test.yml

@@ -36,7 +36,7 @@ jobs:
         poetry install --no-interaction --no-ansi
     - name: Bandit Scan
       run: |
-        poetry run bandit libsast -r
+        poetry run bandit -ll libsast -r
     - name: Unit test
       run: |
         poetry run pytest -v --cache-clear tests

libsast/__init__.py

@@ -12,7 +12,7 @@
 __title__ = 'libsast'
 __authors__ = 'Ajin Abraham'
 __copyright__ = f'Copyright {year} Ajin Abraham, opensecurity.in'
-__version__ = '3.0.2'
+__version__ = '3.1.0'
 __version_info__ = tuple(int(i) for i in __version__.split('.'))
 __all__ = [
     'Scanner',

libsast/core_sgrep/helpers.py

@@ -1,68 +1,29 @@
 # -*- coding: utf_8 -*-
 """Semantic Grep Helpers."""
 import json
-import logging
 import platform
-import multiprocessing
+import subprocess
 
 
-def invoke_semgrep(paths, scan_rules, **kwargs):
-    """Call Semgrep."""
+def invoke_semgrep(paths, scan_rules):
     if platform.system() == 'Windows':
         return None
-    from semgrep import semgrep_main
-    from semgrep.state import get_state
-    from semgrep.constants import OutputFormat
-    from semgrep.output import OutputHandler, OutputSettings
-    try:
-        cpu_count = multiprocessing.cpu_count()
-    except NotImplementedError:
-        cpu_count = 1  # CPU count is not implemented on Windows
-    # Semgrep output formatting
-    state = get_state()
-    state.terminal.configure(
-        verbose=False,
-        debug=False,
-        quiet=True,
-        force_color=False,
-    )
-    logging.getLogger('semgrep').propagate = False
-    output_settings = OutputSettings(
-        output_format=OutputFormat.JSON,
-        output_destination=None,
-        output_per_finding_max_lines_limit=None,
-        output_per_line_max_chars_limit=None,
-        error_on_findings=False,
-        verbose_errors=False,
-        strict=False,
-        timeout_threshold=3,
-    )
-    output_handler = OutputHandler(output_settings)
-    (
-        filtered_matches_by_rule,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-        _,
-    ) = semgrep_main.main(
-        output_handler=output_handler,
-        target=[pt.as_posix() for pt in paths],
-        jobs=cpu_count,
-        pattern=None,
-        lang=None,
-        configs=[scan_rules],
-        timeout=5,
-        timeout_threshold=3,
-        **kwargs,
-    )
-    output_handler.rule_matches = [
-        m for ms in filtered_matches_by_rule.values() for m in ms
+    ps = [pt.as_posix() for pt in paths]
+    command = [
+        'semgrep',
+        '--metrics=off',
+        '--no-rewrite-rule-ids',
+        '--json',
+        '-q',
+        '--config',
+        scan_rules,
+        *ps,
     ]
-    return json.loads(output_handler._build_output())
+    try:
+        result = subprocess.run(command, capture_output=True, text=True, check=True)
+        return json.loads(result.stdout)
+    except subprocess.CalledProcessError as e:
+        try:
+            return json.loads(e.output)
+        except json.JSONDecodeError:
+            return {'errors': e.output}

libsast/core_sgrep/semantic_sgrep.py

@@ -43,9 +43,11 @@ def scan(self, paths: list) -> dict:
 
     def format_output(self, results):
         """Format sgrep results."""
-        errs = self.findings.get('errors')
+        errs = results.get('errors')
         if errs:
             self.findings['errors'] = errs
+        if not results.get('results'):
+            return
         smatches = self.findings['matches']
         for find in results['results']:
             file_details = {
@@ -54,7 +56,7 @@ def format_output(self, results):
                 'match_lines': (find['start']['line'], find['end']['line']),
                 'match_string': find['extra']['lines'],
             }
-            rule_id = find['check_id'].rsplit('.', 1)[1]
+            rule_id = find['check_id']
             if rule_id in smatches:
                 smatches[rule_id]['files'].append(file_details)
             else: