Black

aperisolve/analyzers/pngcheck.py

@@ -24,6 +24,7 @@ class PngcheckAnalyzer(SubprocessAnalyzer):
             return "The file format of the file is not supported (PNG or JNG only)."
         return stdout
 
+
 def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
     """Analyze an image submission using pngcheck."""
     analyzer = PngcheckAnalyzer(input_img, output_dir)

aperisolve/analyzers/steghide.py

@@ -49,8 +49,8 @@ class SteghideAnalyzer(SubprocessAnalyzer):
     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 (stdout + '\n' + stderr).split("\n"):
+        for line in (stdout + "\n" + stderr).split("\n"):
-            if 'wrote extracted data to' in line:
+            if "wrote extracted data to" in line:
                 out.append(line)
         return out
 

aperisolve/app.py

@@ -28,9 +28,7 @@ app: Flask = Flask(__name__)
 app.json.sort_keys = False  # type: ignore
 app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URI")
 app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
-app.config["MAX_CONTENT_LENGTH"] = int(
+app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024))
-    os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024)
-)
 RESULT_FOLDER.mkdir(parents=True, exist_ok=True)
 
 db.init_app(app)
@@ -112,10 +110,7 @@ def upload_image() -> tuple[Response, int]:
     if image.filename is None or image.filename == "":
         return jsonify({"error": "No image provided"}), 400
 
-    if (
+    if "." not in image.filename or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS:
-        "." not in image.filename
-        or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS
-    ):
         return jsonify({"error": "Unsupported file type"}), 400
 
     # Compute hashes and prepare submission
@@ -230,9 +225,7 @@ def get_result(hash_val: str) -> tuple[Response, int]:
     submission = Submission.query.get_or_404(hash_val)
     image = Image.query.get_or_404(submission.image_hash)  # type: ignore
 
-    result_path = (
+    result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
-        RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
-    )
 
     if not result_path.exists():
         return jsonify({"error": "Results not ready yet..."}), 425
@@ -261,9 +254,7 @@ def download_output(hash_val: str, tool: str) -> Response:
 
 @app.route("/image/<img_name>")
 @app.route("/image/<hash_val>/<img_name>")
-def get_image(
+def get_image(hash_val: Optional[str] = None, img_name: Optional[str] = None) -> Response:
-    hash_val: Optional[str] = None, img_name: Optional[str] = None
-) -> Response:
     """
     Download the image with a specific name (usually ones generated with
     decomposer) for a given submission hash.
@@ -283,9 +274,7 @@ def get_image(
         image = Image.query.filter_by(hash=hash_val).first_or_404()
         output_file = Path(image.file)  # type: ignore
 
-    if (not output_file.exists()) or (
+    if (not output_file.exists()) or (output_file.suffix.lower() not in IMAGE_EXTENSIONS):
-        output_file.suffix.lower() not in IMAGE_EXTENSIONS
-    ):
         return abort(404, description="Image not found or unsupported format")
 
     return send_file(output_file, as_attachment=True)

aperisolve/cleanup.py

@@ -49,9 +49,7 @@ def cleanup_old_entries() -> None:
             db.session.delete(submission)  # pylint: disable=no-member
         elif submission.status == "done":  # type: ignore
             # Search for buggy results, delete
-            result_path = (
+            result_path = RESULT_FOLDER / str(submission.image_hash) / str(submission.hash)
-                RESULT_FOLDER / str(submission.image_hash) / str(submission.hash)
-            )
             result_file = result_path / "results.json"
             if not result_file.exists():
                 db.session.delete(submission)  # pylint: disable=no-member

aperisolve/models.py

@@ -97,9 +97,7 @@ def fill_ihdr_db() -> None:
             if exists:  # Skip if already present
                 continue
 
-            db.session.add(  # pylint: disable=no-member
+            db.session.add(IHDR(packed=p.packed, crc=p.crc))  # pylint: disable=no-member
-                IHDR(packed=p.packed, crc=p.crc)
-            )
             count += 1
 
             if count % 5000 == 0:

aperisolve/sentry.py

@@ -14,9 +14,7 @@ from sentry_sdk.integrations.threading import ThreadingIntegration
 SENTRY_DSN = os.environ.get("SENTRY_DSN")
 SENTRY_ENVIRONMENT = os.environ.get("SENTRY_ENVIRONMENT", "development")
 SENTRY_TRACES_SAMPLE_RATE = float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", "0.1"))
-SENTRY_PROFILES_SAMPLE_RATE = float(
+SENTRY_PROFILES_SAMPLE_RATE = float(os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1"))
-    os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1")
-)
 SENTRY_RELEASE = os.environ.get("SENTRY_RELEASE", "1.0.0")