From 16093285d33529d856d5e8808785d4f7dea0e228 Mon Sep 17 00:00:00 2001 From: "Mr. The Plague" Date: Sat, 18 Jul 2026 06:35:08 -0400 Subject: [PATCH 1/2] feat(sessions): bulk clear reverse-shell noise + delete API Explain/ops: public reverse_shell ports attract scanners that briefly register as shells. Add POST /sessions/clear, DELETE /sessions/{id}, CLI clear/delete, and ops UI buttons. Close/clear drop TCP channels. --- src/squidc5/api/routes.py | 51 ++++++++++++++++++++ src/squidc5/cli.py | 53 ++++++++++++++++++++- src/squidc5/policy/engine.py | 1 + src/squidc5/sessions/manager.py | 84 ++++++++++++++++++++++++++++++++- tests/test_sessions_clear.py | 64 +++++++++++++++++++++++++ web/ops-admin.js | 22 +++++++++ 6 files changed, 273 insertions(+), 2 deletions(-) create mode 100644 tests/test_sessions_clear.py diff --git a/src/squidc5/api/routes.py b/src/squidc5/api/routes.py index 5087f17..7aea28e 100644 --- a/src/squidc5/api/routes.py +++ b/src/squidc5/api/routes.py @@ -349,6 +349,57 @@ def build_api_router() -> APIRouter: n = await state.sessions.close_orphaned_shells(probe=probe) return {"closed": n, "probed": probe} + @api.post("/sessions/clear") + async def clear_sessions( + request: Request, + auth: AuthContext = Depends(require_scope("sessions:write", "admin")), + ) -> dict[str, Any]: + """Bulk remove reverse-shell/tcp noise (scanners on public ports).""" + state = get_state(request) + body: dict[str, Any] = {} + try: + body = await request.json() + except Exception: + body = {} + unverified_only = bool(body.get("unverified_only", True)) + closed_only = bool(body.get("closed_only", False)) + active_only = bool(body.get("active_only", False)) + delete = bool(body.get("delete", True)) + if body.get("all_shells"): + unverified_only = False + closed_only = False + active_only = False + decision = await state.policy.check_and_audit( + auth, + "sessions.clear", + extra={ + "unverified_only": unverified_only, + "closed_only": closed_only, + "delete": delete, + }, + ) + if not decision.allowed: + raise HTTPException(403, decision.reason) + result = await state.sessions.clear_shells( + unverified_only=unverified_only, + closed_only=closed_only, + active_only=active_only, + delete=delete, + ) + return result + + @api.delete("/sessions/{session_id}") + async def delete_session( + session_id: str, + request: Request, + auth: AuthContext = Depends(require_scope("sessions:write", "admin")), + ) -> dict[str, str]: + state = get_state(request) + ok = await state.sessions.delete(session_id) + if not ok: + raise HTTPException(404, "session not found") + return {"status": "deleted", "id": session_id} + # ----- Tasks ----- @api.get("/tasks") diff --git a/src/squidc5/cli.py b/src/squidc5/cli.py index 9dee8a5..955d954 100644 --- a/src/squidc5/cli.py +++ b/src/squidc5/cli.py @@ -211,6 +211,24 @@ def cmd_sessions_close(args: argparse.Namespace, client: Client) -> None: pp(client.post(f"/api/v1/sessions/{args.id}/close")) +def cmd_sessions_delete(args: argparse.Namespace, client: Client) -> None: + pp(client.delete(f"/api/v1/sessions/{args.id}")) + + +def cmd_sessions_clear(args: argparse.Namespace, client: Client) -> None: + """Bulk remove reverse-shell noise (default: unverified only).""" + body: dict[str, Any] = { + "delete": not getattr(args, "close_only", False), + "unverified_only": not getattr(args, "all_shells", False), + "closed_only": bool(getattr(args, "closed", False)), + "active_only": bool(getattr(args, "active", False)), + "all_shells": bool(getattr(args, "all_shells", False)), + } + if body["closed_only"] or body["active_only"]: + body["unverified_only"] = bool(getattr(args, "unverified", False)) + pp(client.post("/api/v1/sessions/clear", json=body)) + + def cmd_sessions_reap(args: argparse.Namespace, client: Client) -> None: # Exec probe can take a bit with many sessions — raise timeout old = client._client.timeout @@ -797,9 +815,42 @@ def build_parser() -> argparse.ArgumentParser: s_get = ses_sub.add_parser("get", help="Get session") s_get.add_argument("id") s_get.set_defaults(func=cmd_sessions_get, needs_client=True) - s_close = ses_sub.add_parser("close", help="Close session") + s_close = ses_sub.add_parser("close", help="Close session (keep row)") s_close.add_argument("id") s_close.set_defaults(func=cmd_sessions_close, needs_client=True) + s_del = ses_sub.add_parser("delete", help="Hard-delete one session") + s_del.add_argument("id") + s_del.set_defaults(func=cmd_sessions_delete, needs_client=True) + s_clear = ses_sub.add_parser( + "clear", + help="Bulk remove reverse-shell noise (default: unverified scanners)", + ) + s_clear.add_argument( + "--all-shells", + action="store_true", + help="Remove ALL reverse_shell/tcp sessions (verified too)", + ) + s_clear.add_argument( + "--closed", + action="store_true", + help="Only purge already-closed shell rows", + ) + s_clear.add_argument( + "--active", + action="store_true", + help="Only active shells", + ) + s_clear.add_argument( + "--unverified", + action="store_true", + help="With --closed/--active: only unverified", + ) + s_clear.add_argument( + "--close-only", + action="store_true", + help="Mark closed instead of hard-delete", + ) + s_clear.set_defaults(func=cmd_sessions_clear, needs_client=True) s_reap = ses_sub.add_parser( "reap", help="Close dead reverse shells (no TCP, or fail exec probe)", diff --git a/src/squidc5/policy/engine.py b/src/squidc5/policy/engine.py index ed794a9..cee1eb5 100644 --- a/src/squidc5/policy/engine.py +++ b/src/squidc5/policy/engine.py @@ -46,6 +46,7 @@ DEFAULT_POLICY: dict[str, Any] = { "action_risk": { "sessions.list": 0, "sessions.get": 0, + "sessions.clear": 2, "tasks.list": 0, "tasks.create": 4, "listeners.create": 5, diff --git a/src/squidc5/sessions/manager.py b/src/squidc5/sessions/manager.py index c314a88..809fd38 100644 --- a/src/squidc5/sessions/manager.py +++ b/src/squidc5/sessions/manager.py @@ -71,16 +71,98 @@ class SessionManager: await self.db.update_session(session_id, **fields) await self.metrics.emit("session.heartbeat", {"id": session_id}) - async def close(self, session_id: str) -> None: + async def close(self, session_id: str, *, drop: bool = True) -> None: + if drop and self.drop_channel is not None: + try: + await self.drop_channel(session_id) + except Exception: + pass await self.db.update_session(session_id, status="closed") async with self._lock: self._live.pop(session_id, None) await self.metrics.emit("session.closed", {"id": session_id}) + async def delete(self, session_id: str, *, drop: bool = True) -> bool: + """Hard-remove session row (and drop TCP if any).""" + if drop and self.drop_channel is not None: + try: + await self.drop_channel(session_id) + except Exception: + pass + async with self._lock: + self._live.pop(session_id, None) + ok = await self.db.delete_session(session_id) + if ok: + await self.metrics.emit("session.deleted", {"id": session_id}) + return ok + + async def clear_shells( + self, + *, + unverified_only: bool = False, + closed_only: bool = False, + active_only: bool = False, + delete: bool = True, + kinds: tuple[str, ...] = ("reverse_shell", "tcp"), + ) -> dict[str, int]: + """ + Bulk remove reverse-shell/tcp sessions (scanner noise cleanup). + + unverified_only: only shells that are not exec-verified + closed_only: only status=closed + active_only: only status=active + delete: hard-delete rows (default); if False, mark closed + """ + if closed_only: + rows = await self.db.list_sessions(status="closed") + elif active_only: + rows = await self.db.list_sessions(status="active") + else: + rows = await self.db.list_sessions(status=None) + removed = 0 + closed = 0 + for row in rows: + if row.get("kind") not in kinds: + continue + sid = str(row["id"]) + status = row.get("status") or "" + if closed_only and status != "closed": + continue + if active_only and status != "active": + continue + if unverified_only: + verified = False + if self.verified_check is not None: + verified = bool(self.verified_check(sid)) + meta = row.get("metadata") + if isinstance(meta, str): + try: + meta = json.loads(meta) + except json.JSONDecodeError: + meta = {} + if isinstance(meta, dict) and (meta.get("verified") or meta.get("exec_ok")): + verified = True + if verified: + continue + if delete: + if await self.delete(sid, drop=True): + removed += 1 + else: + await self.close(sid, drop=True) + closed += 1 + if removed: + await self.metrics.incr("sessions.cleared", float(removed)) + return {"removed": removed, "closed": closed} + async def reject(self, session_id: str, reason: str) -> None: """Remove false-positive / scanner sessions entirely.""" async with self._lock: self._live.pop(session_id, None) + if self.drop_channel is not None: + try: + await self.drop_channel(session_id) + except Exception: + pass await self.db.delete_session(session_id) await self.metrics.incr("sessions.rejected") await self.metrics.emit( diff --git a/tests/test_sessions_clear.py b/tests/test_sessions_clear.py new file mode 100644 index 0000000..e042645 --- /dev/null +++ b/tests/test_sessions_clear.py @@ -0,0 +1,64 @@ +import pytest + + +@pytest.mark.asyncio +async def test_sessions_clear_unverified(client, admin_headers, tmp_path): + # create reverse_shell-looking rows via API isn't easy without TCP; + # use DB through register path: implant beacon is different kind. + # Seed sessions by calling create via internal store after login. + from pathlib import Path + + # Create shell sessions via reverse_shell listener + raw connect that goes quiet + lr = await client.post( + "/api/v1/listeners", + headers=admin_headers, + json={"name": "shell-clear-test", "kind": "reverse_shell", "port": 19044}, + ) + assert lr.status_code == 200 + lid = lr.json()["id"] + st = await client.post(f"/api/v1/listeners/{lid}/start", headers=admin_headers) + assert st.status_code == 200 + + import asyncio + + # silent connect — creates session until probe drops it + reader, writer = await asyncio.open_connection("127.0.0.1", 19044) + await asyncio.sleep(0.6) + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + await asyncio.sleep(0.3) + # clear unverified (and closed leftovers) + clr = await client.post( + "/api/v1/sessions/clear", + headers=admin_headers, + json={"unverified_only": False, "all_shells": True, "delete": True}, + ) + assert clr.status_code == 200 + assert "removed" in clr.json() + + rows = await client.get( + "/api/v1/sessions?status=all&kind=reverse_shell,tcp", + headers=admin_headers, + ) + shells = [r for r in rows.json() if r.get("kind") in ("reverse_shell", "tcp")] + # all shells from this test should be gone + assert not any(r.get("listener_id") == lid for r in shells) + + await client.post(f"/api/v1/listeners/{lid}/stop", headers=admin_headers) + + +@pytest.mark.asyncio +async def test_sessions_delete_endpoint(client, admin_headers): + b = await client.post( + "/api/v1/implant/beacon", + json={"hostname": "clear-me", "username": "u"}, + ) + sid = b.json()["session_id"] + d = await client.delete(f"/api/v1/sessions/{sid}", headers=admin_headers) + assert d.status_code == 200 + g = await client.get(f"/api/v1/sessions/{sid}", headers=admin_headers) + assert g.status_code == 404 diff --git a/web/ops-admin.js b/web/ops-admin.js index 801cdce..3953f7d 100644 --- a/web/ops-admin.js +++ b/web/ops-admin.js @@ -89,6 +89,8 @@ ${hint("Every implant or reverse-shell connection the server tracks (beacons, shells, closed). Reap dead probes and drops mute/zombie shells. Close selected uses the Shell dropdown target. List all dumps the full session table for forensics.", "sessions")}
${can("sessions:write") ? '' : ""} + ${can("sessions:write") ? '' : ""} + ${can("sessions:write") ? '' : ""} ${can("sessions:write") ? '' : ""}
@@ -580,6 +582,26 @@ }; } + if ($("clearUnverifiedBtn")) { + $("clearUnverifiedBtn").onclick = async () => { + if (!confirm("Delete all unverified reverse shells? (keeps verified)")) return; + try { + const res = await api("POST", "/api/v1/sessions/clear", { unverified_only: true, delete: true }); + showOk(`Cleared ${res.removed || 0} unverified shell(s)`); + if (typeof refresh === "function") refresh(); + } catch (e) { showError(e.message || String(e)); } + }; + } + if ($("clearClosedBtn")) { + $("clearClosedBtn").onclick = async () => { + if (!confirm("Hard-delete all closed reverse-shell rows?")) return; + try { + const res = await api("POST", "/api/v1/sessions/clear", { closed_only: true, unverified_only: false, delete: true }); + showOk(`Purged ${res.removed || 0} closed shell(s)`); + if (typeof refresh === "function") refresh(); + } catch (e) { showError(e.message || String(e)); } + }; + } if ($("reapBtn")) { $("reapBtn").onclick = async () => { try { From d4f63060e65f6aa1bfe0209732509bb70d86bf9b Mon Sep 17 00:00:00 2001 From: "Mr. The Plague" Date: Sat, 18 Jul 2026 06:36:57 -0400 Subject: [PATCH 2/2] fix: drop unused Path import in test_sessions_clear --- tests/test_sessions_clear.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/tests/test_sessions_clear.py b/tests/test_sessions_clear.py index e042645..5650289 100644 --- a/tests/test_sessions_clear.py +++ b/tests/test_sessions_clear.py @@ -2,12 +2,7 @@ import pytest @pytest.mark.asyncio -async def test_sessions_clear_unverified(client, admin_headers, tmp_path): - # create reverse_shell-looking rows via API isn't easy without TCP; - # use DB through register path: implant beacon is different kind. - # Seed sessions by calling create via internal store after login. - from pathlib import Path - +async def test_sessions_clear_unverified(client, admin_headers): # Create shell sessions via reverse_shell listener + raw connect that goes quiet lr = await client.post( "/api/v1/listeners",