-
Notifications
You must be signed in to change notification settings - Fork 166
New Scanner
Add a custom subprocess-based analyzer in five steps. The v5 base class (BaseSubprocessAnalyzer) does the heavy lifting — you implement output parsing and declare the tool's contract via class attributes. No need to hand-roll subprocess.Popen glue.
If your scanner does its own thing (no external CLI tool — pure-Python analysis on bytes / PE structure / etc.), inherit from BaseAnalyzer directly and implement analyze + cleanup yourself. The driver-style example at the bottom of this page covers that shape.
BaseAnalyzer (abstract)
│
└── BaseSubprocessAnalyzer (template-method base for CLI-tool wrappers)
│
├── YaraStaticAnalyzer / YaraDynamicAnalyzer
├── CheckPlzAnalyzer
├── PESieveAnalyzer / MonetaAnalyzer / PatriotAnalyzer
├── HSBAnalyzer / RedEdrAnalyzer
└── (your new analyzer)
BaseSubprocessAnalyzer lives in app/analyzers/base.py. It:
- Reads tool config from
self.config['analysis'][<section>][<name>]. - Builds a command via
cfg['command'].format(...)with named keyword args. - Runs the subprocess with timeout + optional cwd.
- Calls your
_parse_output(stdout)to convert raw output → structured findings. - Wraps the result in a standard envelope (
{status, findings, errors}).
You declare the tool's contract via class-level attributes; the base class does the rest.
| Section | Target | When to pick |
|---|---|---|
static |
File path | Operates on bytes on disk, no execution required |
dynamic |
PID | Targets a running process by PID |
Static analyzers are run sequentially in parallel (thread pool) over an uploaded file. Dynamic analyzers run in parallel against the spawned payload's PID, except HSB which runs serially after the parallel batch (see _SERIAL_DYNAMIC_ANALYZERS in manager.py).
# app/analyzers/static/myanalyzer.py (or app/analyzers/dynamic/...)
from ..base import BaseSubprocessAnalyzer
class MyAnalyzer(BaseSubprocessAnalyzer):
tool_section = 'static' # or 'dynamic'
tool_name = 'myanalyzer' # must match the config key
target_kwarg = 'file_path' # 'file_path' for static, 'pid' for dynamic
abspath_targets = True # apply abspath() to tool_path + target before formatting
def _parse_output(self, stdout):
"""Convert raw tool stdout into a serializable dict.
Return whatever shape makes sense for your tool — a list of findings,
a nested dict, etc. The frontend / report templates consume this from
`findings` in the envelope, so design it for that consumer.
"""
findings = {'matches': []}
for line in stdout.splitlines():
line = line.strip()
if line.startswith('MATCH:'):
findings['matches'].append(line[len('MATCH:'):].strip())
return findingsThat's the whole analyzer. Some hooks you can optionally override:
| Hook | Purpose |
|---|---|
_preprocess_stdout(stdout) |
Strip ANSI / dedupe / chunk before parsing |
_postprocess_findings(findings) |
Enrich / sort / aggregate after parsing |
_build_envelope(...) |
Custom result shape (rarely needed) |
_get_cwd(cfg) |
Return a custom working directory for the subprocess |
_on_timeout(cfg), _on_error(exc)
|
Customize failure responses |
Class-level attributes you can set:
| Attribute | Default | Purpose |
|---|---|---|
tool_section |
— | Required: 'static' or 'dynamic'
|
tool_name |
— | Required: matches the Config/config.yaml key |
target_kwarg |
'pid' |
Format-string keyword: 'file_path' / 'pid' / 'directory'
|
extra_format_kwargs |
() |
Additional config keys forwarded to command.format() (e.g. ('rules_path',) for YARA) |
abspath_targets |
False |
Run os.path.abspath() on tool_path and string targets |
use_tool_path_as_cwd |
False |
Run subprocess with cwd=dirname(tool_path)
|
use_timeout |
True |
Set False for streaming-output tools that should run open-ended |
# app/analyzers/manager.py
from .static.myanalyzer import MyAnalyzer
class AnalysisManager:
STATIC_ANALYZERS = {
'yara': YaraStaticAnalyzer,
'checkplz': CheckPlzAnalyzer,
'stringnalyzer': StringsAnalyzer,
'myanalyzer': MyAnalyzer, # <- add here
}
# OR for dynamic:
DYNAMIC_ANALYZERS = {
'yara': YaraDynamicAnalyzer,
'pe_sieve': PESieveAnalyzer,
# ...
'myanalyzer': MyAnalyzer, # <- add here
}The dict key MUST match tool_name and the config key (Step 4).
# Config/config.yaml
analysis:
static: # or dynamic:
myanalyzer:
enabled: true
tool_path: ".\\Scanners\\MyTool\\mytool.exe"
command: "{tool_path} --format json {file_path}"
timeout: 120enabled: false keeps the analyzer registered but disables it at run time — cleaner than commenting out the block.
For YARA-style analyzers that need additional inputs ({rules_path}):
class YaraStaticAnalyzer(BaseSubprocessAnalyzer):
tool_section = 'static'
tool_name = 'yara'
target_kwarg = 'file_path'
extra_format_kwargs = ('rules_path',) # forwarded from config to command.format()yara:
enabled: true
tool_path: ".\\Scanners\\Yara\\yara64.exe"
rules_path: ".\\Scanners\\Yara\\LitterBox.yar"
command: "{tool_path} -s -m {rules_path} {file_path}"
timeout: 120This is optional but is what makes the analyzer visible to operators.
Frontend tab. Each scanner has a JS module under app/static/js/results/tools/<scanner>.js that renders its sub-tab on the dynamic / static results page. Pattern: a render function that consumes the findings dict from your analyzer's envelope. See app/static/js/results/tools/_shared.js for shared helpers (severity tags, table builders, expand-on-click behavior).
Risk score. If your analyzer should bump the Detection Score Explained number, add a contribution in app/utils/risk_analyzer.py. Add a small helper called from _calculate_dynamic_risk() (or _calculate_static_risk()) that reads your tool's findings and returns a (score, factors) pair. Cap your contribution so a single noisy analyzer can't dominate the total.
Some analyzers don't wrap a CLI tool — they read the file directly. Inherit from BaseAnalyzer directly:
# app/analyzers/static/hash_analyzer.py
import hashlib
import os
from ..base import BaseAnalyzer
class HashAnalyzer(BaseAnalyzer):
def analyze(self, file_path):
try:
cfg = self.config['analysis']['static'].get('hash_analyzer', {})
block_size = cfg.get('block_size', 65536)
md5, sha1, sha256 = hashlib.md5(), hashlib.sha1(), hashlib.sha256()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(block_size), b''):
for h in (md5, sha1, sha256):
h.update(chunk)
self.results = {
'status': 'completed',
'findings': {
'hashes': {
'md5': md5.hexdigest(),
'sha1': sha1.hexdigest(),
'sha256': sha256.hexdigest(),
},
'file_size': os.path.getsize(file_path),
},
'errors': None,
}
except Exception as e:
self.results = {'status': 'error', 'error': str(e)}
def cleanup(self):
pass# Config/config.yaml
analysis:
static:
hash_analyzer:
enabled: true
block_size: 65536# app/analyzers/manager.py
STATIC_ANALYZERS = {
# ...
'hash_analyzer': HashAnalyzer,
}Every analyzer's self.results should be one of these shapes — the rest of the system (rendering, scoring, report generation) expects it:
# success
{'status': 'completed', 'findings': {...}, 'errors': None}
# failed (subprocess returned non-zero)
{'status': 'failed', 'findings': {...}, 'errors': 'stderr text or None'}
# timeout
{'status': 'timeout', 'error': 'Analysis timed out after 120 seconds'}
# generic error
{'status': 'error', 'error': 'exception message'}BaseSubprocessAnalyzer builds this shape automatically. If you handcraft it, match the keys exactly.
- Application Architecture — analyzer pipeline and saved-JSON layout
- Configuration Reference — every existing analyzer's config block
- Detection Score Explained — how to wire a new analyzer into the score
-
YARA Rules Management — concrete example of
extra_format_kwargsusage -
app/analyzers/base.py— fullBaseSubprocessAnalyzersource