Merge pull request #84 from Zeecka/fix_81

Fix 81

aperisolve/analyzers/binwalk.py

@@ -4,7 +4,7 @@ import shutil
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_binwalk(input_img: Path, output_dir: Path) -> None:
@@ -22,6 +22,7 @@ def analyze_binwalk(input_img: Path, output_dir: Path) -> None:
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
         stderr += data.stderr
 
@@ -34,6 +35,7 @@ def analyze_binwalk(input_img: Path, output_dir: Path) -> None:
                 capture_output=True,
                 text=True,
                 check=False,
+                timeout=MAX_PENDING_TIME,
             )
             zip_exist = True
             stderr += zip_data.stderr

aperisolve/analyzers/exiftool.py

@@ -3,7 +3,7 @@
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_exiftool(input_img: Path, output_dir: Path) -> None:
@@ -11,7 +11,11 @@ def analyze_exiftool(input_img: Path, output_dir: Path) -> None:
 
     try:
         data = subprocess.run(
-            ["exiftool", input_img], capture_output=True, text=True, check=True
+            ["exiftool", input_img],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=MAX_PENDING_TIME,
         )
 
         if data.stderr:

aperisolve/analyzers/foremost.py

@@ -4,7 +4,7 @@ import shutil
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_foremost(input_img: Path, output_dir: Path) -> None:
@@ -21,6 +21,7 @@ def analyze_foremost(input_img: Path, output_dir: Path) -> None:
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
         # Note: foremost use stderr as standard output for results :o)
         # So we can't use it as error dectection :
@@ -33,6 +34,7 @@ def analyze_foremost(input_img: Path, output_dir: Path) -> None:
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         stderr += zip_data.stderr

aperisolve/analyzers/outguess.py

@@ -5,7 +5,7 @@ import subprocess
 from pathlib import Path
 from typing import Optional
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_outguess(
@@ -38,6 +38,7 @@ def analyze_outguess(
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         if data.returncode != 0:
@@ -58,6 +59,7 @@ def analyze_outguess(
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         stderr += zip_data.stderr

aperisolve/analyzers/steghide.py

@@ -6,7 +6,7 @@ import subprocess
 from pathlib import Path
 from typing import Optional
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_steghide(
@@ -30,6 +30,7 @@ def analyze_steghide(
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         match = re.search(r'embedded file "([^"]+)"', data.stdout)
@@ -75,6 +76,7 @@ def analyze_steghide(
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
         stderr += data.stderr
 
@@ -85,6 +87,7 @@ def analyze_steghide(
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         stderr += zip_data.stderr

aperisolve/analyzers/strings.py

@@ -3,7 +3,7 @@
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_strings(input_img: Path, output_dir: Path) -> None:
@@ -11,7 +11,11 @@ def analyze_strings(input_img: Path, output_dir: Path) -> None:
 
     try:
         data = subprocess.run(
-            ["strings", input_img], capture_output=True, text=True, check=False
+            ["strings", input_img],
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=MAX_PENDING_TIME,
         )
         if data.stderr:
             update_data(

aperisolve/analyzers/template_analyzer.py

@@ -4,7 +4,7 @@ import shutil
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None:
@@ -25,6 +25,7 @@ def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None:
             capture_output=True,
             text=True,
             check=False,
+            timeout=MAX_PENDING_TIME,
         )
 
         stderr += data.stderr
@@ -48,6 +49,7 @@ def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None:
                 capture_output=True,
                 text=True,
                 check=False,
+                timeout=MAX_PENDING_TIME,
             )
             zip_exist = True
 

aperisolve/analyzers/utils.py

@@ -9,6 +9,8 @@ from typing import Any
 
 _thread_lock = threading.Lock()
 
+MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600"))  # 10 minutes by default
+
 
 def update_data(
     output_dir: Path, new_data: dict[Any, Any], json_filename: str = "results.json"

aperisolve/analyzers/zsteg.py

@@ -3,7 +3,7 @@
 import subprocess
 from pathlib import Path
 
-from .utils import update_data
+from .utils import MAX_PENDING_TIME, update_data
 
 
 def analyze_zsteg(input_img: Path, output_dir: Path) -> None:
@@ -11,7 +11,11 @@ def analyze_zsteg(input_img: Path, output_dir: Path) -> None:
 
     try:
         data = subprocess.run(
-            ["zsteg", input_img], capture_output=True, text=True, check=False
+            ["zsteg", input_img],
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=MAX_PENDING_TIME,
         )
         if data.stderr or "PNG::NotSupported" in data.stdout[:100]:
             error_msg = data.stderr or data.stdout