Z-Sans uses a modular design: core functionality is separated from tool implementations, the engine handles scheduling, breeders handle discovery, and tools execute the concrete external calls.
Project Structure¶
flowchart TD
root["Z-Sans/"]
main["main.py<br/>CLI, BreedingEngine main control, plugin loading"]
webapp["webapp.py<br/>Web console backend (HTTP + SSE)"]
front["web_frontend.py<br/>Web frontend SPA (Vue 3)"]
static["web_static/<br/>Vue runtime static assets"]
config["breeding-config.yaml<br/>Default configuration"]
templates["templates/<br/>Full reference config for breeding-config.yaml"]
core["core/"]
eng["core/zsans_engine.py<br/>Asset model, queue, asset graph"]
out["core/output.py<br/>JSON/CSV/GraphML/HTML"]
i18n["core/i18n.py<br/>gettext internationalization"]
breeders["core/breeders/<br/>Breeders + factory"]
tools["core/tools/<br/>ToolOrchestrator"]
assets["assets/<br/>JSfinder/dnsxs/free-subfinder/port"]
ehole["tools/ehole/<br/>Multi-arch fingerprint binaries"]
plugins["plugins/<br/>Plugin directory"]
lang["i18n/<br/>en / zh_CN language packs"]
outdir["output/<br/>Timestamped output directory"]
root --> main & webapp & front & static & config & templates & core & assets & ehole & plugins & lang & outdir
webapp --> front
core --> eng & out & i18n & breeders & tools
Runtime Data Flow¶
flowchart TD
main["main() seeds (-d/-u)"] --> addseed["add_seed()"]
addseed --> assetgraph["AssetGraph.add_asset()<br/>New asset into store (dedup)"]
addseed --> queue["PriorityBreedingQueue.add()"]
assetgraph --> run["run() → start() → _concurrent_breed()"]
queue --> run
run --> pool["ThreadPoolExecutor(max_tasks)"]
pool --> getnext["queue.get_next(strategy)"]
getnext --> process["_process_asset(asset)"]
process --> check["Depth/type/resource/exclusion checks"]
check -->|fail| excluded["excluded"]
check -->|pass| breeder["BreederFactory.get_breeder(type)"]
breeder --> exec_node["breeder.execute(asset, ToolOrchestrator)"]
exec_node --> tools2["External tool calls / internal implementations"]
exec_node --> newloop["New asset loop"]
newloop --> newasset["add_asset → add_edge(discovered)"]
newasset --> hook["emit_hook(on_asset_discovered)"]
hook --> qadd["queue.add()"]
exec_node --> scannedhook["emit_hook(on_asset_scanned)"]
qadd --> queue
process --> empty{"Queue empty?"}
empty -->|no| pool
empty -->|yes| done["completed → stop()"]
done --> ckpt["save_checkpoint()"]
done --> gen["OutputHandler.generate_output()"]
Module Responsibilities¶
main.py — engine main control + entry point¶
BreedingEngine: scan lifecycle state machine (initialized / running / completed / stopped / failed), signal handling, event bus, checkpoints, metrics- CLI parsing and i18n initialization
- Plugin loader: directory scanning,
on_*hook detection, manifest parsing, conflict handling,register_cli --watchmonitoring loop
core/zsans_engine.py — data model¶
Assetclass hierarchy and normalizationPriorityBreedingQueue: strategy-based queueAssetGraph: thread-safe topological storage and exportDEFAULT_CONFIG: built-in default configuration
core/output.py — output¶
OutputHandler: creates the timestamped directory, generates JSON / two CSVs / GraphML / self-contained HTML reportrun_direxposes the current run's output directory to plugins and other modules
core/tools/tools.py — tool orchestration¶
ToolOrchestrator: tool detection, execution, result parsing, fallback, concurrency control, fingerprinting (EHole/WhatWeb), plugin tool registration
core/breeders/breeders.py — breeders¶
BreederFactory: dispatches by asset type- Each breeder invokes a set of tools to produce new assets
core/i18n.py — internationalization¶
- gettext stack: language loading,
_()translation function, current language query
webapp.py + web_frontend.py — Web console¶
- Standard-library HTTP server + SSE real-time logs
- Single-file Vue 3 SPA with local static assets
Extension Points¶
1. Add a new asset type¶
- Define a subclass and register a factory in
core/zsans_engine.py - Add a breeder in
core/breeders/breeders.py - Enable the toggle and priority in the
asset_typesconfig - Extend columns and colors in the output module as needed
2. Add a new tool¶
- Add a
run_<name>method incore/tools/tools.py(detect → execute → parse) - Or call
register_tool(name, path, version, extra_args)from a plugin to register/override an external tool - Configure paths and concurrency in
external_tools.paths/concurrency.tools
3. Add new behavior — plugin system (recommended)¶
No changes to the main program needed; implement on_* hooks to hook into the scan lifecycle. See Plugin Development for details.
Threading Model¶
| Component | Threads |
|---|---|
| Asset processing | ThreadPoolExecutor(concurrency.max_tasks), thread prefix zsans |
| Tool execution | Submitted to the thread pool via submit_task(), each tool has its own independent concurrency limit |
| Web tasks | One daemon thread per task, zsans-web-<id> |
| Signals | SIGINT/SIGTERM graceful shutdown registered in CLI mode; not registered in Web/background mode (register_signals=False) |
Shutdown Chain¶
flowchart TD
sig["SIGINT / POST stop / Ctrl+C"]
req["_stop_requested = True"]
state["config.state = 'stopped'"]
shut["executor.shutdown(wait=False, cancel_futures=True)"]
ckpt["save_checkpoint()"]
gen["output_handler.generate_output()"]
tools["tool_orchestrator.shutdown()"]
sig --> req --> state --> shut
shut --> ckpt --> gen --> tools
watch and stop coupling
stop() deliberately does not set the global _stop_signaled, otherwise --watch would misinterpret it as "signal received" after the first round and exit immediately. The flag is only ever set by the real signal handler.