Zeecka
@@ -0,0 +1,26 @@
|
||||
# Application configuration
|
||||
MAX_CONTENT_LENGTH=1048576
|
||||
MAX_PENDING_TIME=600
|
||||
MAX_STORE_TIME=259200
|
||||
CLEAR_AT_RESTART=1
|
||||
SKIP_IHDR_FILL=0
|
||||
|
||||
# Flask configuration
|
||||
FLASK_DEBUG=0
|
||||
FLASK_ENV=development
|
||||
|
||||
# Redis configuration
|
||||
RQ_DASHBOARD_REDIS_URL=redis://redis:6379/0
|
||||
|
||||
# Database configuration
|
||||
POSTGRES_DB=aperisolve
|
||||
POSTGRES_USER=aperiuser
|
||||
POSTGRES_PASSWORD=aperipass
|
||||
DB_URI=postgresql://aperiuser:aperipass@postgres:5432/aperisolve
|
||||
|
||||
# Sentry configuration
|
||||
SENTRY_DSN=https://your-dsn@sentry.io/project-id
|
||||
SENTRY_TRACES_SAMPLE_RATE=0.1
|
||||
SENTRY_PROFILES_SAMPLE_RATE=0.1
|
||||
SENTRY_RELEASE=1.0.0
|
||||
SENTRY_ENVIRONMENT=development
|
||||
@@ -0,0 +1,61 @@
|
||||
name: Deploy to production
|
||||
|
||||
on:
|
||||
workflow_run:
|
||||
workflows:
|
||||
- Docker Image CI
|
||||
types:
|
||||
- completed
|
||||
|
||||
jobs:
|
||||
deploy:
|
||||
# ne déploie que si le build a réussi
|
||||
if: ${{ github.event.workflow_run.conclusion == 'success' }}
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
environment: production
|
||||
|
||||
steps:
|
||||
- name: Deploy to production server
|
||||
uses: appleboy/ssh-action@v1.0.3
|
||||
with:
|
||||
host: ${{ secrets.PROD_SSH_HOST }}
|
||||
username: ${{ secrets.PROD_SSH_USER }}
|
||||
key: ${{ secrets.PROD_SSH_KEY }}
|
||||
script: |
|
||||
set -e
|
||||
cd /root/AperiSolve
|
||||
|
||||
# update code
|
||||
git pull
|
||||
|
||||
# create .env
|
||||
cat <<EOF > .env
|
||||
MAX_CONTENT_LENGTH=${{ vars.MAX_CONTENT_LENGTH }}
|
||||
MAX_PENDING_TIME=${{ vars.MAX_PENDING_TIME }}
|
||||
MAX_STORE_TIME=${{ vars.MAX_STORE_TIME }}
|
||||
CLEAR_AT_RESTART=${{ vars.CLEAR_AT_RESTART }}
|
||||
SKIP_IHDR_FILL=${{ vars.SKIP_IHDR_FILL }}
|
||||
|
||||
FLASK_DEBUG=${{ vars.FLASK_DEBUG }}
|
||||
FLASK_ENV=${{ vars.FLASK_ENV }}
|
||||
|
||||
RQ_DASHBOARD_REDIS_URL=${{ secrets.REDIS_URL }}
|
||||
|
||||
POSTGRES_USER=${{ secrets.POSTGRES_USER }}
|
||||
POSTGRES_PASSWORD=${{ secrets.POSTGRES_PASSWORD }}
|
||||
POSTGRES_DB=${{ secrets.POSTGRES_DB }}
|
||||
DB_URI=${{ secrets.DATABASE_URL }}
|
||||
|
||||
SENTRY_DSN=${{ secrets.SENTRY_DSN }}
|
||||
SENTRY_TRACES_SAMPLE_RATE=${{ secrets.SENTRY_TRACES_SAMPLE_RATE }}
|
||||
SENTRY_PROFILES_SAMPLE_RATE=${{ secrets.SENTRY_PROFILES_SAMPLE_RATE }}
|
||||
SENTRY_RELEASE=${{ secrets.SENTRY_RELEASE }}
|
||||
SENTRY_ENVIRONMENT=${{ secrets.SENTRY_ENVIRONMENT }}
|
||||
EOF
|
||||
|
||||
# Pull images and deploy
|
||||
docker compose pull
|
||||
docker compose down
|
||||
docker compose up -d
|
||||
docker image prune -f
|
||||
@@ -34,6 +34,7 @@ jobs:
|
||||
uses: wearerequired/lint-action@v2
|
||||
with:
|
||||
black: true
|
||||
black_args: "--line-length=100"
|
||||
flake8: true
|
||||
flake8_args: "--ignore=E203,E501,W503"
|
||||
pylint: true
|
||||
|
||||
+2
-1
@@ -1,5 +1,6 @@
|
||||
# Python
|
||||
.env
|
||||
__pycache__/
|
||||
.mypy_cache/
|
||||
results/
|
||||
.env
|
||||
.vscode
|
||||
|
||||
@@ -9,6 +9,12 @@ repos:
|
||||
rev: 25.11.0
|
||||
hooks:
|
||||
- id: black
|
||||
args: ['--line-length', '100']
|
||||
- repo: https://github.com/pycqa/isort
|
||||
rev: 5.11.2
|
||||
hooks:
|
||||
- id: isort
|
||||
args: ['--profile', 'black']
|
||||
- repo: https://github.com/PyCQA/flake8
|
||||
rev: 7.3.0
|
||||
hooks:
|
||||
|
||||
+210
@@ -0,0 +1,210 @@
|
||||
u# Contributing to Aperi'Solve
|
||||
|
||||
Thank you for your interest in contributing! We welcome all contributions, whether it's bug fixes, new features, or documentation improvements.
|
||||
|
||||
## Getting Started
|
||||
|
||||
> [!TIP]
|
||||
> New to the project? Adding a new analyzer is a great first contribution! Check out our [Adding a New Analyzer Guide](docs/adding-analyzer.md).
|
||||
|
||||
### Quick Steps
|
||||
|
||||
1. **Fork the repository**
|
||||
2. **Create a feature branch** (`git checkout -b feature/amazing-feature`)
|
||||
3. **Make your changes**
|
||||
4. **Test your changes**
|
||||
5. **Commit with clear messages** (`git commit -m 'Add amazing feature'`)
|
||||
6. **Push to your fork** (`git push origin feature/amazing-feature`)
|
||||
7. **Open a Pull Request** describing your changes
|
||||
|
||||
## What Can You Contribute?
|
||||
|
||||
### 🔧 Adding New Analyzers
|
||||
|
||||
Want to add support for a new steganography or forensics tool? We'd love that!
|
||||
|
||||
> [!NOTE]
|
||||
> Check out our detailed tutorial: [Adding a New Analyzer](docs/adding-analyzer.md)
|
||||
|
||||
**Quick checklist for new analyzers:**
|
||||
- [ ] Create analyzer file in `analyzers/`
|
||||
- [ ] Extend `SubprocessAnalyzer` base class
|
||||
- [ ] Register in `workers.py`
|
||||
- [ ] Add tool to Docker setup if needed
|
||||
- [ ] Test with sample images
|
||||
- [ ] Update documentation
|
||||
|
||||
### 🐛 Bug Fixes
|
||||
|
||||
Found a bug? Please:
|
||||
- Check if it's already reported in [Issues](../../issues)
|
||||
- If not, open a new issue with reproduction steps
|
||||
- Feel free to submit a PR with the fix!
|
||||
|
||||
### 📚 Documentation
|
||||
|
||||
Help us improve:
|
||||
- Fix typos or unclear explanations
|
||||
- Add examples and use cases
|
||||
- Improve installation instructions
|
||||
- Translate documentation
|
||||
|
||||
### 🎨 UI/UX Improvements
|
||||
|
||||
Contributions to the web interface are welcome:
|
||||
- Better error messages
|
||||
- Improved result presentation
|
||||
- Mobile responsiveness
|
||||
- Accessibility improvements
|
||||
|
||||
## Code Style & Quality
|
||||
|
||||
> [!IMPORTANT]
|
||||
> Please follow the code style and run linters before submitting.
|
||||
|
||||
The project adheres to:
|
||||
|
||||
- **Black** (line length 100)
|
||||
- **Isort** (profile black)
|
||||
- **Flake8** (ignoring E203, E501, W503)
|
||||
- **Pylint** (ignoring W0718, R0903, R0801)
|
||||
- **Mypy** (ignoring unused-awaitable)
|
||||
|
||||
See [.pre-commit-config.yaml](.pre-commit-config.yaml) for exact configuration.
|
||||
|
||||
### Running Linters Locally
|
||||
|
||||
> [!TIP]
|
||||
> Install pre-commit hooks to automatically check your code before commits:
|
||||
> ```bash
|
||||
> pip install pre-commit
|
||||
> pre-commit install
|
||||
> ```
|
||||
|
||||
**Manual checks:**
|
||||
```bash
|
||||
# Format code
|
||||
black . --line-length 100
|
||||
isort . --profile black
|
||||
|
||||
# Check for issues
|
||||
flake8 .
|
||||
pylint **/*.py
|
||||
mypy .
|
||||
```
|
||||
|
||||
> [!NOTE]
|
||||
> CI will automatically run these checks on every PR. Make sure they pass!
|
||||
|
||||
## Pull Request Guidelines
|
||||
|
||||
Use clear, descriptive titles:
|
||||
- ✅ `Add stegdetect analyzer`
|
||||
- ✅ `Fix binwalk extraction error handling`
|
||||
- ✅ `Update documentation for password-protected tools`
|
||||
- ❌ `Update code`
|
||||
- ❌ `Fix bug`
|
||||
|
||||
|
||||
### Commit Messages
|
||||
|
||||
Write clear, concise commit messages:
|
||||
```bash
|
||||
# Good
|
||||
git commit -m "Add zsteg analyzer for PNG steganography detection"
|
||||
git commit -m "Fix foremost error detection when no files extracted"
|
||||
|
||||
# Not so good
|
||||
git commit -m "updates"
|
||||
git commit -m "fix"
|
||||
```
|
||||
|
||||
## Adding New Dependencies
|
||||
|
||||
> [!CAUTION]
|
||||
> Adding new dependencies requires careful consideration.
|
||||
|
||||
If your contribution needs new dependencies:
|
||||
|
||||
1. **Python packages**: Add to `requirements.txt`
|
||||
2. **System tools**: Add to `Dockerfile`
|
||||
3. **Explain why** in your PR description
|
||||
4. **Keep dependencies minimal** - avoid adding large libraries for small features
|
||||
|
||||
|
||||
## Useful commands
|
||||
|
||||
```bash
|
||||
# development environment (hot reload and local volumes)
|
||||
docker compose -f compose.dev.yml up --build
|
||||
|
||||
# Stop and remove containers, networks and volumes (results are stored in a volume)
|
||||
docker compose down -v
|
||||
|
||||
# Enter web container shell
|
||||
docker exec -it aperisolve-web bash
|
||||
|
||||
# Enter Postgres shell (from host)
|
||||
docker exec -it postgres psql -U aperiuser -d aperisolve
|
||||
|
||||
# Backup all uploaded files
|
||||
docker cp -r aperisolve-web:/aperisolve/results /path/to/backup/location
|
||||
|
||||
# Backup a single uploaded file
|
||||
docker cp aperisolve-web:/aperisolve/results/filename.ext /path/to/backup/filename.ext
|
||||
```
|
||||
|
||||
> [!WARNING]
|
||||
> If switching between dev and production compose files, remove the `results` directory or mounted volume to avoid conflicts:
|
||||
> ```bash
|
||||
> rm -rf aperisolve/results
|
||||
> ```
|
||||
|
||||
## Docker and Tool Installation
|
||||
|
||||
When adding a new analyzer that requires a new tool:
|
||||
|
||||
```dockerfile
|
||||
# In Dockerfile
|
||||
RUN apt-get update && apt-get install -y \
|
||||
your-new-tool \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
```
|
||||
|
||||
Or for tools requiring compilation:
|
||||
```dockerfile
|
||||
RUN git clone https://github.com/tool/repo.git /tmp/tool && \
|
||||
cd /tmp/tool && \
|
||||
make && make install && \
|
||||
rm -rf /tmp/tool
|
||||
```
|
||||
|
||||
> [!NOTE]
|
||||
> Test the Docker build locally before submitting!
|
||||
|
||||
## Code Review Process
|
||||
|
||||
1. **Automated checks** run first (linters, tests)
|
||||
2. **Maintainer review** - we'll provide feedback
|
||||
3. **Address feedback** - update your PR as needed
|
||||
4. **Approval & merge** - once everything looks good!
|
||||
|
||||
> [!TIP]
|
||||
> Don't worry if you need to make changes - it's a normal part of the process!
|
||||
|
||||
## Need Help?
|
||||
|
||||
- 💬 **Questions?** Open a [Discussion](../../discussions)
|
||||
- 🐛 **Found a bug?** Open an [Issue](../../issues)
|
||||
- 💡 **Ideas?** We'd love to hear them in Discussions!
|
||||
|
||||
## Recognition
|
||||
|
||||
> [!NOTE]
|
||||
> All contributors will be recognized in our [Contributors](../../graphs/contributors) page and in release notes!
|
||||
|
||||
---
|
||||
|
||||
**Thank you for contributing! 🎉**
|
||||
|
||||
Your efforts help make this tool better for the entire security and CTF community!
|
||||
+9
-6
@@ -5,7 +5,7 @@ WORKDIR /
|
||||
# Install build tools and dependencies
|
||||
RUN apt-get update && apt-get install -y \
|
||||
wget \
|
||||
default-jre \
|
||||
default-jdk \
|
||||
ruby \
|
||||
zip \
|
||||
7zip
|
||||
@@ -26,17 +26,20 @@ RUN wget https://github.com/syvaidya/openstego/releases/download/openstego-0.8.6
|
||||
dpkg -i /tmp/openstego.deb && \
|
||||
rm /tmp/openstego.deb
|
||||
|
||||
# Clean up apt cache and remove useless packages
|
||||
RUN apt-get remove -y wget && \
|
||||
apt-get clean && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
# Copy application
|
||||
COPY aperisolve/ /aperisolve/
|
||||
|
||||
RUN mkdir -p /data
|
||||
|
||||
# Install Python dependencies
|
||||
RUN pip install --no-cache-dir -r /aperisolve/requirements.txt
|
||||
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
# Auto clean after build
|
||||
RUN apt-get remove -y wget && \
|
||||
apt-get autoremove -y && \
|
||||
apt-get clean && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "--access-logfile", "-", "--error-logfile", "-", "--log-level", "info", "--capture-output", "aperisolve.wsgi:application"]
|
||||
|
||||
@@ -24,31 +24,20 @@ Support Aperi'Solve:
|
||||
|
||||
Aperi'Solve is an open-source steganalysis web platform that performs automated analysis on images to detect and extract hidden data using common steganography tools and techniques.
|
||||
|
||||
Table of contents
|
||||
- [Key features](#key-features)
|
||||
- [Quick start (Docker)](#quick-start-docker)
|
||||
- [Development](#development)
|
||||
- [Adding a new analyzer](#adding-a-new-analyzer)
|
||||
- [Configuration & environment variables](#configuration--environment-variables)
|
||||
- [Architecture](#architecture)
|
||||
- [Troubleshooting & tips](#troubleshooting--tips)
|
||||
- [Roadmap](#roadmap)
|
||||
- [Contributing](#contributing)
|
||||
|
||||
## Key features
|
||||
|
||||
- Visualize each bit layer (LSB and other layers) per image channel (R/G/B/Alpha).
|
||||
- Browse and download each bit-layer image.
|
||||
- Integrates and displays outputs from:
|
||||
- [zsteg](https://github.com/zed-0xff/zsteg) (LSB text/data extraction)
|
||||
- [steghide](https://steghide.sourceforge.net/) (extraction with password)
|
||||
- [outguess](https://www.rbcafe.com/software/outguess/) (extraction with password)
|
||||
- [openstego](https://www.openstego.com/) (extraction with password)
|
||||
- [exiftool](https://exiftool.org/) (metadata and geolocation)
|
||||
- [binwalk](https://github.com/ReFirmLabs/binwalk) (embedded archives)
|
||||
- [exiftool](https://exiftool.org/) (metadata and geolocation)
|
||||
- [foremost](https://foremost.sourceforge.net/) (carved files)
|
||||
- [openstego](https://www.openstego.com/) (extraction with password)
|
||||
- [outguess](https://www.rbcafe.com/software/outguess/) (extraction with password)
|
||||
- [pngcheck](https://www.libpng.org/pub/png/apps/pngcheck.html)
|
||||
- [steghide](https://steghide.sourceforge.net/) (extraction with password)
|
||||
- [strings](https://pubs.opengroup.org/onlinepubs/9799919799/utilities/strings.html)
|
||||
- [zsteg](https://github.com/zed-0xff/zsteg) (LSB text/data extraction)
|
||||
- Worker queue architecture for offloading heavy/slow analyzers (Redis + background workers).
|
||||
- Results stored for later browsing and download.
|
||||
|
||||
@@ -56,100 +45,16 @@ Table of contents
|
||||
|
||||
In case you want to host your own version of https://www.aperisolve.com/.
|
||||
|
||||
Recommended: Docker + Docker Compose.
|
||||
> Required: Docker + Docker Compose.
|
||||
|
||||
Clone and start the full stack (production-like):
|
||||
```bash
|
||||
git clone https://github.com/Zeecka/AperiSolve.git
|
||||
cd AperiSolve
|
||||
cp .env.example .env
|
||||
docker compose up -d
|
||||
```
|
||||
|
||||
Default: http://localhost:5000/
|
||||
|
||||
## Development
|
||||
Development compose:
|
||||
```bash
|
||||
# development environment (hot reload and local volumes)
|
||||
docker compose -f compose.dev.yml up --build
|
||||
```
|
||||
|
||||
Useful commands:
|
||||
```bash
|
||||
# Stop and remove containers, networks and volumes (results are stored in a volume)
|
||||
docker compose down -v
|
||||
|
||||
# Enter web container shell
|
||||
docker exec -it aperisolve-web bash
|
||||
|
||||
# Enter Postgres shell (from host)
|
||||
docker exec -it postgres psql -U aperiuser -d aperisolve
|
||||
```
|
||||
|
||||
> [!WARNING]
|
||||
> If switching between dev and production compose files, remove the `results` directory or mounted volume to avoid conflicts:
|
||||
> ```bash
|
||||
> rm -rf aperisolve/results
|
||||
> ```
|
||||
|
||||
### Adding a new analyzer
|
||||
|
||||
Adding a custom analyzer is straightforward:
|
||||
|
||||
1. Create your analyzer file:
|
||||
- Copy `aperisolve/analyzers/template_analyzer.py` -> `aperisolve/analyzers/myanalyzer.py`
|
||||
- Implement function signature similar to:
|
||||
```python
|
||||
# aperisolve/analyzers/myanalyzer.py
|
||||
def analyze_myanalyzer(image_path: str, results_dir: str) -> dict:
|
||||
"""
|
||||
Perform analysis on image located at image_path.
|
||||
Produce outputs in results_dir and return a result dict (json-serializable).
|
||||
"""
|
||||
# your analysis logic here
|
||||
return {"name": "myanalyzer", "status": "ok", "outputs": [...]}
|
||||
```
|
||||
|
||||
2. Register the analyzer in the worker pipeline:
|
||||
- Edit `aperisolve/workers.py`:
|
||||
```python
|
||||
# import
|
||||
from .analyzers.myanalyzer import analyze_myanalyzer
|
||||
|
||||
# add to analyzers list (preserve the order)
|
||||
analyzers = [
|
||||
analyze_zsteg,
|
||||
analyze_steghide,
|
||||
# ...
|
||||
analyze_myanalyzer,
|
||||
]
|
||||
```
|
||||
|
||||
3. Add your analyzer to the UI order:
|
||||
- Edit `aperisolve/static/js/aperisolve.js` and append `myanalyzer` to `TOOL_ORDER` so it appears in the frontend.
|
||||
|
||||
4. Test locally: run the worker and submit jobs to ensure outputs are produced and displayed.
|
||||
|
||||
> [!TIP]
|
||||
> - Keep analyzers idempotent and write outputs to the provided `results_dir`.
|
||||
> - Return structured JSON so the frontend can render links/downloads automatically.
|
||||
|
||||
## Configuration & environment variables
|
||||
|
||||
Typical services:
|
||||
- Web app (Flask)
|
||||
- Workers (Python)
|
||||
- Redis (RQ)
|
||||
- PostgreSQL
|
||||
|
||||
Main environment variables (examples):
|
||||
- DATABASE_URL=postgresql://aperiuser:password@postgres:5432/aperisolve
|
||||
- REDIS_URL=redis://redis:6379/0
|
||||
- SECRET_KEY=change_me
|
||||
- FLASK_ENV=production/development
|
||||
|
||||
> [!NOTE]
|
||||
> If using Docker Compose, defaults are set in the compose files. For production deployments, set secure secrets via your orchestrator or environment.
|
||||
Then browse url: http://localhost:5000/
|
||||
|
||||
## Architecture
|
||||
|
||||
@@ -160,44 +65,13 @@ Main environment variables (examples):
|
||||
|
||||
This separation keeps heavy tools (binwalk, foremost, zsteg, etc.) isolated and avoids blocking the web worker.
|
||||
|
||||
## Troubleshooting & tips
|
||||
|
||||
- If analyzers don't produce output, check worker logs:
|
||||
```bash
|
||||
docker compose logs -f aperisolve-worker
|
||||
```
|
||||
- To force re-analysis, remove results for the image (both file and in database) and re-submit the job.
|
||||
- Ensure system packages needed by native tools (binwalk, foremost) are available in the analyzer containers or host image.
|
||||
|
||||
## Roadmap
|
||||
|
||||
- [ ] **[Bug]** Error 500 when file already exist on disc
|
||||
- [ ] **[Bug]** Multiple bug on buggy Jpeg filebugSomething isn't working
|
||||
- [ ] **[Bug]** Binwalk Extractor errorbugSomething isn't working
|
||||
- [ ] **[Bug]** Grayscale on a RGB image ?bugSomething isn't working
|
||||
- [ ] **[Feature]** Implement healthcheck and auto reboot
|
||||
- [ ] **[Feature]** Implement Sentry for better error handling
|
||||
- [ ] **[Feature]** Zsteg: full extraction (--all) and download of discovered files (mp3, etc.)
|
||||
- [ ] **[Feature]** Mobile-friendly UI / test on mobile
|
||||
- [ ] **[Feature]** i18n (internationalization)
|
||||
- [ ] **[Feature]** Rootless / unprivileged analyzers
|
||||
- [ ] **[Feature]** Improve analyzer sandboxing (e.g., per-analyzer containers)
|
||||
See [Issues](../../issues).
|
||||
|
||||
## Contributing
|
||||
|
||||
Contributions are welcome! Please:
|
||||
1. Fork the repository
|
||||
2. Create a feature branch
|
||||
3. Open a pull request describing your change
|
||||
|
||||
> [!IMPORTANT]
|
||||
> Please follow the code style and run linters before submitting. The project adheres to:
|
||||
> - Black
|
||||
> - Flake8 (ignoring E203, E501, W503)
|
||||
> - Pylint (ignoring W0718, R0903, R0801)
|
||||
> - Mypy (ignoring unused-awaitable)
|
||||
|
||||
CI will run these checks on each PR.
|
||||
See [CONTRIBUTING.md](CONTRIBUTING.md).
|
||||
|
||||
## Security
|
||||
|
||||
@@ -206,5 +80,13 @@ CI will run these checks on each PR.
|
||||
|
||||
## Credits
|
||||
|
||||
Acknowledgements:
|
||||
- Thanks to contributors and the open-source community for the tools integrated ([zsteg](https://github.com/zed-0xff/zsteg), [steghide](https://steghide.sourceforge.net/), [outguess](https://www.rbcafe.com/software/outguess/), [openstego](https://www.openstego.com/), [exiftool](https://exiftool.org/), [binwalk](https://github.com/ReFirmLabs/binwalk), [foremost](https://foremost.sourceforge.net/), [pngcheck](https://www.libpng.org/pub/png/apps/pngcheck.html), [strings](https://pubs.opengroup.org/onlinepubs/9799919799/utilities/strings.html), ...).
|
||||
Thanks to donors:
|
||||
- [Philip Zimmermann](https://github.com/Philip-Zimmermann)
|
||||
|
||||
Thanks to contributors:
|
||||
- [Zeecka](https://www.zeecka.fr/) - **(author)**
|
||||
- [aradhyacp](https://github.com/aradhyacp)
|
||||
- [Philip Zimmermann](https://github.com/Philip-Zimmermann)
|
||||
|
||||
Thanks to the open-source community:
|
||||
[binwalk](https://github.com/ReFirmLabs/binwalk), [exiftool](https://exiftool.org/), [foremost](https://foremost.sourceforge.net/), [openstego](https://www.openstego.com/), [outguess](https://www.rbcafe.com/software/outguess/), [pngcheck](https://www.libpng.org/pub/png/apps/pngcheck.html), [steghide](https://steghide.sourceforge.net/), [strings](https://pubs.opengroup.org/onlinepubs/9799919799/utilities/strings.html), [zsteg](https://github.com/zed-0xff/zsteg), ...
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,W0613,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Base Analyzer class for image analysis tools."""
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
from abc import ABC
|
||||
from pathlib import Path
|
||||
from shutil import rmtree
|
||||
from typing import Any, Optional, overload
|
||||
|
||||
_thread_lock = threading.Lock()
|
||||
MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default
|
||||
|
||||
|
||||
class SubprocessAnalyzer(ABC):
|
||||
"""Analyzer that runs a subprocess command."""
|
||||
|
||||
name: str
|
||||
input_img: Path
|
||||
output_dir: Path
|
||||
has_archive: bool
|
||||
cmd: list[str] | None = None
|
||||
img: str
|
||||
make_folder: bool = True
|
||||
|
||||
def __init__(
|
||||
self, name: str, input_img: Path, output_dir: Path, has_archive: bool = False
|
||||
) -> None:
|
||||
self.name = name
|
||||
self.input_img = input_img
|
||||
self.img = f"../{self.input_img.name}"
|
||||
self.output_dir = output_dir
|
||||
self.has_archive = has_archive
|
||||
|
||||
def run_command(
|
||||
self, cmd: list[str], cwd: Optional[Path] = None
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
"""Run a subprocess command."""
|
||||
return subprocess.run(
|
||||
cmd,
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
def generate_archive(self, output_dir: Path, extracted_dir: Path) -> str:
|
||||
"""Zip the extracted files and remove the directory."""
|
||||
zip_data = self.run_command(["7z", "a", f"../{self.name}.7z", "*"], cwd=extracted_dir)
|
||||
rmtree(extracted_dir)
|
||||
return zip_data.stderr
|
||||
|
||||
def update_result(self, result: dict[str, Any]) -> None:
|
||||
"""Thread-safe and process-safe JSON update using lock file and atomic replace."""
|
||||
json_file = self.output_dir / "results.json"
|
||||
new_data = {self.name: result}
|
||||
lock_file = json_file.with_suffix(json_file.suffix + ".lock")
|
||||
tmp_file = json_file.with_suffix(json_file.suffix + ".tmp")
|
||||
|
||||
json_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with _thread_lock: # synchronizes across threads
|
||||
with open(lock_file, "w", encoding="utf-8") as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX) # synchronizes across processes
|
||||
|
||||
try:
|
||||
# Read existing JSON
|
||||
data: dict[Any, Any] = {}
|
||||
if json_file.exists():
|
||||
try:
|
||||
with open(json_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except json.JSONDecodeError:
|
||||
data = {}
|
||||
|
||||
# Update with new data
|
||||
data.update(new_data)
|
||||
|
||||
# Write safely to a temp file
|
||||
with open(tmp_file, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, sort_keys=False)
|
||||
|
||||
os.replace(tmp_file, json_file) # ensures file write is atomic
|
||||
finally:
|
||||
fcntl.flock(lock, fcntl.LOCK_UN)
|
||||
|
||||
def get_extracted_dir(self) -> Path:
|
||||
"""Get the extracted directory path. Can be overridden but work as it is."""
|
||||
return self.output_dir / self.name
|
||||
|
||||
@overload
|
||||
def build_cmd(self) -> list[str]: ...
|
||||
|
||||
@overload
|
||||
def build_cmd(self, password: str) -> list[str]: ...
|
||||
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
"""Build the command to run. Can be overridden."""
|
||||
if self.cmd is None:
|
||||
raise NotImplementedError("cmd must be set or build_cmd overridden")
|
||||
return self.cmd
|
||||
|
||||
def get_results(self, password: Optional[str] = None) -> dict[str, Any]:
|
||||
"""Get results of command before returning."""
|
||||
result: dict[str, Any]
|
||||
extracted_dir = None
|
||||
if self.has_archive:
|
||||
extracted_dir = self.get_extracted_dir()
|
||||
if self.make_folder:
|
||||
extracted_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
cmd: list[str]
|
||||
if password:
|
||||
cmd = list(map(str, self.build_cmd(password)))
|
||||
else:
|
||||
cmd = list(map(str, self.build_cmd()))
|
||||
data = self.run_command(cmd, cwd=self.output_dir)
|
||||
returncode = data.returncode
|
||||
stderr = data.stderr
|
||||
stdout = data.stdout
|
||||
|
||||
zip_exist = False
|
||||
if extracted_dir and extracted_dir.exists() and any(extracted_dir.iterdir()):
|
||||
self.generate_archive(self.output_dir, extracted_dir) # keep archive
|
||||
zip_exist = True
|
||||
|
||||
if self.is_error(returncode, stdout, stderr, zip_exist):
|
||||
result = {
|
||||
"status": "error",
|
||||
}
|
||||
result["error"] = self.process_error(stdout, stderr)
|
||||
else:
|
||||
result = {
|
||||
"status": "ok",
|
||||
"output": self.process_output(stdout, stderr),
|
||||
}
|
||||
note: Optional[str] = self.process_note(stdout, stderr)
|
||||
if note:
|
||||
result["note"] = note
|
||||
if zip_exist:
|
||||
result["download"] = f"/download/{self.output_dir.name}/{self.name}"
|
||||
return result
|
||||
|
||||
@overload
|
||||
def analyze(self) -> None: ...
|
||||
|
||||
@overload
|
||||
def analyze(self, password: str) -> None: ...
|
||||
|
||||
def analyze(self, password: Optional[str] = None) -> None:
|
||||
"""Run the subprocess command and handle results."""
|
||||
result: dict[str, Any]
|
||||
try:
|
||||
result = self.get_results(password)
|
||||
self.update_result(result)
|
||||
except Exception as e:
|
||||
self.update_result({"status": "error", "error": str(e)})
|
||||
raise
|
||||
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
"""Check if the result is an error."""
|
||||
return len(stderr) > 0
|
||||
|
||||
def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
|
||||
"""Process the stdout into a list of lines."""
|
||||
return [line for line in stdout.split("\n") if line] if stdout else []
|
||||
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
"""Process the stderr."""
|
||||
return stderr
|
||||
|
||||
def process_note(self, stdout: str, stderr: str) -> Optional[str]:
|
||||
"""Process the stdout for informational purposes."""
|
||||
return None
|
||||
@@ -1,70 +1,29 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Binwalk 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 BinwalkAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for binwalk."""
|
||||
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("binwalk", input_img, output_dir, 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
|
||||
|
||||
|
||||
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", "-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)
|
||||
|
||||
if len(stderr) > 0:
|
||||
err = {
|
||||
"binwalk": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
output_data = {
|
||||
"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()
|
||||
|
||||
@@ -1,16 +1,26 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Bits Decomposer Analyzer for Image Submissions."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
from .utils import update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
|
||||
class DecomposerAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for bits decomposer."""
|
||||
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("decomposer", input_img, output_dir)
|
||||
|
||||
def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
|
||||
"""Analyze an image submission using bits decomposition."""
|
||||
img = Image.open(input_img)
|
||||
img = Image.open(self.input_img)
|
||||
converted = False
|
||||
|
||||
if img.mode == "P":
|
||||
@@ -41,8 +51,8 @@ def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
|
||||
rgb_image = np.stack(rgb_planes, axis=-1).astype(np.uint8)
|
||||
rgb_img = Image.fromarray(rgb_image, mode="RGB")
|
||||
img_name = f"superimposed_bit_{bit}.png"
|
||||
out_path = output_dir / img_name
|
||||
dl_path = Path(output_dir.name) / img_name
|
||||
out_path = self.output_dir / img_name
|
||||
dl_path = Path(self.output_dir.name) / img_name
|
||||
superimposed_json.append("/image/" + str(dl_path))
|
||||
rgb_img.save(out_path)
|
||||
|
||||
@@ -58,21 +68,25 @@ def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
|
||||
bit_plane = ((channel_data & bit_mask) >> bit) * 255
|
||||
bit_img = Image.fromarray(bit_plane.astype(np.uint8), mode="L")
|
||||
img_name = f"{channel_label}_bit_{bit}.png"
|
||||
out_path = output_dir / img_name
|
||||
dl_path = Path(output_dir.name) / img_name
|
||||
out_path = self.output_dir / img_name
|
||||
dl_path = Path(self.output_dir.name) / img_name
|
||||
channel_json.append("/image/" + str(dl_path))
|
||||
bit_img.save(out_path)
|
||||
image_json[channel_label] = channel_json
|
||||
|
||||
output = {
|
||||
"decomposer": {
|
||||
"status": "ok",
|
||||
"images": image_json,
|
||||
}
|
||||
}
|
||||
if converted:
|
||||
output["decomposer"][
|
||||
"note"
|
||||
] = "Image contains a color palette and was converted to RGB for processing."
|
||||
output["note"] = (
|
||||
"Image contains a color palette and was converted to RGB for processing."
|
||||
)
|
||||
|
||||
update_data(output_dir, output)
|
||||
return output
|
||||
|
||||
|
||||
def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using bits decomposition."""
|
||||
analyzer = DecomposerAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,45 +1,31 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("exiftool", input_img, output_dir)
|
||||
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()
|
||||
|
||||
@@ -1,59 +1,32 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("foremost", input_img, output_dir, 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()
|
||||
|
||||
@@ -1,57 +1,42 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Image size bruteforcer for common PNG sizes."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, List
|
||||
from typing import Any, List, Optional
|
||||
|
||||
from ..app import app
|
||||
from ..models import IHDR
|
||||
from .utils import PNG, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
from .utils import PNG
|
||||
|
||||
|
||||
def write_recovered_png(
|
||||
png_bytes: bytearray,
|
||||
width: int,
|
||||
height: int,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""
|
||||
Rebuilds a PNG by patching IHDR width/height and writes it to disk.
|
||||
"""
|
||||
height_bytes = height.to_bytes(4, byteorder="big")
|
||||
width_bytes = width.to_bytes(4, byteorder="big")
|
||||
class ResizeAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for image resize."""
|
||||
|
||||
full_png_data = (
|
||||
png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
|
||||
+ width_bytes
|
||||
+ height_bytes
|
||||
+ png_bytes[24:]
|
||||
)
|
||||
png: PNG
|
||||
|
||||
with output_path.open("wb") as out_f:
|
||||
out_f.write(full_png_data)
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("image_resize", input_img, output_dir, has_archive=True)
|
||||
|
||||
|
||||
def lookup_crc(png: PNG) -> List[PNG]:
|
||||
def _lookup_crc(self) -> List[PNG]:
|
||||
"""
|
||||
Look up PNG images in the database by their CRC value.
|
||||
|
||||
Args:
|
||||
png (PNG): A PNG object containing the CRC value to search for.
|
||||
|
||||
Returns:
|
||||
List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their packed
|
||||
representations.
|
||||
List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their
|
||||
packed representations.
|
||||
"""
|
||||
with app.app_context():
|
||||
imgs = IHDR.query.filter_by(crc=png.crc).all()
|
||||
imgs = IHDR.query.filter_by(crc=self.png.crc).all()
|
||||
results = []
|
||||
for img in imgs:
|
||||
packed = img.packed
|
||||
results.append(PNG.from_packed(packed, crc=png.crc))
|
||||
results.append(PNG.from_packed(packed, crc=self.png.crc))
|
||||
return results
|
||||
|
||||
|
||||
def search_height_crc(png: PNG) -> List[PNG]:
|
||||
def _search_height_crc(self) -> List[PNG]:
|
||||
"""
|
||||
Search for PNG images with matching CRC by iterating through possible heights.
|
||||
|
||||
@@ -59,9 +44,6 @@ def search_height_crc(png: PNG) -> List[PNG]:
|
||||
CRC checksum as the input PNG by systematically varying the height parameter while
|
||||
keeping all other properties constant.
|
||||
|
||||
Args:
|
||||
png (PNG): The reference PNG object whose CRC checksum will be matched.
|
||||
|
||||
Returns:
|
||||
List[PNG]: A list of PNG objects with matching CRC values. Returns an empty list
|
||||
if no matches are found.
|
||||
@@ -73,17 +55,39 @@ def search_height_crc(png: PNG) -> List[PNG]:
|
||||
results = []
|
||||
for height in range(100, 3501):
|
||||
candidate = PNG(
|
||||
size=(png.width, height),
|
||||
bit_depth=png.bit_depth,
|
||||
color_type=png.color_type,
|
||||
interlace=png.interlace,
|
||||
size=(self.png.width, height),
|
||||
bit_depth=self.png.bit_depth,
|
||||
color_type=self.png.color_type,
|
||||
interlace=self.png.interlace,
|
||||
)
|
||||
if candidate.crc == png.crc:
|
||||
if candidate.crc == self.png.crc:
|
||||
results.append(candidate)
|
||||
return results
|
||||
|
||||
def _write_recovered_png(self, png_bytes: bytearray, png_match: PNG) -> str:
|
||||
"""
|
||||
Rebuilds a PNG by patching IHDR width/height and writes it to disk.
|
||||
"""
|
||||
|
||||
def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
|
||||
img_name = f"recovered_{png_match.width}x{png_match.height}.png"
|
||||
output_path = self.output_dir / img_name
|
||||
|
||||
height_bytes = png_match.height.to_bytes(4, byteorder="big")
|
||||
width_bytes = png_match.width.to_bytes(4, byteorder="big")
|
||||
|
||||
full_png_data = (
|
||||
png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
|
||||
+ width_bytes
|
||||
+ height_bytes
|
||||
+ png_bytes[24:]
|
||||
)
|
||||
|
||||
with output_path.open("wb") as out_f:
|
||||
out_f.write(full_png_data)
|
||||
|
||||
return img_name
|
||||
|
||||
def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
|
||||
"""
|
||||
Analyze a PNG image and recover resized versions based on CRC matching.
|
||||
|
||||
@@ -91,12 +95,8 @@ def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
|
||||
by looking up its CRC (Cyclic Redundancy Check) in a database. If matches are
|
||||
found, it recovers and saves PNG images with the matched dimensions.
|
||||
|
||||
Args:
|
||||
input_img (Path): Path to the input PNG image file to analyze.
|
||||
output_dir (Path): Path to the directory where recovered PNG images will be saved.
|
||||
|
||||
Returns:
|
||||
None
|
||||
dict[str, Any]: A dictionary containing the analysis status, any error message,
|
||||
|
||||
Raises:
|
||||
Handles exceptions gracefully by updating output data with error status.
|
||||
@@ -108,88 +108,57 @@ def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
|
||||
|
||||
Notes:
|
||||
- If the PNG has invalid structure or IHDR chunk is not first, an error is logged.
|
||||
- The function performs a two-step lookup: first by CRC, then by height/CRC if no matches
|
||||
found.
|
||||
- The function performs a two-step lookup: first by CRC, then by height/CRC if no
|
||||
matches found.
|
||||
- Recovered images are named as "recovered_{width}x{height}.png".
|
||||
- Analysis results (logs, status, and image URLs) are persisted via update_data().
|
||||
"""
|
||||
|
||||
# 1. Setup Paths
|
||||
input_img = Path(input_img)
|
||||
output_dir = Path(output_dir)
|
||||
|
||||
# 3. Initialize Logs
|
||||
logs = []
|
||||
|
||||
try:
|
||||
with input_img.open("rb") as image:
|
||||
with self.input_img.open("rb") as image:
|
||||
img_bytes = bytearray(image.read())
|
||||
try:
|
||||
png = PNG.from_bytearray(img_bytes)
|
||||
self.png = PNG.from_bytearray(img_bytes)
|
||||
except ValueError:
|
||||
png = None
|
||||
|
||||
if png is None:
|
||||
logs.append(
|
||||
"Failure: IHDR chunk is not the first chunk, or PNG has invalid structure."
|
||||
)
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"image_resize": {
|
||||
return {
|
||||
"status": "error",
|
||||
"error": "IHDR chunk is not the first chunk, or PNG has invalid structure.",
|
||||
}
|
||||
},
|
||||
)
|
||||
return
|
||||
|
||||
logs.append(f"Target CRC found: 0x{png.crc:08x}")
|
||||
|
||||
if png.crc == png.compute_crc():
|
||||
note = (
|
||||
f"PNG is already valid with dimensions {png.width}x{png.height} "
|
||||
f"and crc 0x{png.crc:08x}."
|
||||
)
|
||||
output_no_compute: dict[str, Any] = {
|
||||
"image_resize": {"status": "ok", "note": note}
|
||||
# Already correct
|
||||
if self.png.crc == self.png.compute_crc():
|
||||
return {
|
||||
"status": "ok",
|
||||
"note": f"PNG is already valid with dimensions "
|
||||
f"{self.png.width}x{self.png.height} and crc 0x{self.png.crc:08x}.",
|
||||
}
|
||||
update_data(output_dir, output_no_compute)
|
||||
return
|
||||
|
||||
saved_img_urls = []
|
||||
|
||||
db_matches = lookup_crc(png)
|
||||
if not db_matches:
|
||||
db_matches = search_height_crc(png)
|
||||
if not db_matches:
|
||||
logs.append("Failure: No matching dimensions found.")
|
||||
else:
|
||||
# Create parent directories if they don't exist
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Iterate through ALL candidates found
|
||||
for im_match in db_matches:
|
||||
img_name = f"recovered_{im_match.width}x{im_match.height}.png"
|
||||
output_path = output_dir / img_name
|
||||
write_recovered_png(
|
||||
img_bytes, im_match.width, im_match.height, output_path
|
||||
)
|
||||
# No match found
|
||||
if not (db_matches := self._lookup_crc()) and not (db_matches := self._search_height_crc()):
|
||||
return {
|
||||
"status": "error",
|
||||
"error": "Failure: No matching dimensions found.",
|
||||
}
|
||||
|
||||
# Match found, saving image(s)
|
||||
logs.append(f"Target CRC found: 0x{self.png.crc:08x}")
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
for png_match in db_matches:
|
||||
img_name = self._write_recovered_png(img_bytes, png_match)
|
||||
logs.append(f"Image saved: {img_name}")
|
||||
# Append the formatted URL to our list
|
||||
saved_img_urls.append("/image/" + str(Path(output_dir.name) / img_name))
|
||||
saved_img_urls.append("/image/" + str(Path(self.output_dir.name) / img_name))
|
||||
|
||||
# 4. Final Data Update
|
||||
output_data: dict[str, Any] = {
|
||||
"image_resize": {
|
||||
return {
|
||||
"status": "ok",
|
||||
"output": logs,
|
||||
"png_images": saved_img_urls,
|
||||
}
|
||||
}
|
||||
|
||||
update_data(output_dir, output_data)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"image_resize": {"status": "error", "error": str(e)}})
|
||||
def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
|
||||
"""Analyze an image submission using ResizeAnalyzer."""
|
||||
analyzer = ResizeAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,114 +1,66 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""OpenStego Analyzer for Image Submissions."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from .utils import MAX_PENDING_TIME, update_data
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
|
||||
class OpenStegoAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for openstego."""
|
||||
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("openstego", input_img, output_dir, has_archive=True)
|
||||
self.algo = 0
|
||||
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
"""Iterator that return command for OpenStego for every algorithms"""
|
||||
if password is None:
|
||||
password = ""
|
||||
|
||||
algorithms = ["AES128", "AES256"]
|
||||
algo = algorithms[self.algo]
|
||||
self.algo += 1
|
||||
cmd = ["openstego", "extract", "-a", "randomlsb", "--cryptalgo", algo, "-sf"]
|
||||
cmd += [self.img, "-xd", str(self.get_extracted_dir()), "-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."""
|
||||
return (
|
||||
len(list(self.get_extracted_dir().glob("*"))) == 0 and "Extracted file: " not in stderr
|
||||
)
|
||||
|
||||
def analyze(self, password: Optional[str] = None) -> None:
|
||||
"""Run the subprocess command and handle results."""
|
||||
result: dict[str, Any]
|
||||
try:
|
||||
result = self.get_results(password)
|
||||
if result["status"] == "error": # check second algorithm if failed
|
||||
result = self.get_results(password)
|
||||
self.update_result(result)
|
||||
except Exception as e:
|
||||
self.update_result({"status": "error", "error": str(e)})
|
||||
raise
|
||||
|
||||
def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
|
||||
"""Process the stdout/stderr."""
|
||||
return stderr
|
||||
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
"""Process the stderr."""
|
||||
if "OpenStego is a steganography application" in stderr:
|
||||
return "OpenStego needs a password to work."
|
||||
return stderr
|
||||
|
||||
|
||||
def analyze_openstego(input_img: Path, output_dir: Path, password: str = "") -> None:
|
||||
"""Analyze an image submission using openstego."""
|
||||
|
||||
image_name = input_img.name
|
||||
try:
|
||||
stderr = ""
|
||||
# Try to extract the embedded file
|
||||
extracted_dir = output_dir / "openstego"
|
||||
extracted_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for algo in ["AES128", "AES256"]:
|
||||
cmd: list[str] = [
|
||||
"openstego",
|
||||
"extract",
|
||||
"-a",
|
||||
"randomlsb",
|
||||
"--cryptalgo",
|
||||
algo,
|
||||
"-sf",
|
||||
"../" + str(image_name),
|
||||
"-xd",
|
||||
str(extracted_dir),
|
||||
"-p",
|
||||
password,
|
||||
]
|
||||
|
||||
data = subprocess.run(
|
||||
cmd,
|
||||
cwd=output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
# Check if extraction was successful
|
||||
if data.returncode != 0:
|
||||
stderr += data.stderr.replace(
|
||||
f'"../{image_name}" ', ""
|
||||
) # hide file name
|
||||
err = {
|
||||
"openstego": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Find the extracted file
|
||||
extracted_files = list(extracted_dir.glob("*"))
|
||||
if not extracted_files:
|
||||
err = {
|
||||
"openstego": {
|
||||
"status": "error",
|
||||
"error": "No file extracted, password may be incorrect.",
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Filter stdout for success messages
|
||||
stdout = []
|
||||
for f in extracted_files:
|
||||
stdout.append(f"Recovered file: {f.name}")
|
||||
|
||||
# Zip extracted files
|
||||
zip_data = subprocess.run(
|
||||
["7z", "a", "../openstego.7z", "*"],
|
||||
cwd=extracted_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=MAX_PENDING_TIME,
|
||||
)
|
||||
|
||||
stderr += zip_data.stderr
|
||||
|
||||
if len(stderr) > 0:
|
||||
err = {
|
||||
"openstego": {
|
||||
"status": "error",
|
||||
"error": stderr,
|
||||
}
|
||||
}
|
||||
update_data(output_dir, err)
|
||||
return None
|
||||
|
||||
# Remove the extracted directory
|
||||
shutil.rmtree(extracted_dir)
|
||||
|
||||
update_data(
|
||||
output_dir,
|
||||
{
|
||||
"openstego": {
|
||||
"status": "ok",
|
||||
"output": stdout,
|
||||
"download": f"/download/{output_dir.name}/openstego",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
update_data(output_dir, {"openstego": {"status": "error", "error": str(e)}})
|
||||
return None
|
||||
analyzer = OpenStegoAnalyzer(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,96 +1,33 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("outguess", input_img, output_dir, 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]
|
||||
return ["outguess", "-r", self.img, out]
|
||||
|
||||
|
||||
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:
|
||||
cmd = ["outguess", "-r", "../../" + str(image_name), outguess_file]
|
||||
|
||||
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
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,53 +1,31 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("pngcheck", input_img, output_dir)
|
||||
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 process_error(self, stdout: str, stderr: str) -> str:
|
||||
"""Process the stderr."""
|
||||
if "this is neither a PNG or JNG image nor a MNG stream" in stdout:
|
||||
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."""
|
||||
|
||||
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()
|
||||
|
||||
@@ -1,127 +1,71 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("steghide", input_img, output_dir, 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_file = match.group(1)
|
||||
safe_hidden_file = Path(hidden_file).name
|
||||
outfile = str(self.get_extracted_dir() / safe_hidden_file)
|
||||
cmd = ["steghide", "extract", "-sf", self.img, "-xf", outfile, "-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 + stderr)
|
||||
embedded_filename = match.group(1) if match else None
|
||||
|
||||
match = re.search(r'wrote extracted data to "([^"]+)".*', stdout + stderr)
|
||||
extracted_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 embedded_filename is None and extracted_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)
|
||||
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"):
|
||||
if "wrote extracted data to" in line:
|
||||
out.append(line)
|
||||
return out
|
||||
|
||||
cmd_ext: list[str] = [
|
||||
"steghide",
|
||||
"extract",
|
||||
"-sf",
|
||||
"../../" + str(image_name),
|
||||
"-xf",
|
||||
str(embedded_file.name),
|
||||
]
|
||||
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)."
|
||||
return stderr
|
||||
|
||||
|
||||
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:
|
||||
cmd_ext += ["-p", password]
|
||||
analyzer.analyze(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"):
|
||||
if line.startswith('wrote extracted data to "'):
|
||||
stdout.append(line)
|
||||
elif line:
|
||||
stderr += line + "\n"
|
||||
|
||||
# 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,
|
||||
)
|
||||
|
||||
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
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,44 +1,22 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("strings", input_img, output_dir)
|
||||
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,76 +1,30 @@
|
||||
"""Custom Analyzer for Image Submissions."""
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""<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 TemplateAnalyzer(SubprocessAnalyzer):
|
||||
"""Analyzer for <toolname>."""
|
||||
|
||||
image_name = input_img.name
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("<toolname>", input_img, output_dir, 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 = TemplateAnalyzer(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
|
||||
@@ -1,55 +1,13 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Utility functions for analyzers modules."""
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import struct
|
||||
import threading
|
||||
import zlib
|
||||
from functools import cached_property
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Self
|
||||
|
||||
_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"
|
||||
) -> None:
|
||||
"""Thread-safe and process-safe JSON update using lock file and atomic replace."""
|
||||
json_file = Path(output_dir) / json_filename
|
||||
lock_file = json_file.with_suffix(json_file.suffix + ".lock")
|
||||
tmp_file = json_file.with_suffix(json_file.suffix + ".tmp")
|
||||
|
||||
json_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with _thread_lock: # synchronizes across threads
|
||||
with open(lock_file, "w", encoding="utf-8") as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX) # synchronizes across processes
|
||||
|
||||
try:
|
||||
# Read existing JSON
|
||||
data: dict[Any, Any] = {}
|
||||
if json_file.exists():
|
||||
try:
|
||||
with open(json_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except json.JSONDecodeError:
|
||||
data = {}
|
||||
|
||||
# Update with new data
|
||||
data.update(new_data)
|
||||
|
||||
# Write safely to a temp file
|
||||
with open(tmp_file, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, sort_keys=False)
|
||||
|
||||
os.replace(tmp_file, json_file) # ensures file write is atomic
|
||||
finally:
|
||||
fcntl.flock(lock, fcntl.LOCK_UN)
|
||||
from typing import Iterable, Self
|
||||
|
||||
|
||||
def get_resolutions() -> list[tuple[int, int]]:
|
||||
@@ -288,9 +246,7 @@ class PNG:
|
||||
crc = struct.unpack(">I", data[offset : offset + 4])[0]
|
||||
|
||||
# Unpack IHDR fields
|
||||
width, height, depth, color, _, _, interlace = struct.unpack(
|
||||
">IIBBBBB", ihdr_data
|
||||
)
|
||||
width, height, depth, color, _, _, interlace = struct.unpack(">IIBBBBB", ihdr_data)
|
||||
|
||||
return cls(
|
||||
size=(width, height),
|
||||
|
||||
@@ -1,40 +1,31 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Zsteg 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, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("zsteg", input_img, output_dir)
|
||||
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()
|
||||
|
||||
+55
-30
@@ -1,3 +1,6 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Aperi'Solve Flask application."""
|
||||
|
||||
import hashlib
|
||||
@@ -8,10 +11,15 @@ from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
import sentry_sdk
|
||||
from flask import Flask, Response, abort, jsonify, render_template, request, send_file
|
||||
from redis import Redis
|
||||
from rq import Queue
|
||||
|
||||
from .sentry import initialize_sentry
|
||||
|
||||
initialize_sentry()
|
||||
|
||||
from .cleanup import cleanup_old_entries
|
||||
from .config import IMAGE_EXTENSIONS, RESULT_FOLDER, WORKER_FILES
|
||||
from .models import Image, Submission, db, init_db
|
||||
@@ -20,14 +28,13 @@ 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(
|
||||
os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024)
|
||||
)
|
||||
app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024))
|
||||
RESULT_FOLDER.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
db.init_app(app)
|
||||
with app.app_context():
|
||||
init_db(app)
|
||||
|
||||
redis_conn = Redis(host="redis", port=6379)
|
||||
queue = Queue(connection=redis_conn)
|
||||
|
||||
@@ -35,6 +42,14 @@ queue = Queue(connection=redis_conn)
|
||||
@app.errorhandler(413)
|
||||
def too_large(_: Any) -> tuple[Response, int]:
|
||||
"""Error Handler for Max File Size."""
|
||||
sentry_sdk.set_context(
|
||||
"upload",
|
||||
{
|
||||
"content_length": request.content_length,
|
||||
"max_allowed": app.config["MAX_CONTENT_LENGTH"],
|
||||
},
|
||||
)
|
||||
sentry_sdk.capture_message("Upload failed: 413 Payload Too Large", level="warning")
|
||||
return jsonify({"error": "Image size exceeded"}), 413
|
||||
|
||||
|
||||
@@ -44,6 +59,9 @@ def not_found(_: Any) -> str:
|
||||
return render_template("error.html", message="Resource not found", statuscode=404)
|
||||
|
||||
|
||||
# Sentry automatically captures 500 errors
|
||||
|
||||
|
||||
@app.route("/")
|
||||
def index() -> str:
|
||||
"""Render the main index page."""
|
||||
@@ -92,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 (
|
||||
"." not in image.filename
|
||||
or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS
|
||||
):
|
||||
if "." 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
|
||||
@@ -112,21 +127,28 @@ def upload_image() -> tuple[Response, int]:
|
||||
submission_hash = hashlib.md5(submission_data).hexdigest()
|
||||
submission_path = RESULT_FOLDER / img_hash / submission_hash
|
||||
|
||||
# If submission already exists, return submission hash
|
||||
if submission_path.exists():
|
||||
submission: Submission = Submission.query.filter_by(
|
||||
hash=submission_hash
|
||||
).first()
|
||||
# Check if hash is present on the DB
|
||||
submission: Submission = Submission.query.filter_by(hash=submission_hash).first()
|
||||
|
||||
# Only return if BOTH the file exists AND the DB entry exists
|
||||
if submission_path.exists() and submission is not None:
|
||||
return jsonify({"submission_hash": submission.hash}), 200
|
||||
|
||||
# If image is new, create a new Image entry
|
||||
# Self-Healing Logic
|
||||
new_img_path = RESULT_FOLDER / img_hash / image_name
|
||||
if not new_img_path.parent.exists():
|
||||
|
||||
# This fixes the "Ghost File" issue (DB exists, but file is missing)
|
||||
if not new_img_path.exists():
|
||||
new_img_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(new_img_path, "wb") as f: # Write file to disk
|
||||
with open(new_img_path, "wb") as f:
|
||||
f.write(image_data)
|
||||
|
||||
new_img = Image(
|
||||
# Try to find image in DB by hash
|
||||
sub_img = Image.query.filter_by(hash=img_hash).first()
|
||||
|
||||
# If DB entry is missing (even if file exists on disk), create it!
|
||||
if sub_img is None:
|
||||
sub_img = Image(
|
||||
file=str(new_img_path),
|
||||
hash=img_hash,
|
||||
size=len(image_data),
|
||||
@@ -134,15 +156,19 @@ def upload_image() -> tuple[Response, int]:
|
||||
first_submission_date=datetime.now(timezone.utc),
|
||||
last_submission_date=datetime.now(timezone.utc),
|
||||
)
|
||||
db.session.add(new_img) # pylint: disable=no-member
|
||||
db.session.add(sub_img) # pylint: disable=no-member
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
|
||||
sub_img = Image.query.filter_by(hash=img_hash).first() # type: ignore
|
||||
# Increment upload count for the image
|
||||
sub_img.upload_count += 1
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
|
||||
# Create new Submission entry
|
||||
submission_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Only create a new submission if it does not already exist
|
||||
# New submission for new image
|
||||
if not submission:
|
||||
submission = Submission(
|
||||
filename=image.filename,
|
||||
password=password,
|
||||
@@ -150,14 +176,19 @@ def upload_image() -> tuple[Response, int]:
|
||||
hash=submission_hash,
|
||||
status="pending",
|
||||
date=time.time(),
|
||||
image_hash=sub_img.hash, # type: ignore
|
||||
image_hash=sub_img.hash,
|
||||
)
|
||||
db.session.add(submission) # pylint: disable=no-member
|
||||
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
|
||||
# Re-analysis case to prevent early return
|
||||
submission.status = "pending" # type: ignore
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
|
||||
# Start the analysis jobs
|
||||
queue.enqueue("aperisolve.workers.analyze_image", submission.hash, job_timeout=300)
|
||||
|
||||
# Return submission identifier
|
||||
return jsonify({"submission_hash": submission.hash}), 200
|
||||
|
||||
|
||||
@@ -194,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_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
|
||||
)
|
||||
result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
|
||||
|
||||
if not result_path.exists():
|
||||
return jsonify({"error": "Results not ready yet..."}), 425
|
||||
@@ -225,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(
|
||||
hash_val: Optional[str] = None, img_name: Optional[str] = None
|
||||
) -> Response:
|
||||
def get_image(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.
|
||||
@@ -247,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 (
|
||||
output_file.suffix.lower() not in IMAGE_EXTENSIONS
|
||||
):
|
||||
if (not output_file.exists()) or (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)
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Cleanup old entries in the database and file system."""
|
||||
|
||||
import shutil
|
||||
@@ -46,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_FOLDER / str(submission.image_hash) / str(submission.hash)
|
||||
)
|
||||
result_path = 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
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Aperi'Solve configuration variables."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
+34
-10
@@ -1,7 +1,14 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""This module defines the database models for the Aperi'Solve application."""
|
||||
|
||||
import itertools
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from os import getenv
|
||||
from pathlib import Path
|
||||
from shutil import rmtree
|
||||
|
||||
from flask import Flask
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
@@ -56,7 +63,7 @@ class IHDR(db.Model): # type: ignore
|
||||
packed = Column(BigInteger, nullable=False)
|
||||
|
||||
|
||||
def create_crc_db() -> None:
|
||||
def fill_ihdr_db() -> None:
|
||||
"""
|
||||
Creates a database of common PNG IHDR (Image Header) entries by generating
|
||||
combinations of standard PNG parameters such as resolutions, bit depths,
|
||||
@@ -67,6 +74,11 @@ def create_crc_db() -> None:
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
# Skip if table already has entries
|
||||
if db.session.query(IHDR.iid).first() is not None:
|
||||
print("IHDR table already populated, skipping fill.")
|
||||
return
|
||||
|
||||
# Standard PNG IHDR parameters
|
||||
resolutions = get_resolutions()
|
||||
bit_color_pairs = get_valid_depth_color_pairs()
|
||||
@@ -91,9 +103,7 @@ def create_crc_db() -> None:
|
||||
if exists: # Skip if already present
|
||||
continue
|
||||
|
||||
db.session.add( # pylint: disable=no-member
|
||||
IHDR(packed=p.packed, crc=p.crc)
|
||||
)
|
||||
db.session.add(IHDR(packed=p.packed, crc=p.crc)) # pylint: disable=no-member
|
||||
count += 1
|
||||
|
||||
if count % 5000 == 0:
|
||||
@@ -101,7 +111,7 @@ def create_crc_db() -> None:
|
||||
db.session.commit()
|
||||
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
print(f"Precomputed {count} common IHDR entries.")
|
||||
print(f"Precomputed {count} common IHDR entries. IHDR table filled successfully.")
|
||||
|
||||
|
||||
def init_db(app: Flask) -> None:
|
||||
@@ -123,11 +133,25 @@ def init_db(app: Flask) -> None:
|
||||
- Populates the IHDR CRC lookup table with initial data
|
||||
- Prints status messages to console during initialization
|
||||
"""
|
||||
# Detect if running as RQ worker by checking command line
|
||||
# Worker: /usr/local/bin/rq worker ...
|
||||
# Web: /usr/local/bin/flask run ... or gunicorn/wsgi
|
||||
is_worker = any("rq" in arg and "flask" not in arg for arg in sys.argv[:2])
|
||||
|
||||
with app.app_context():
|
||||
if not db.engine.dialect.has_table(db.engine.connect(), "image"):
|
||||
print("Creating database...")
|
||||
# Workers should never clear the database
|
||||
if not is_worker and getenv("CLEAR_AT_RESTART", "0") == "1":
|
||||
print("Clearing database and file system at restart...")
|
||||
db.session.remove() # pylint: disable=no-member
|
||||
db.drop_all()
|
||||
rmtree(Path("./results"), ignore_errors=True) # Clear results folder
|
||||
|
||||
# Always create tables (idempotent - safe to call multiple times)
|
||||
db.create_all()
|
||||
print("Database structure created successfully.")
|
||||
print("Filling IHDR lookup table...")
|
||||
create_crc_db()
|
||||
print("IHDR table filled successfully.")
|
||||
|
||||
# Workers should never fill IHDR table
|
||||
if is_worker:
|
||||
print("Running as worker, skipping IHDR table fill.")
|
||||
else:
|
||||
fill_ihdr_db()
|
||||
|
||||
@@ -8,3 +8,4 @@ types-Flask-SQLAlchemy
|
||||
sqlalchemy-stubs
|
||||
Pillow
|
||||
gunicorn
|
||||
sentry-sdk[flask,sqlalchemy,rq]
|
||||
@@ -0,0 +1,48 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Sentry initialization for Aperisolve."""
|
||||
|
||||
import os
|
||||
|
||||
import sentry_sdk
|
||||
from sentry_sdk.integrations.flask import FlaskIntegration
|
||||
from sentry_sdk.integrations.rq import RqIntegration
|
||||
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
|
||||
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(os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1"))
|
||||
SENTRY_RELEASE = os.environ.get("SENTRY_RELEASE", "1.0.0")
|
||||
|
||||
|
||||
def initialize_sentry() -> None:
|
||||
"""Initialize Sentry SDK once, safely."""
|
||||
|
||||
# No DSN → do nothing
|
||||
# Prevent double initialization (Flask + RQ, CLI, tests, etc.)
|
||||
if not SENTRY_DSN or sentry_sdk.Hub.current.client is not None:
|
||||
return
|
||||
|
||||
try:
|
||||
sentry_sdk.init(
|
||||
dsn=SENTRY_DSN,
|
||||
integrations=[
|
||||
FlaskIntegration(),
|
||||
SqlalchemyIntegration(),
|
||||
RqIntegration(),
|
||||
ThreadingIntegration(propagate_hub=True),
|
||||
],
|
||||
traces_sample_rate=SENTRY_TRACES_SAMPLE_RATE,
|
||||
profiles_sample_rate=SENTRY_PROFILES_SAMPLE_RATE,
|
||||
environment=SENTRY_ENVIRONMENT,
|
||||
release=SENTRY_RELEASE,
|
||||
send_default_pii=False,
|
||||
enable_tracing=True,
|
||||
)
|
||||
except sentry_sdk.utils.BadDsn as e:
|
||||
print(f"Sentry DSN is invalid, skipping Sentry initialization: {e}")
|
||||
except Exception as e:
|
||||
print(f"Sentry initialization failed: {e}")
|
||||
@@ -18,10 +18,15 @@ html {
|
||||
--bs-progress-bar-bg: #f00 !important;
|
||||
}
|
||||
|
||||
* {
|
||||
box-sizing: border-box;
|
||||
}
|
||||
|
||||
body {
|
||||
background-color: var(--dark-bg);
|
||||
color: #e0e0e0;
|
||||
font-family: 'Fira Code', monospace;
|
||||
overflow-x: hidden;
|
||||
}
|
||||
|
||||
/* ======== Typography ======== */
|
||||
@@ -219,7 +224,6 @@ pre {
|
||||
background-color: var(--dark-bg-hover);
|
||||
box-shadow: 0 0 10px var(--theme-green);
|
||||
}
|
||||
|
||||
#upload-form {
|
||||
transition: height 0.4s ease, padding 0.4s ease, margin 0.4s ease;
|
||||
overflow: hidden;
|
||||
@@ -269,6 +273,7 @@ td {
|
||||
display: inline-block;
|
||||
padding: 10px;
|
||||
box-sizing: border-box;
|
||||
min-width: 150px;
|
||||
}
|
||||
|
||||
/* ======== Modal ======== */
|
||||
@@ -289,7 +294,10 @@ td {
|
||||
}
|
||||
|
||||
#image-modal-container img {
|
||||
width: 60%;
|
||||
width: 90%;
|
||||
max-width: 800px;
|
||||
max-height: 90vh;
|
||||
object-fit: contain;
|
||||
border: 2px solid var(--theme-green);
|
||||
box-shadow: 0 0 20px var(--theme-green);
|
||||
cursor: pointer;
|
||||
@@ -304,6 +312,7 @@ td {
|
||||
display: inline-block;
|
||||
padding: 10px;
|
||||
width: 25%;
|
||||
min-width: 150px;
|
||||
}
|
||||
|
||||
/* ======== FAQ ======== */
|
||||
@@ -345,6 +354,19 @@ td {
|
||||
#cheatsheet img,
|
||||
#cheatsheet table {
|
||||
margin-bottom: 15px;
|
||||
max-width: 100%;
|
||||
height: auto;
|
||||
}
|
||||
|
||||
#cheatsheet table {
|
||||
min-width: auto;
|
||||
width: auto;
|
||||
}
|
||||
|
||||
.command {
|
||||
overflow-x: auto;
|
||||
-webkit-overflow-scrolling: touch;
|
||||
word-wrap: break-word;
|
||||
}
|
||||
|
||||
/* ======== Analyzer Blocks ======== */
|
||||
@@ -355,15 +377,14 @@ td {
|
||||
|
||||
#result-infos {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 15px;
|
||||
}
|
||||
|
||||
#result-infos>div {
|
||||
height: 100%;
|
||||
width: 50%;
|
||||
height: auto;
|
||||
flex: 1 1 0;
|
||||
width: 50%;
|
||||
display: inline-block;
|
||||
flex: 1 1 300px;
|
||||
min-width: 280px;
|
||||
padding: 10px;
|
||||
box-sizing: border-box;
|
||||
vertical-align: top;
|
||||
@@ -379,6 +400,10 @@ td {
|
||||
height: 400px;
|
||||
}
|
||||
|
||||
.a-foremost textarea {
|
||||
height: 100px;
|
||||
}
|
||||
|
||||
/* ======== Footer ======== */
|
||||
|
||||
footer {
|
||||
@@ -387,6 +412,7 @@ footer {
|
||||
margin: 0 20px 20px 0;
|
||||
justify-content: flex-end !important;
|
||||
}
|
||||
|
||||
@media screen and (min-width: 1800px) {
|
||||
footer {
|
||||
position: fixed;
|
||||
@@ -421,3 +447,388 @@ footer a i {
|
||||
#httperror h1, #httperror h2{
|
||||
text-align: center;
|
||||
}
|
||||
|
||||
/* ======== Responsive Design ======== */
|
||||
|
||||
/* Large tablets and small desktops */
|
||||
@media screen and (max-width: 1200px) {
|
||||
.results_img {
|
||||
width: 33.333%;
|
||||
}
|
||||
|
||||
.browse-images {
|
||||
width: 33.333%;
|
||||
}
|
||||
}
|
||||
|
||||
/* Tablets */
|
||||
@media screen and (max-width: 992px) {
|
||||
.container {
|
||||
padding-left: 15px;
|
||||
padding-right: 15px;
|
||||
}
|
||||
|
||||
.results_img {
|
||||
width: 50%;
|
||||
}
|
||||
|
||||
.browse-images {
|
||||
width: 50%;
|
||||
}
|
||||
|
||||
#result-infos {
|
||||
flex-direction: column;
|
||||
}
|
||||
|
||||
#result-infos > div {
|
||||
width: 100%;
|
||||
flex: 1 1 100%;
|
||||
}
|
||||
|
||||
table {
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
th, td {
|
||||
padding: 0.5rem;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/* Large phones and small tablets */
|
||||
@media screen and (max-width: 768px) {
|
||||
body {
|
||||
font-size: 14px;
|
||||
}
|
||||
|
||||
.container {
|
||||
margin-top: 20px !important;
|
||||
padding-left: 10px;
|
||||
padding-right: 10px;
|
||||
}
|
||||
|
||||
h1 {
|
||||
font-size: 1.5rem;
|
||||
}
|
||||
|
||||
h2 {
|
||||
font-size: 1.25rem;
|
||||
}
|
||||
|
||||
h3 {
|
||||
font-size: 1.1rem;
|
||||
}
|
||||
|
||||
#water-logo {
|
||||
width: 50px;
|
||||
}
|
||||
|
||||
.card, .form-control, .table, .border, .card-hack {
|
||||
border-radius: 0.35rem;
|
||||
}
|
||||
|
||||
.card {
|
||||
padding: 15px !important;
|
||||
}
|
||||
|
||||
#drop-zone {
|
||||
padding: 20px !important;
|
||||
}
|
||||
|
||||
#drop-zone p {
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
.results_img {
|
||||
width: 50%;
|
||||
padding: 5px;
|
||||
}
|
||||
|
||||
.browse-images {
|
||||
width: 50%;
|
||||
padding: 5px;
|
||||
}
|
||||
|
||||
textarea.form-control, pre {
|
||||
padding: 15px;
|
||||
font-size: 0.85rem;
|
||||
}
|
||||
|
||||
.a-strings textarea {
|
||||
height: 250px;
|
||||
}
|
||||
|
||||
.a-exiftool .table-container {
|
||||
max-height: 300px;
|
||||
}
|
||||
|
||||
/* Make tables scrollable horizontally */
|
||||
.table-container {
|
||||
overflow-x: auto;
|
||||
-webkit-overflow-scrolling: touch;
|
||||
}
|
||||
|
||||
table {
|
||||
font-size: 0.8rem;
|
||||
min-width: 400px;
|
||||
}
|
||||
|
||||
th, td {
|
||||
padding: 0.4rem;
|
||||
word-break: break-word;
|
||||
}
|
||||
|
||||
footer {
|
||||
justify-content: center !important;
|
||||
margin: 20px 0;
|
||||
flex-wrap: wrap;
|
||||
gap: 15px;
|
||||
}
|
||||
|
||||
footer a i {
|
||||
font-size: 28px;
|
||||
}
|
||||
|
||||
#coffee {
|
||||
height: 28px;
|
||||
}
|
||||
|
||||
/* FAQ/Cheatsheet adjustments */
|
||||
#cheatsheet > div {
|
||||
padding: 8px;
|
||||
margin-bottom: 30px;
|
||||
}
|
||||
|
||||
#cheatsheet img {
|
||||
max-width: 100%;
|
||||
height: auto;
|
||||
}
|
||||
|
||||
.command {
|
||||
padding: 8px;
|
||||
font-size: 0.85rem;
|
||||
overflow-x: auto;
|
||||
white-space: nowrap;
|
||||
}
|
||||
}
|
||||
|
||||
/* Phones */
|
||||
@media screen and (max-width: 576px) {
|
||||
body {
|
||||
font-size: 13px;
|
||||
}
|
||||
|
||||
.container {
|
||||
margin-top: 15px !important;
|
||||
padding-left: 8px;
|
||||
padding-right: 8px;
|
||||
}
|
||||
|
||||
h1 {
|
||||
font-size: 1.3rem;
|
||||
}
|
||||
|
||||
h2 {
|
||||
font-size: 1.1rem;
|
||||
}
|
||||
|
||||
h3 {
|
||||
font-size: 1rem;
|
||||
}
|
||||
|
||||
#water-logo {
|
||||
width: 40px;
|
||||
}
|
||||
|
||||
.card {
|
||||
padding: 12px !important;
|
||||
}
|
||||
|
||||
#drop-zone {
|
||||
padding: 15px !important;
|
||||
}
|
||||
|
||||
#drop-zone p {
|
||||
font-size: 0.85rem;
|
||||
}
|
||||
|
||||
.btn, .btn-primary, .btn-hack {
|
||||
width: 100%;
|
||||
padding: 12px;
|
||||
font-size: 1rem;
|
||||
}
|
||||
|
||||
.results_img {
|
||||
width: 100%;
|
||||
padding: 5px;
|
||||
}
|
||||
|
||||
.browse-images {
|
||||
width: 100%;
|
||||
padding: 5px;
|
||||
}
|
||||
|
||||
#result-infos > div {
|
||||
padding: 5px;
|
||||
}
|
||||
|
||||
textarea.form-control, pre {
|
||||
padding: 10px;
|
||||
font-size: 0.8rem;
|
||||
}
|
||||
|
||||
.a-strings textarea {
|
||||
height: 200px;
|
||||
}
|
||||
|
||||
.a-exiftool .table-container {
|
||||
max-height: 250px;
|
||||
}
|
||||
|
||||
table {
|
||||
font-size: 0.75rem;
|
||||
}
|
||||
|
||||
th, td {
|
||||
padding: 0.3rem;
|
||||
}
|
||||
|
||||
.copy-icon {
|
||||
display: block;
|
||||
right: 5px;
|
||||
margin-top: -35px;
|
||||
}
|
||||
|
||||
footer {
|
||||
gap: 12px;
|
||||
}
|
||||
|
||||
footer a i {
|
||||
font-size: 24px;
|
||||
}
|
||||
|
||||
#coffee {
|
||||
height: 24px;
|
||||
}
|
||||
|
||||
#image-modal-container img {
|
||||
width: 95%;
|
||||
}
|
||||
|
||||
.analyzer {
|
||||
margin-top: 15px;
|
||||
}
|
||||
|
||||
.analyzer h2 + div {
|
||||
margin-top: 10px;
|
||||
}
|
||||
|
||||
.analyzer h3 {
|
||||
margin-top: 15px;
|
||||
}
|
||||
|
||||
/* FAQ/Cheatsheet adjustments */
|
||||
#cheatsheet li {
|
||||
padding-left: 20px;
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
.command {
|
||||
font-size: 0.75rem;
|
||||
padding: 6px;
|
||||
}
|
||||
|
||||
#httperror {
|
||||
padding-top: 50px;
|
||||
}
|
||||
}
|
||||
|
||||
/* Extra small phones */
|
||||
@media screen and (max-width: 380px) {
|
||||
body {
|
||||
font-size: 12px;
|
||||
}
|
||||
|
||||
h1 {
|
||||
font-size: 1.1rem;
|
||||
}
|
||||
|
||||
#water-logo {
|
||||
width: 35px;
|
||||
}
|
||||
|
||||
.btn, .btn-primary, .btn-hack {
|
||||
padding: 10px;
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
footer a i {
|
||||
font-size: 22px;
|
||||
}
|
||||
|
||||
#coffee {
|
||||
height: 22px;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/* Touch device optimizations */
|
||||
@media (hover: none) and (pointer: coarse) {
|
||||
.copy-icon {
|
||||
display: block;
|
||||
opacity: 0.7;
|
||||
}
|
||||
|
||||
.btn, .btn-primary, .btn-hack {
|
||||
min-height: 44px;
|
||||
}
|
||||
|
||||
input, textarea, select {
|
||||
font-size: 16px; /* Prevents zoom on iOS */
|
||||
}
|
||||
|
||||
a, button {
|
||||
min-height: 44px;
|
||||
min-width: 44px;
|
||||
}
|
||||
}
|
||||
|
||||
/* Landscape mode on phones */
|
||||
@media screen and (max-height: 500px) and (orientation: landscape) {
|
||||
.container {
|
||||
margin-top: 10px !important;
|
||||
}
|
||||
|
||||
#image-modal-container img {
|
||||
max-height: 85vh;
|
||||
width: auto;
|
||||
max-width: 95%;
|
||||
}
|
||||
}
|
||||
|
||||
/* Print styles */
|
||||
@media print {
|
||||
#three-bg {
|
||||
display: none;
|
||||
}
|
||||
|
||||
body {
|
||||
background-color: white;
|
||||
color: black;
|
||||
}
|
||||
|
||||
.card, .form-control, .table, .border, .card-hack {
|
||||
background-color: white;
|
||||
border-color: #333 !important;
|
||||
box-shadow: none;
|
||||
}
|
||||
|
||||
h1, h2, h3, h4, h5, h6, strong, .highlight {
|
||||
color: #333;
|
||||
text-shadow: none;
|
||||
}
|
||||
|
||||
a, .text-accent {
|
||||
color: #333;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -252,6 +252,38 @@ function parseResult(result) {
|
||||
|
||||
analyzer.innerHTML += `<h2>${capitalize(tool.replace("_"," "))}</h2>`;
|
||||
|
||||
// Parse text output
|
||||
if (typeof result[tool]["output"] === "string") {
|
||||
output = escapeHtml(result[tool]["output"]);
|
||||
analyzer.innerHTML += `<div class="alert alert-success" role="alert">${output}</div>`;
|
||||
} else if (Array.isArray(result[tool]["output"])) {
|
||||
if (result[tool]["output"].length > 0) {
|
||||
var texarea_content = `<div class="textarea-container">`;
|
||||
texarea_content += `<textarea class="form-control w-100 mb-2" rows="8" readonly>`;
|
||||
for (const line of result[tool]["output"]) {
|
||||
texarea_content += escapeHtml(`${line}\n`);
|
||||
}
|
||||
texarea_content = texarea_content.trim();
|
||||
texarea_content += `</textarea>`;
|
||||
texarea_content += `<i class="fas fa-copy copy-icon"></i>`;
|
||||
texarea_content += `</div>`;
|
||||
analyzer.innerHTML += texarea_content;
|
||||
}
|
||||
} else if (typeof result[tool]["output"] === "object") {
|
||||
var table_content = `<div class="table-container">`;
|
||||
table_content += `<table>`;
|
||||
for (const key in result[tool]["output"]) {
|
||||
table_content += `<tr><td>${escapeHtml(key)}</td>`;
|
||||
table_content += `<td>${escapeHtml(
|
||||
result[tool]["output"][key]
|
||||
)}</td></tr>`;
|
||||
}
|
||||
table_content += `</table>`;
|
||||
table_content += `</table>`;
|
||||
analyzer.innerHTML += table_content;
|
||||
}
|
||||
|
||||
// Parse images, downloads, ...
|
||||
if (result[tool]["status"] === "ok") {
|
||||
if ("images" in result[tool]) {
|
||||
// Parse image output
|
||||
@@ -286,41 +318,6 @@ function parseResult(result) {
|
||||
}
|
||||
}
|
||||
|
||||
// Parse text output
|
||||
|
||||
|
||||
if (typeof result[tool]["output"] === "string") {
|
||||
analyzer.innerHTML += `<pre>${escapeHtml(
|
||||
result[tool]["output"]
|
||||
)}</pre>`;
|
||||
} else if (Array.isArray(result[tool]["output"])) {
|
||||
if (result[tool]["output"].length > 0) {
|
||||
var texarea_content = `<div class="textarea-container">`;
|
||||
texarea_content += `<textarea class="form-control w-100 mb-2" rows="8" readonly>`;
|
||||
for (const line of result[tool]["output"]) {
|
||||
texarea_content += escapeHtml(`${line}\n`);
|
||||
}
|
||||
texarea_content += `</textarea>`;
|
||||
texarea_content += `<i class="fas fa-copy copy-icon"></i>`;
|
||||
texarea_content += `</div>`;
|
||||
analyzer.innerHTML += texarea_content;
|
||||
}
|
||||
} else if (typeof result[tool]["output"] === "object") {
|
||||
var table_content = `<div class="table-container">`;
|
||||
table_content += `<table>`;
|
||||
for (const key in result[tool]["output"]) {
|
||||
table_content += `<tr><td>${escapeHtml(key)}</td>`;
|
||||
table_content += `<td>${escapeHtml(
|
||||
result[tool]["output"][key]
|
||||
)}</td></tr>`;
|
||||
}
|
||||
table_content += `</table>`;
|
||||
table_content += `</table>`;
|
||||
analyzer.innerHTML += table_content;
|
||||
} else {
|
||||
// analyzer.innerHTML += `<pre>${result[tool]["output"]}</pre>`;
|
||||
}
|
||||
|
||||
if ("download" in result[tool]) {
|
||||
// Parse download link
|
||||
analyzer.innerHTML += `<br/><a href="${escapeHtml(
|
||||
@@ -489,6 +486,8 @@ if (browseBtn) {
|
||||
} catch (e) {
|
||||
showDanger("❌ Invalid server response.", true);
|
||||
}
|
||||
} else if (xhr.status == 413) {
|
||||
showDanger("❌ File too large. Please upload a smaller file.", true);
|
||||
}else {
|
||||
showDanger(`❌ HTTP error ${xhr.status}`, true);
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<title>Aperi'Solve</title>
|
||||
<link rel="icon" type="image/svg+xml" href="/static/img/logo.svg" />
|
||||
<link href="https://fonts.googleapis.com/css2?family=Fira+Code&display=swap" rel="stylesheet" />
|
||||
|
||||
+41
-10
@@ -1,10 +1,19 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Asynchronous worker for analyzing image submissions."""
|
||||
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .analyzers.binwalk import analyze_binwalk
|
||||
import sentry_sdk
|
||||
|
||||
from .sentry import initialize_sentry
|
||||
|
||||
initialize_sentry()
|
||||
|
||||
from .analyzers.binwalk import analyze_binwalk # pylint: disable=C0413
|
||||
from .analyzers.decomposer import analyze_decomposer
|
||||
from .analyzers.exiftool import analyze_exiftool
|
||||
from .analyzers.foremost import analyze_foremost
|
||||
@@ -51,6 +60,9 @@ def analyze_image(submission_hash: str) -> None:
|
||||
image = Image.query.get_or_404(submission.image_hash) # type: ignore
|
||||
|
||||
if not submission or not image: # No submission found
|
||||
sentry_sdk.capture_message(
|
||||
f"Submission or image not found: {submission_hash}", level="warning"
|
||||
)
|
||||
return
|
||||
|
||||
submission.status = "running" # type: ignore
|
||||
@@ -65,22 +77,38 @@ def analyze_image(submission_hash: str) -> None:
|
||||
|
||||
def run_analyzer(analyzer_func: Any, *args: Any) -> None:
|
||||
"""Run an analyzer function in a separate thread."""
|
||||
analyzer_name = analyzer_func.__name__.replace("analyze_", "")
|
||||
try:
|
||||
analyzer_func(*args)
|
||||
except Exception as e:
|
||||
print(f"Error in {analyzer_func.__name__}: {e}")
|
||||
print(f"Error in {analyzer_name}: {e}")
|
||||
with sentry_sdk.push_scope() as scope:
|
||||
scope.set_tag("analyzer", analyzer_name)
|
||||
scope.set_tag("submission_hash", submission_hash)
|
||||
scope.set_context(
|
||||
"analyzer_info",
|
||||
{
|
||||
"tool": analyzer_name,
|
||||
"image_path": str(img_path),
|
||||
"result_path": str(result_path),
|
||||
"filename": submission.filename,
|
||||
"deep_analysis": submission.deep_analysis,
|
||||
},
|
||||
)
|
||||
scope.fingerprint = ["analyzer-error", analyzer_name]
|
||||
sentry_sdk.capture_exception(e)
|
||||
|
||||
analyzers = [
|
||||
(analyze_binwalk, img_path, result_path),
|
||||
(analyze_decomposer, img_path, result_path),
|
||||
(analyze_exiftool, img_path, result_path),
|
||||
(analyze_foremost, img_path, result_path),
|
||||
(analyze_strings, img_path, result_path),
|
||||
(analyze_pngcheck, img_path, result_path),
|
||||
(analyze_steghide, img_path, result_path, submission.password),
|
||||
(analyze_openstego, img_path, result_path, submission.password),
|
||||
(analyze_zsteg, img_path, result_path),
|
||||
(analyze_image_resize, img_path, result_path),
|
||||
(analyze_openstego, img_path, result_path, submission.password),
|
||||
(analyze_pngcheck, img_path, result_path),
|
||||
(analyze_strings, img_path, result_path),
|
||||
(analyze_steghide, img_path, result_path, submission.password),
|
||||
(analyze_zsteg, img_path, result_path),
|
||||
]
|
||||
|
||||
# Deep analysis only
|
||||
@@ -99,8 +127,11 @@ def analyze_image(submission_hash: str) -> None:
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
submission.status = "completed"
|
||||
except Exception:
|
||||
submission.status = "error"
|
||||
submission.status = "completed" # type: ignore
|
||||
except Exception as e:
|
||||
sentry_sdk.capture_exception(e)
|
||||
submission.status = "error" # type: ignore
|
||||
finally:
|
||||
db.session.commit() # pylint: disable=no-member
|
||||
# Flush Sentry events - wait up to 5 seconds for events to be sent
|
||||
sentry_sdk.flush(timeout=5)
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
# flake8: noqa: E203,E501,W503
|
||||
# pylint: disable=C0413,W0718,R0903,R0801
|
||||
# mypy: disable-error-code=unused-awaitable
|
||||
"""Aperi'Solve WSGI runner."""
|
||||
|
||||
from .app import app as application
|
||||
|
||||
+7
-6
@@ -17,10 +17,11 @@ services:
|
||||
worker:
|
||||
condition: service_started
|
||||
env_file:
|
||||
- .env
|
||||
- .env.example
|
||||
environment:
|
||||
- FLASK_DEBUG=true
|
||||
- FLASK_ENV=development
|
||||
- CLEAR_AT_RESTART=0 # Don't clear on Flask debug auto-reload
|
||||
|
||||
worker:
|
||||
#image: ghcr.io/zeecka/aperisolve:latest
|
||||
@@ -34,17 +35,18 @@ services:
|
||||
- redis
|
||||
- postgres
|
||||
env_file:
|
||||
- .env
|
||||
- .env.example
|
||||
environment:
|
||||
- FLASK_DEBUG=true
|
||||
- FLASK_ENV=development
|
||||
- CLEAR_AT_RESTART=0 # Always false for worker to prevent double clearing
|
||||
|
||||
redis:
|
||||
image: redis:7
|
||||
container_name: redis
|
||||
restart: always
|
||||
env_file:
|
||||
- .env
|
||||
- .env.example
|
||||
|
||||
postgres:
|
||||
image: postgres:14
|
||||
@@ -58,8 +60,7 @@ services:
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
env_file:
|
||||
- .env
|
||||
|
||||
- .env.example
|
||||
|
||||
rqdashboard:
|
||||
image: eoranged/rq-dashboard
|
||||
@@ -70,7 +71,7 @@ services:
|
||||
depends_on:
|
||||
- redis
|
||||
env_file:
|
||||
- .env
|
||||
- .env.example
|
||||
|
||||
volumes:
|
||||
pgdata:
|
||||
+10
-2
@@ -17,6 +17,12 @@ services:
|
||||
condition: service_started
|
||||
env_file:
|
||||
- .env
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "dd if=/dev/urandom of=/tmp/healthcheck.png bs=1 count=1000000 && curl -X POST -F 'image=@/tmp/healthcheck.png' http://localhost:5000/upload -s -o /dev/null -w '%{http_code}' | grep -q 200"]
|
||||
interval: 3600s
|
||||
timeout: 30s
|
||||
retries: 3
|
||||
start_period: 900s
|
||||
|
||||
worker:
|
||||
image: ghcr.io/zeecka/aperisolve:prod
|
||||
@@ -31,13 +37,15 @@ services:
|
||||
- postgres
|
||||
env_file:
|
||||
- .env
|
||||
environment:
|
||||
- CLEAR_AT_RESTART=0 # Always false for worker to prevent double clearing
|
||||
|
||||
postgres:
|
||||
image: postgres:14
|
||||
image: postgres:18
|
||||
container_name: postgres
|
||||
restart: always
|
||||
volumes:
|
||||
- pgdata:/var/lib/postgresql/data
|
||||
- pgdata:/var/lib/postgresql
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U aperiuser -d aperisolve"]
|
||||
interval: 5s
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
# Adding a New Analyzer
|
||||
|
||||
This guide shows you how to add a new image analysis tool to the system.
|
||||
|
||||
## Quick Start
|
||||
|
||||
1. Create a new file in `analyzers/` from template file `template_analyzer.py`.
|
||||
2. Register it in `workers.py`
|
||||
|
||||
## Basic Template
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
class MyToolAnalyzer(SubprocessAnalyzer):
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("mytool", input_img, output_dir)
|
||||
self.cmd = ["mytool", self.img]
|
||||
|
||||
def analyze_mytool(input_img: Path, output_dir: Path) -> None:
|
||||
analyzer = MyToolAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
```
|
||||
|
||||
You can also see `template_analyzer.py`.
|
||||
|
||||
## Common Scenarios
|
||||
|
||||
### Simple Text Output Tool
|
||||
|
||||
Like `strings` or `exiftool` - just reads the file and outputs text:
|
||||
|
||||
```python
|
||||
class StringsAnalyzer(SubprocessAnalyzer):
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("strings", input_img, output_dir)
|
||||
self.cmd = ["strings", self.img]
|
||||
```
|
||||
|
||||
### Tool That Extracts Files
|
||||
|
||||
Like `binwalk` or `foremost` - extracts files that get zipped for download:
|
||||
|
||||
```python
|
||||
class ForemostAnalyzer(SubprocessAnalyzer):
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("foremost", input_img, output_dir, has_archive=True)
|
||||
self.cmd = ["foremost", "-o", str(self.get_extracted_dir()), "-i", self.img]
|
||||
```
|
||||
|
||||
Note the `has_archive=True` parameter.
|
||||
|
||||
### Tool With Password Support
|
||||
|
||||
Like `steghide` or `openstego`:
|
||||
|
||||
```python
|
||||
class OutguessAnalyzer(SubprocessAnalyzer):
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("outguess", input_img, output_dir, has_archive=True)
|
||||
|
||||
def build_cmd(self, password: Optional[str] = None) -> list[str]:
|
||||
out = str(self.get_extracted_dir() / "outguess.data")
|
||||
if password:
|
||||
return ["outguess", "-k", password, "-r", self.img, out]
|
||||
return ["outguess", "-r", self.img, out]
|
||||
|
||||
def analyze_outguess(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
|
||||
analyzer = OutguessAnalyzer(input_img, output_dir)
|
||||
if password:
|
||||
analyzer.analyze(password)
|
||||
else:
|
||||
analyzer.analyze()
|
||||
```
|
||||
|
||||
## Customizing Behavior
|
||||
|
||||
### Custom Error Detection
|
||||
|
||||
Override `is_error()` to define what counts as an error:
|
||||
|
||||
```python
|
||||
def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
|
||||
# Only error if no files were extracted
|
||||
return len(stderr) > 0 and not zip_exist
|
||||
```
|
||||
|
||||
### Custom Output Formatting
|
||||
|
||||
Override `process_output()` to format the results:
|
||||
|
||||
```python
|
||||
def process_output(self, stdout: str, stderr: str) -> dict[str, str]:
|
||||
# Parse into key-value pairs
|
||||
metadata = {}
|
||||
for line in stdout.split("\n"):
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
metadata[key.strip()] = value.strip()
|
||||
return metadata
|
||||
```
|
||||
|
||||
### User-Friendly Error Messages
|
||||
|
||||
Override `process_error()` to provide helpful messages:
|
||||
|
||||
```python
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
if "not supported" in stdout:
|
||||
return "This file format is not supported (PNG only)"
|
||||
return stderr
|
||||
```
|
||||
|
||||
## Registering Your Analyzer
|
||||
|
||||
Add it to `workers.py` in the `analyze_image` function:
|
||||
|
||||
```python
|
||||
# Import at the top
|
||||
from .analyzers.mytool import analyze_mytool
|
||||
|
||||
# Add to analyzers list
|
||||
analyzers = [
|
||||
(analyze_binwalk, img_path, result_path),
|
||||
(analyze_exiftool, img_path, result_path),
|
||||
# ... other analyzers ...
|
||||
(analyze_mytool, img_path, result_path),
|
||||
]
|
||||
|
||||
# If it needs a password, pass submission.password:
|
||||
# (analyze_mytool, img_path, result_path, submission.password),
|
||||
|
||||
# If only for deep analysis, add it conditionally:
|
||||
if submission.deep_analysis:
|
||||
analyzers.extend([
|
||||
(analyze_mytool, img_path, result_path),
|
||||
])
|
||||
```
|
||||
|
||||
## Available Methods to Override
|
||||
|
||||
| Method | Purpose | Default Behavior |
|
||||
|--------|---------|------------------|
|
||||
| `build_cmd()` | Build command to run | Uses `self.cmd` |
|
||||
| `is_error()` | Check if result is error | Returns `True` if stderr exists |
|
||||
| `process_output()` | Format success output | Returns stdout as list of lines |
|
||||
| `process_error()` | Format error message | Returns stderr as-is |
|
||||
| `process_note()` | Add informational note | Returns `None` |
|
||||
| `get_extracted_dir()` | Set extraction directory | Returns `output_dir/toolname/` |
|
||||
|
||||
## Output Format
|
||||
|
||||
Your analyzer produces JSON in `results.json`:
|
||||
|
||||
**Success:**
|
||||
```json
|
||||
{
|
||||
"mytool": {
|
||||
"status": "ok",
|
||||
"output": "Tool output here",
|
||||
"download": "/download/hash/mytool"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Error:**
|
||||
```json
|
||||
{
|
||||
"mytool": {
|
||||
"status": "error",
|
||||
"error": "Error message"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Tips
|
||||
|
||||
- Use `self.img` in commands (relative path), not `self.input_img`
|
||||
- Set `has_archive=True` if your tool extracts files
|
||||
- Tools run in parallel threads - no need to worry about concurrency
|
||||
- The base class handles timeouts (10 minutes default)
|
||||
- Extracted files are automatically zipped into `.7z` archives
|
||||
- All exceptions are caught and logged to Sentry
|
||||
- Keep analyzers idempotent and write outputs to the provided `results_dir`
|
||||
- Return structured JSON so the frontend can render links/downloads automatically
|
||||
|
||||
## Example: Complete Analyzer
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
from .base_analyzer import SubprocessAnalyzer
|
||||
|
||||
class PngcheckAnalyzer(SubprocessAnalyzer):
|
||||
def __init__(self, input_img: Path, output_dir: Path) -> None:
|
||||
super().__init__("pngcheck", input_img, output_dir)
|
||||
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" in stdout
|
||||
|
||||
def process_error(self, stdout: str, stderr: str) -> str:
|
||||
if "neither a PNG or JNG" in stdout:
|
||||
return "File format not supported (PNG/JNG only)"
|
||||
return stdout
|
||||
|
||||
def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
|
||||
analyzer = PngcheckAnalyzer(input_img, output_dir)
|
||||
analyzer.analyze()
|
||||
```
|
||||
|
||||
That's it! Your analyzer will now run in parallel with the others and results will appear in the web interface.
|
||||
Reference in New Issue
Block a user