Update analysers (WIP)
This commit is contained in:
Zeecka
@@ -3,75 +3,27 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Binwalk Analyzer for Image Submissions."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class BinwalkAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for binwalk."""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__("binwalk", *args, has_archive=True)
|
||||
self.cmd = ["binwalk", "--matryoshka", "-e", self.img, "--run-as=root"]
|
||||
self.make_folder = False # Binwalk already create its own folder
|
||||
|
||||
def get_extracted_dir(self) -> Path:
|
||||
return self.output_dir / f"_{self.input_img.name}.extracted"
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
return len(stderr) > 0 and not zip_exist and False
|
||||
|
||||
|
||||
def analyze_binwalk(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using binwalk."""
|
||||
|
||||
image_name = input_img.name
|
||||
extracted_dir = output_dir / f"_{image_name}.extracted"
|
||||
|
||||
try:
|
||||
stderr = ""
|
||||
# Run binwalk
|
||||
data = subprocess.run(
|
||||
["binwalk", "--matryoshka", "-e", "../" + str(image_name), "--run-as=root"],
|
||||
cwd=output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
stderr += data.stderr
|
||||
|
||||
zip_exist = False
|
||||
if extracted_dir.exists():
|
||||
# Zip extracted files
|
||||
zip_data = subprocess.run(
|
||||
["7z", "a", "../binwalk.7z", "*"],
|
||||
cwd=extracted_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
zip_exist = True
|
||||
stderr += zip_data.stderr
|
||||
|
||||
# Remove the extracted directory
|
||||
if extracted_dir.exists():
|
||||
shutil.rmtree(extracted_dir)
|
||||
|
||||
# Only report error if stderr exists AND extraction failed.
|
||||
# If zip_exist is True, binwalk worked (despite warnings), so we proceed.
|
||||
if len(stderr) > 0 and not zip_exist:
|
||||
err: dict[str, dict[str, Any]] = {
|
||||
"binwalk": {
|
||||
"status": "error",
|
||||
"output": data.stdout.split("\n") if data else [],
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
output_data: dict[str, dict[str, Any]] = {
|
||||
"binwalk": {
|
||||
"status": "ok",
|
||||
"output": data.stdout.split("\n") if data else [],
|
||||
}
|
||||
}
|
||||
if zip_exist:
|
||||
output_data["binwalk"]["download"] = f"/download/{output_dir.name}/binwalk"
|
||||
|
||||
update_data(output_dir, output_data)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"binwalk": {"status": "error", "error": str(e)}})
|
||||
return None
|
||||
analyzer = BinwalkAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -3,46 +3,29 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Exiftool Analyzer for Image Submissions."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class ExiftoolAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for exiftool."""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__("exiftool", *args)
|
||||
self.cmd = ["exiftool", "-a", "-u", "-g1", self.img]
|
||||
|
||||
def process_output(self, stdout: str, stderr: str) -> dict[str, str]:
|
||||
"""Process the stdout into a list of lines."""
|
||||
metadata: dict[str, str] = {}
|
||||
for line in stdout.split("\n"):
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
metadata[key.strip()] = value.strip()
|
||||
return metadata
|
||||
|
||||
|
||||
def analyze_exiftool(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using exiftool."""
|
||||
|
||||
try:
|
||||
data = subprocess.run(
|
||||
["exiftool", input_img],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
metadata: dict[str, str] = {}
|
||||
for line in data.stdout.split("\n"):
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
metadata[key.strip()] = value.strip()
|
||||
|
||||
if data.stderr:
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"exiftool": {
|
||||
"status": "error",
|
||||
"output": metadata,
|
||||
"error": data.stderr,
|
||||
}
|
||||
},
|
||||
)
|
||||
return
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{"exiftool": {"status": "ok", "output": metadata}},
|
||||
)
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"exiftool": {"status": "error", "error": str(e)}})
|
||||
analyzer = ExiftoolAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -3,60 +3,30 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Foremost Analyzer for Image Submissions."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class ForemostAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for foremost."""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__("foremost", *args, has_archive=True)
|
||||
self.cmd = ["foremost", "-o", str(self.get_extracted_dir()), "-i", self.img]
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
"""Check if the result is an error."""
|
||||
return len(stderr) > 60
|
||||
|
||||
def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
|
||||
"""Process the stdout/stderr."""
|
||||
if "Processing" in stderr and "|*|" in stderr:
|
||||
return stderr.strip().split("\n")
|
||||
return []
|
||||
|
||||
|
||||
def analyze_foremost(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using foremost."""
|
||||
|
||||
try:
|
||||
foremost_dir = output_dir / "foremost"
|
||||
foremost_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
stderr = ""
|
||||
# Run foremost
|
||||
data = subprocess.run(
|
||||
["foremost", "-o", foremost_dir, "-i", input_img],
|
||||
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 detection :
|
||||
# stderr += data.stderr
|
||||
|
||||
# Zip extracted files
|
||||
zip_data = subprocess.run(
|
||||
["7z", "a", "../foremost.7z", "*"],
|
||||
cwd=foremost_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
stderr += zip_data.stderr
|
||||
|
||||
# Remove the extracted directory
|
||||
shutil.rmtree(foremost_dir)
|
||||
|
||||
data_strings = data.stdout.split("\n") if data else []
|
||||
data_strings = [s for s in data_strings if s]
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"foremost": {
|
||||
"status": "ok",
|
||||
"output": data_strings,
|
||||
"download": f"/download/{output_dir.name}/foremost",
|
||||
}
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"foremost": {"status": "error", "error": str(e)}})
|
||||
analyzer = ForemostAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -3,97 +3,32 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Outguess Analyzer for Image Submissions."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
def analyze_outguess(
|
||||
input_img: Path, output_dir: Path, password: Optional[str] = None
|
||||
) -> None:
|
||||
"""Analyze an image submission using outguess."""
|
||||
class OutguessAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for outguess."""
|
||||
|
||||
image_name = input_img.name
|
||||
try:
|
||||
outguess_dir = output_dir / "outguess"
|
||||
outguess_dir.mkdir(parents=True, exist_ok=True)
|
||||
outguess_file = "outguess.data"
|
||||
stderr = ""
|
||||
cmd: list[str]
|
||||
def __init__(self, *args):
|
||||
super().__init__("outguess", *args, has_archive=True)
|
||||
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
extracted_dir = self.get_extracted_dir()
|
||||
assert extracted_dir is not None # since has_archive is True
|
||||
out = str(extracted_dir / "outguess.data")
|
||||
if password:
|
||||
cmd = [
|
||||
"outguess",
|
||||
"-k",
|
||||
password,
|
||||
"-r",
|
||||
"../../" + str(image_name),
|
||||
outguess_file,
|
||||
]
|
||||
return ["outguess", "-k", password, "-r", self.img, out]
|
||||
else:
|
||||
cmd = ["outguess", "-r", "../../" + str(image_name), outguess_file]
|
||||
return ["outguess", "-r", self.img, out]
|
||||
|
||||
data = subprocess.run(
|
||||
cmd,
|
||||
cwd=outguess_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
if data.returncode != 0:
|
||||
stderr += data.stderr
|
||||
err = {
|
||||
"outguess": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Zip extracted files
|
||||
zip_data = subprocess.run(
|
||||
["7z", "a", "../outguess.7z", "*"],
|
||||
cwd=outguess_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
stderr += zip_data.stderr
|
||||
|
||||
if len(stderr) > 0:
|
||||
err = {
|
||||
"outguess": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Remove the extracted directory
|
||||
shutil.rmtree(outguess_dir)
|
||||
|
||||
data_strings = data.stderr.split("\n") if data else []
|
||||
data_strings = [s for s in data_strings if s]
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"outguess": {
|
||||
"status": "ok",
|
||||
"output": data_strings,
|
||||
"download": f"/download/{output_dir.name}/outguess",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"outguess": {"status": "error", "error": str(e)}})
|
||||
return None
|
||||
def analyze_outguess(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
|
||||
"""Analyze an image submission using outguess."""
|
||||
analyzer = OutguessAnalyzer(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -3,54 +3,23 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Pngcheck Analyzer for Image Submissions."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class PngcheckAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for pngcheck."""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__("pngcheck", *args)
|
||||
self.cmd = ["pngcheck", "-v", self.img]
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
return "this is neither a PNG or JNG image nor a MNG stream" in stdout
|
||||
|
||||
|
||||
def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using pngcheck."""
|
||||
|
||||
try:
|
||||
data = subprocess.run(
|
||||
["pngcheck", "-v", input_img],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
# Error based on https://github.com/pnggroup/pngcheck/issues/59
|
||||
# if data.stderr:
|
||||
# update_data(
|
||||
# output_dir, {"pngcheck": {"status": "error", "error": data.stderr}}
|
||||
# )
|
||||
# return
|
||||
|
||||
if "this is neither a PNG or JNG image nor a MNG stream" in data.stdout:
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"pngcheck": {
|
||||
"status": "error",
|
||||
"error": "ERRORS DETECTED: this is neither a PNG or JNG image nor a "
|
||||
"MNG stream.",
|
||||
}
|
||||
},
|
||||
)
|
||||
return
|
||||
|
||||
data_pngcheck = data.stdout.split("\n") if data else []
|
||||
data_pngcheck = [s for s in data_pngcheck if s]
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"pngcheck": {
|
||||
"status": "ok",
|
||||
"output": data_pngcheck,
|
||||
}
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"pngcheck": {"status": "error", "error": str(e)}})
|
||||
analyzer = PngcheckAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -4,127 +4,67 @@
|
||||
"""Steghide Analyzer for Image Submissions."""
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
def analyze_steghide(
|
||||
input_img: Path, output_dir: Path, password: Optional[str] = None
|
||||
) -> None:
|
||||
"""Analyze an image submission using steghide."""
|
||||
class SteghideAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for steghide."""
|
||||
|
||||
image_name = input_img.name
|
||||
try:
|
||||
stderr = ""
|
||||
# Get embedded file information first
|
||||
cmd: list[str] = ["steghide", "info", "../" + str(image_name)]
|
||||
if password:
|
||||
cmd += ["-p", password]
|
||||
else:
|
||||
cmd += ["-p", ""]
|
||||
def __init__(self, *args):
|
||||
super().__init__("steghide", *args, has_archive=True)
|
||||
|
||||
data = subprocess.run(
|
||||
cmd,
|
||||
cwd=output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
if password is None:
|
||||
password = ""
|
||||
|
||||
# First, get the embedded file name, and return the test cmd if failed
|
||||
cmd_test = ["steghide", "info", self.img, "-p", password]
|
||||
data = self.run_command(cmd_test, cwd=self.output_dir)
|
||||
err = self.is_error(data.returncode, data.stdout, data.stderr, False)
|
||||
if err:
|
||||
return cmd_test
|
||||
|
||||
match = re.search(r'embedded file "([^"]+)"', data.stdout)
|
||||
assert match is not None
|
||||
hidden = match.group(1)
|
||||
cmd = ["steghide", "extract", "-sf", self.img, "-xf", hidden, "-p", password]
|
||||
return cmd
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
"""Check if the result is an error."""
|
||||
|
||||
match = re.search(r'embedded file "([^"]+)"', stdout)
|
||||
embedded_filename = match.group(1) if match else None
|
||||
|
||||
# if bad file format or wrong password, raise an error
|
||||
if data.returncode != 0 or embedded_filename is None:
|
||||
stderr += data.stderr.replace(f'"../{image_name}" ', "") # hide file name
|
||||
# the file format of the file \"../foobar.png\" is not supported.
|
||||
# become
|
||||
# the file format of the file is not supported.
|
||||
err = {
|
||||
"steghide": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
return returncode != 0 or embedded_filename is None
|
||||
|
||||
# The passphrase is correct, we can extract the file
|
||||
# prevent path traversal
|
||||
embedded_file = output_dir / "steghide" / str(Path(embedded_filename).name)
|
||||
extracted_dir = embedded_file.parent
|
||||
extracted_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
cmd_ext: list[str] = [
|
||||
"steghide",
|
||||
"extract",
|
||||
"-sf",
|
||||
"../../" + str(image_name),
|
||||
"-xf",
|
||||
str(embedded_file.name),
|
||||
]
|
||||
if password:
|
||||
cmd_ext += ["-p", password]
|
||||
else:
|
||||
cmd_ext += ["-p", ""]
|
||||
|
||||
data = subprocess.run(
|
||||
cmd_ext,
|
||||
cwd=extracted_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
# Filter out lines that indicate success
|
||||
stdout = []
|
||||
for line in data.stderr.split("\n"):
|
||||
def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
|
||||
"""Process the stdout into a list of lines."""
|
||||
out = []
|
||||
for line in stderr.split("\n"):
|
||||
if line.startswith('wrote extracted data to "'):
|
||||
stdout.append(line)
|
||||
elif line:
|
||||
stderr += line + "\n"
|
||||
out.append(line)
|
||||
return out
|
||||
|
||||
# Zip extracted files
|
||||
zip_data = subprocess.run(
|
||||
["7z", "a", "../steghide.7z", "*"],
|
||||
cwd=extracted_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
"""Process stderr."""
|
||||
if "the file format of the file" in stderr and "not supported" in stderr:
|
||||
return "The file format of the file is not supported (JPEG or BMP only)."
|
||||
out = ""
|
||||
for line in stderr.split("\n"):
|
||||
if line and not line.startswith('wrote extracted data to "'):
|
||||
out += line + "\n"
|
||||
return out
|
||||
|
||||
stderr += zip_data.stderr
|
||||
|
||||
if len(stderr) > 0:
|
||||
err = {
|
||||
"steghide": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Remove the extracted directory
|
||||
shutil.rmtree(extracted_dir)
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"steghide": {
|
||||
"status": "ok",
|
||||
"output": stdout,
|
||||
"download": f"/download/{output_dir.name}/steghide",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"steghide": {"status": "error", "error": str(e)}})
|
||||
return None
|
||||
def analyze_steghide(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
|
||||
"""Analyze an image submission using steghide."""
|
||||
analyzer = SteghideAnalyzer(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -3,45 +3,20 @@
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Strings Analyzer for Image Submissions."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class StringsAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for strings."""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__("strings", *args)
|
||||
self.cmd = ["strings", self.img]
|
||||
|
||||
|
||||
def analyze_strings(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using strings."""
|
||||
|
||||
try:
|
||||
data = subprocess.run(
|
||||
["strings", input_img],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
if data.stderr:
|
||||
update_data(
|
||||
output_dir, {"strings": {"status": "error", "error": data.stderr}}
|
||||
)
|
||||
return
|
||||
|
||||
# strings_dir = output_dir / "strings"
|
||||
# strings_dir.mkdir(parents=True, exist_ok=True)
|
||||
# with open(strings_dir / "strings.txt", "w", encoding="utf-8") as f:
|
||||
# subprocess.run(["strings", input_img], stdout=f, check=True)
|
||||
|
||||
data_strings = data.stdout.split("\n") if data else []
|
||||
data_strings = [s for s in data_strings if s]
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"strings": {
|
||||
"status": "ok",
|
||||
"output": data_strings,
|
||||
}
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"strings": {"status": "error", "error": str(e)}})
|
||||
analyzer = StringsAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,79 +1,30 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Custom Analyzer for Image Submissions."""
|
||||
"""<toolname> Analyzer for Image Submissions."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using myanalyzer."""
|
||||
class TemplateAnalyze(SubprocessAnalyzer):
|
||||
"""Analyzer for <toolname>."""
|
||||
|
||||
image_name = input_img.name
|
||||
def __init__(self, *args):
|
||||
super().__init__("<toolname>", *args, has_archive=True)
|
||||
|
||||
# Dir generated by the tool
|
||||
extracted_dir = output_dir / "myanalyzer_output_dir"
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
if password:
|
||||
return ["<toolname>", "-p", password, "-i", self.img]
|
||||
return ["<toolname>", "-i", self.img]
|
||||
|
||||
try:
|
||||
stderr, stdout = "", ""
|
||||
|
||||
# Run myanalyzer
|
||||
data = subprocess.run(
|
||||
["myanalyzer", "-e", "../" + str(image_name)],
|
||||
cwd=output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
stderr += data.stderr
|
||||
stdout += data.stdout
|
||||
|
||||
if len(stderr) > 0: # Return error if error has been raised
|
||||
err = {
|
||||
"myanalyzer": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
zip_exist = False
|
||||
if extracted_dir.exists(): # If result file(s) exist, zip them
|
||||
subprocess.run(
|
||||
["7z", "a", "../myanalyzer.7z", "*"],
|
||||
cwd=extracted_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
zip_exist = True
|
||||
|
||||
if extracted_dir.exists(): # Remove the extracted directory
|
||||
shutil.rmtree(extracted_dir)
|
||||
|
||||
output = data.stdout.split("\n") if data else []
|
||||
output_data = {
|
||||
"myanalyzer": {
|
||||
"status": "ok",
|
||||
"output": output,
|
||||
}
|
||||
}
|
||||
|
||||
if zip_exist:
|
||||
output_data["myanalyzer"][
|
||||
"download"
|
||||
] = f"/download/{output_dir.name}/myanalyzer"
|
||||
|
||||
update_data(output_dir, output_data)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"myanalyzer": {"status": "error", "error": str(e)}})
|
||||
return None
|
||||
def analyze_template(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
|
||||
"""Analyze an image submission using <toolname>."""
|
||||
analyzer = TemplateAnalyze(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,43 +1,31 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Zsteg Analyzer for Image Submissions."""
|
||||
"""Strings Analyzer for Image Submissions."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class ZstegAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for zsteg."""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__("zsteg", *args, **kwargs)
|
||||
self.cmd = ["zsteg", self.img]
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
return bool(stderr) or "PNG::NotSupported" in stdout[:100]
|
||||
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
"""Process the stderr."""
|
||||
if "PNG::NotSupported" in stdout[:100]:
|
||||
return "The file format of the file is not supported (PNG only)."
|
||||
return stderr
|
||||
|
||||
|
||||
def analyze_zsteg(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using zsteg."""
|
||||
|
||||
try:
|
||||
data = subprocess.run(
|
||||
["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
|
||||
update_data(output_dir, {"zsteg": {"status": "error", "error": error_msg}})
|
||||
return
|
||||
|
||||
# zsteg_dir = output_dir / "zsteg"
|
||||
# zsteg_dir.mkdir(parents=True, exist_ok=True)
|
||||
# with open(zsteg_dir / "zsteg.txt", "w", encoding="utf-8") as f:
|
||||
# subprocess.run(["zsteg", input_img], stdout=f, check=True)
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"zsteg": {
|
||||
"status": "ok",
|
||||
"output": data.stdout.split("\n") if data else [],
|
||||
}
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"zsteg": {"status": "error", "error": str(e)}})
|
||||
analyzer = ZstegAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
Reference in New Issue
Block a user