"""Tests for the pentest path: the sandbox network profile and the tool catalog.

Two units are covered together because they are one permission chain.
``agents.engagement`` and ``agents.pentest_catalog`` decide whether a pentest
tool call may be reconstructed at all; the Bubblewrap profile decides what the
reconstructed call runs inside. ``allow_network`` is the single flag that widens
that profile, and it appeared nowhere in the suite - which is how the profile
shipped with an argument vector ``bwrap`` refuses at run time.

Nothing here opens a socket, starts a scan, or launches Bubblewrap. The argv the
provider builds is read at the single point where it is handed to
``asyncio.create_subprocess_exec``, which the fixture below replaces. That is
the same seam ``tests/test_execution_provider.py::no_process_launch`` already
uses to prove nothing was spawned; no production hook was added to make the
vector observable.
"""

from __future__ import annotations

import asyncio
from pathlib import Path

import pytest

from cyberbrain26.agents.engagement import (
    MAX_ENGAGEMENT_LEVEL,
    TOOL_CLASS_BY_LEVEL,
    TOOL_CLASSES,
    EngagementAuthorization,
    TargetScope,
    TargetScopeError,
)
from cyberbrain26.agents.pentest_catalog import (
    PentestRefusal,
    authorize_tool_call,
    tool_class_for_level,
)
from cyberbrain26.execution_plane.bubblewrap_provider import (
    NETWORK_TMPFS_SIZE_BYTES,
    BubblewrapExecutionProvider,
)
from cyberbrain26.security_plane.audit import AuditEventType, AuditLog, AuditQuery


class _ArgvCaptured(Exception):
    """Raised by the stubbed launcher so a test stops before a process exists."""

    def __init__(self, argv: list[str]) -> None:
        super().__init__("argv captured before launch")
        self.argv = argv


@pytest.fixture
def captured_bwrap_argv(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
    """Return the argument vectors the provider builds, launching nothing.

    The provider hands its argv to ``asyncio.create_subprocess_exec`` and to
    nothing else, so replacing that one call is what lets the vector be read on
    a host with no Bubblewrap installed, no namespace permission and no network.
    The returned list stays empty when the provider refused before reaching the
    launcher.
    """
    recorded: list[list[str]] = []

    async def _record(*args: object, **kwargs: object) -> None:
        argv = [str(arg) for arg in args]
        recorded.append(argv)
        raise _ArgvCaptured(argv)

    monkeypatch.setattr(asyncio, "create_subprocess_exec", _record)
    # The vector is built after the binary lookup, so the provider still has to
    # believe Bubblewrap exists. Nothing is ever run against this path.
    monkeypatch.setattr(
        "cyberbrain26.execution_plane.bubblewrap_provider.shutil.which",
        lambda name: f"/usr/bin/{name}",
    )
    return recorded


@pytest.fixture
def host_resolver(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> str:
    """Point the provider at a resolver file that provably exists.

    The bind is emitted only when the host's resolver is a file, so on a host
    without one the ``allow_network`` profile legitimately skips it. Redirecting
    the module constant at a file this test creates makes the assertions below
    about the profile's *rule* rather than about the machine the suite happens
    to run on.
    """
    resolver = tmp_path / "resolv.conf"
    resolver.write_text("nameserver 127.0.0.53\n", encoding="utf-8")
    monkeypatch.setattr(
        "cyberbrain26.execution_plane.bubblewrap_provider.RESOLVER_PATH", str(resolver)
    )
    return str(resolver)


def _ro_binds(argv: list[str]) -> list[tuple[str, str]]:
    """Return the ``(source, destination)`` of every ``--ro-bind`` in ``argv``."""
    return [
        (argv[index + 1], argv[index + 2])
        for index, token in enumerate(argv)
        if token == "--ro-bind"
    ]


def _limiter_source(argv: list[str]) -> str:
    """Return the rlimit shim the provider execs, located inside ``argv``."""
    return argv[argv.index("-c") + 1]


@pytest.fixture
def engagement() -> EngagementAuthorization:
    """A level-1 recon engagement over exactly one authorised network."""
    return EngagementAuthorization(
        owner="security-reviewer",
        targets=("10.10.10.0/24",),
        max_level=1,
        allowed_tool_classes=frozenset({"recon"}),
        rules_of_engagement=("no raw sockets",),
    )


@pytest.fixture
def audit_log() -> AuditLog:
    """An in-memory audit log; nothing is persisted to the developer's data dir."""
    return AuditLog()


def _assert_refusal_audited(
    audit: AuditLog,
    engagement: EngagementAuthorization,
    reason_code: str,
    task_id: str,
) -> None:
    """Assert a refusal left exactly one attributable audit record."""
    events = audit.query(
        AuditQuery(event_type=AuditEventType.PERMISSION_CHECK, task_id=task_id)
    )
    assert len(events) == 1, "a refusal has to leave exactly one audit record"
    event = events[0]
    assert event.result == "refused"
    assert event.error == reason_code
    assert event.actor == "pentester"
    assert event.agent_id == "pentester"
    assert event.metadata["engagement_digest"] == engagement.digest
    assert event.metadata["detail"]


# ----------------------------------------------------------------------
# The allow_network sandbox profile
#
# The provider builds its argv, then launches. The tests below read the built
# vector and stop at the launch, so the default path and the engagement profile
# can both be pinned exactly, on a host that has neither Bubblewrap nor a
# network namespace available.
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_allow_network_shares_the_namespace_and_ceils_tmpfs_before_declaring_it(
    sandbox_workspaces_root: Path,
    captured_bwrap_argv: list[list[str]],
    host_resolver: str,
) -> None:
    workspace = sandbox_workspaces_root / "workspace"
    workspace.mkdir()
    provider = BubblewrapExecutionProvider(allow_network=True)

    with pytest.raises(_ArgvCaptured):
        await provider.execute("print('never runs')", "python", workspace=str(workspace))

    assert len(captured_bwrap_argv) == 1
    argv = captured_bwrap_argv[0]
    assert argv[0] == "/usr/bin/bwrap"

    # Egress is shared, never unshared-and-guessed.
    assert "--share-net" in argv
    # Both flags, in this order. ``--share-net`` alone is a well-formed vector
    # bwrap accepts, and it drops every unshare directive - the sandbox then
    # reported /proc/1/comm == "systemd" and could read /proc/<pid>/environ of
    # host processes. bwrap applies --share-net to override an EARLIER
    # --unshare-all, so the re-share is expressed as a pair, never as a
    # replacement. The runtime half of this is
    # tests/test_execution_provider.py::test_the_network_profile_keeps_its_own_pid_namespace
    assert "--unshare-all" in argv
    assert argv.index("--unshare-all") < argv.index("--share-net")

    # bwrap reads --size as a modifier of the tmpfs that FOLLOWS it, so the
    # ceiling has to come first. The other order is what bwrap refused at run
    # time ("--size must be followed by --tmpfs"), and it made every
    # allow_network call fail before any tool ran.
    tmpfs_index = argv.index("--tmpfs")
    assert argv[tmpfs_index : tmpfs_index + 2] == ["--tmpfs", "/tmp"]
    assert argv[tmpfs_index - 2 : tmpfs_index] == ["--size", str(NETWORK_TMPFS_SIZE_BYTES)]
    assert argv.index("--size") < tmpfs_index

    # The resolver is the only piece of host configuration bound in, it is bound
    # read-only, and it lands on the same path inside the sandbox.
    assert _ro_binds(argv).count((host_resolver, host_resolver)) == 1

    # Sharing the host's network namespace is the one place parallel forking
    # becomes possible, so the fork ceiling rides along in this profile.
    assert "RLIMIT_NPROC" in _limiter_source(argv)


@pytest.mark.asyncio
async def test_the_default_profile_still_unshares_everything_and_declares_a_plain_tmpfs(
    sandbox_workspaces_root: Path,
    captured_bwrap_argv: list[list[str]],
    host_resolver: str,
) -> None:
    # ``host_resolver`` exists for this test on purpose: the resolver bind is
    # then absent because ``allow_network`` is False, not because the host
    # happens to lack the file. The default vector is pinned at its exact
    # position, which is the path ``test_sandbox_has_no_external_network_interface``
    # already exercises against a real sandbox.
    workspace = sandbox_workspaces_root / "workspace"
    workspace.mkdir()
    provider = BubblewrapExecutionProvider()

    with pytest.raises(_ArgvCaptured):
        await provider.execute("print('never runs')", "python", workspace=str(workspace))

    argv = captured_bwrap_argv[0]
    assert "--unshare-all" in argv
    assert "--share-net" not in argv

    # No ceiling at all: a tmpfs sized for a shared network namespace is not part
    # of the default profile.
    assert "--size" not in argv

    proc_index = argv.index("--proc")
    assert argv[proc_index : proc_index + 4] == ["--proc", "/proc", "--tmpfs", "/tmp"]

    # No resolver, no host configuration, nothing but the read-only runtime.
    assert host_resolver not in argv
    assert host_resolver not in {source for source, _ in _ro_binds(argv)}
    assert ("/usr", "/usr") in _ro_binds(argv)

    # The fork ceiling is a property of the shared namespace, not of the default.
    assert "RLIMIT_NPROC" not in _limiter_source(argv)


@pytest.mark.asyncio
async def test_the_per_call_flag_can_grant_the_profile_but_never_withdraw_it(
    sandbox_workspaces_root: Path,
    captured_bwrap_argv: list[list[str]],
) -> None:
    # The per-call argument exists for ``AuthorizedExecutionService``, which
    # derives it from the authenticated engagement profile. It is combined with
    # ``or``, so a call can grant what the constructor granted but a caller can
    # never take a grant away - the narrower of the two is always the default.
    workspace = sandbox_workspaces_root / "workspace"
    workspace.mkdir()
    granted_by_call = BubblewrapExecutionProvider()
    granted_by_constructor = BubblewrapExecutionProvider(allow_network=True)

    for provider, allow_network in ((granted_by_call, True), (granted_by_constructor, False)):
        with pytest.raises(_ArgvCaptured):
            await provider.execute(
                "print('never runs')", "python", workspace=str(workspace),
                allow_network=allow_network,
            )

    per_call_argv, per_constructor_argv = captured_bwrap_argv
    assert "--share-net" in per_call_argv
    assert "--share-net" in per_constructor_argv
    for argv in (per_call_argv, per_constructor_argv):
        # The grant is monotone, but the namespace pair travels with it: a
        # network grant re-shares ONE namespace, it does not un-isolate the rest.
        assert "--unshare-all" in argv
        assert argv.index("--unshare-all") < argv.index("--share-net")
        tmpfs_index = argv.index("--tmpfs")
        assert argv[tmpfs_index - 2 : tmpfs_index + 2] == [
            "--size",
            str(NETWORK_TMPFS_SIZE_BYTES),
            "--tmpfs",
            "/tmp",
        ]


# ----------------------------------------------------------------------
# Catalog authorization
#
# The model never supplies an argv. These tests pin what the catalog reconstructs
# from a declared intent, and - more importantly - that each refusal is refused
# *and* recorded.
# ----------------------------------------------------------------------
def test_a_scoped_level_1_scan_is_reconstructed_from_the_catalog(
    engagement: EngagementAuthorization,
) -> None:
    result = authorize_tool_call(
        {
            "tool": "nmap",
            "target": "10.10.10.5",
            "args": ["-sT", "-p22,443"],
            "level": 1,
        },
        engagement=engagement,
    )

    # Positive control: the refusals below are specific, not universal.
    assert result.argv == ("/usr/bin/nmap", "-sT", "-p22,443", "10.10.10.5")
    assert result.level == 1
    assert result.action == "scan_network"
    assert result.target == "10.10.10.5"
    assert result.engagement_digest == engagement.digest


@pytest.mark.parametrize(
    ("token", "reason_code"),
    [
        ("-sS", "forbidden_argument"),  # raw sockets need privileges we do not grant
        ("-oNreport.xml", "forbidden_argument"),  # writes outside the workspace
        ("--script=vuln", "forbidden_argument"),  # NSE scripts execute code
        ("--traceroute", "forbidden_argument"),
        ("--not-in-the-catalog", "argument_not_allowed"),
    ],
)
def test_catalog_refuses_an_argument_it_does_not_permit(
    engagement: EngagementAuthorization,
    audit_log: AuditLog,
    token: str,
    reason_code: str,
) -> None:
    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nmap", "target": "10.10.10.5", "args": [token], "level": 1,
             "task_id": "pentest-args"},
            engagement=engagement,
            audit_log=audit_log,
        )

    assert excinfo.value.reason_code == reason_code
    _assert_refusal_audited(audit_log, engagement, reason_code, "pentest-args")


@pytest.mark.parametrize(
    "target",
    [
        "192.168.5.5",  # a different network entirely
        "10.0.0.0/8",  # a mask that would widen the authorised /24
        "elsewhere.example.com",  # a name nobody authorised
    ],
)
def test_catalog_refuses_a_target_outside_the_engagement_scope(
    engagement: EngagementAuthorization,
    audit_log: AuditLog,
    target: str,
) -> None:
    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nmap", "target": target, "args": ["-sT"], "level": 1,
             "task_id": "pentest-scope"},
            engagement=engagement,
            audit_log=audit_log,
        )

    assert excinfo.value.reason_code == "target_out_of_scope"
    assert target in str(excinfo.value)
    _assert_refusal_audited(audit_log, engagement, "target_out_of_scope", "pentest-scope")


def test_the_scope_the_engagement_derives_is_the_one_the_target_is_checked_against(
    engagement: EngagementAuthorization,
) -> None:
    # The whitelist is a value with its own parser, consulted on every
    # invocation; a bare IP listed in the engagement never widens to its /24.
    scope = engagement.scope

    assert isinstance(scope, TargetScope)
    assert scope.allows("10.10.10.5") is True
    assert scope.allows("10.10.10.0/28") is True
    assert scope.allows("10.0.0.0/8") is False
    assert scope.allows("10.10.11.5") is False


def test_a_tool_above_the_engagement_max_level_is_refused(
    engagement: EngagementAuthorization, audit_log: AuditLog
) -> None:
    # nuclei is a level-2 tool and this engagement signed off at level 1.
    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nuclei", "target": "10.10.10.5", "args": [], "level": 2,
             "task_id": "pentest-level"},
            engagement=engagement,
            audit_log=audit_log,
        )

    assert excinfo.value.reason_code == "level_not_engaged"
    assert "level 1" in str(excinfo.value)
    _assert_refusal_audited(audit_log, engagement, "level_not_engaged", "pentest-level")


def test_a_level_the_engagement_numbers_but_does_not_authorise_is_refused(
    audit_log: AuditLog,
) -> None:
    # max_level=3 with only "recon" named is a valid artifact. It must not let a
    # level-2 vulnerability scan inherit simply because the number fits: the
    # class belonging to the level has to be in allowed_tool_classes as well.
    narrow = EngagementAuthorization(
        owner="security-reviewer",
        targets=("10.10.10.0/24",),
        max_level=3,
        allowed_tool_classes=frozenset({"recon"}),
    )

    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nuclei", "target": "10.10.10.5", "args": [], "level": 2,
             "task_id": "pentest-class"},
            engagement=narrow,
            audit_log=audit_log,
        )

    assert excinfo.value.reason_code == "level_not_engaged"
    assert "vulnerability" in str(excinfo.value)
    _assert_refusal_audited(audit_log, narrow, "level_not_engaged", "pentest-class")


@pytest.mark.parametrize("level", ["1", 1.0, True, [1], b"1"])
def test_a_malformed_level_is_refused_rather_than_coerced(
    engagement: EngagementAuthorization, audit_log: AuditLog, level: object
) -> None:
    # ``int("1")``, ``int(1.0)`` and ``int(True)`` all produce an int. A blind
    # conversion would turn a malformed declaration into an authorised one, so
    # the level is narrowed instead and anything that is not already a plain int
    # is refused - a bool included, since ``isinstance(True, int)`` is True.
    assert engagement.engages_level(level) is False
    assert engagement.allows_level(level) is False

    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nmap", "target": "10.10.10.5", "args": [], "level": level,
             "task_id": "pentest-level-type"},
            engagement=engagement,
            audit_log=audit_log,
        )

    assert excinfo.value.reason_code == "level_malformed"
    _assert_refusal_audited(audit_log, engagement, "level_malformed", "pentest-level-type")


def test_a_missing_level_declares_nothing_and_is_never_coerced(
    engagement: EngagementAuthorization,
) -> None:
    # An absent level falls back to the engagement's own ceiling, which is a
    # narrowing rather than a coercion - the caller cannot raise its own level
    # by omitting the field.
    assert engagement.engages_level(None) is False
    result = authorize_tool_call(
        {"tool": "nmap", "target": "10.10.10.5", "args": ["-sT"]}, engagement=engagement
    )

    assert result.level == engagement.max_level


def test_an_unknown_tool_class_is_refused_when_the_artifact_is_built() -> None:
    with pytest.raises(TargetScopeError, match="unknown engagement tool class"):
        EngagementAuthorization(
            owner="security-reviewer",
            targets=("10.10.10.0/24",),
            allowed_tool_classes=frozenset({"exfiltration"}),
        )


def test_an_unregistered_level_has_no_tool_class_and_is_refused() -> None:
    with pytest.raises(PentestRefusal) as excinfo:
        tool_class_for_level(MAX_ENGAGEMENT_LEVEL + 1)

    assert excinfo.value.reason_code == "unknown_level"


def test_every_level_the_engagement_defines_has_a_declared_tool_class() -> None:
    # The two maps live next to each other on purpose: an engagement and the
    # catalog that enforces it must not be able to disagree about which class a
    # number means.
    assert set(TOOL_CLASS_BY_LEVEL.values()) == set(TOOL_CLASSES)
    assert sorted(TOOL_CLASS_BY_LEVEL) == list(range(MAX_ENGAGEMENT_LEVEL + 1))


def test_a_closed_engagement_window_refuses_the_call(
    engagement: EngagementAuthorization,
) -> None:
    closed = EngagementAuthorization(
        owner=engagement.owner,
        targets=engagement.targets,
        max_level=1,
        allowed_tool_classes=frozenset({"recon"}),
        window_start=100.0,
        window_end=200.0,
    )

    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nmap", "target": "10.10.10.5", "args": ["-sT"], "level": 1},
            engagement=closed,
            now=1.0,
        )

    assert excinfo.value.reason_code == "engagement_window_closed"


def test_no_engagement_artifact_is_a_refusal_and_not_a_default_engagement() -> None:
    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {"tool": "nmap", "target": "10.10.10.5", "args": []}, engagement=None
        )

    assert excinfo.value.reason_code == "no_engagement"


# ----------------------------------------------------------------------
# The two numeric value rules stay separate
#
# ``int_range`` and ``float_range`` are different validators. Collapsing them
# would let an integer option accept a fractional token, so both are pinned: a
# float stays a float, an integer stays an integer, and a fractional token for an
# integer option is a malformed value rather than a silently truncated one.
# ----------------------------------------------------------------------
def _exploitation_engagement() -> EngagementAuthorization:
    return EngagementAuthorization(
        owner="security-reviewer",
        targets=("10.10.10.0/24",),
        max_level=3,
        allowed_tool_classes=frozenset({"exploitation"}),
    )


@pytest.mark.parametrize(
    ("token", "authorised"),
    [("-T12.5", True), ("-T60", True), ("-T0", True), ("-T61", False), ("-Tfast", False)],
)
def test_the_hydra_task_size_option_keeps_its_float_range(token: str, authorised: bool) -> None:
    # Options are written in attached form: a bare "-l" would be an option with
    # no value, and the catalog refuses to guess which token the value was.
    intent = {
        "tool": "hydra",
        "target": "10.10.10.5",
        "args": ["-ladmin", token],
        "level": 3,
    }

    if not authorised:
        with pytest.raises(PentestRefusal) as excinfo:
            authorize_tool_call(intent, engagement=_exploitation_engagement())
        assert excinfo.value.reason_code in {"value_out_of_range", "value_malformed"}
        return

    result = authorize_tool_call(intent, engagement=_exploitation_engagement())
    # The validated token is the one that reaches the argv, and hydra's trailing
    # protocol token is appended after the target.
    assert result.argv == ("/usr/bin/hydra", "-ladmin", token, "10.10.10.5", "ssh")


def test_an_integer_option_still_refuses_a_fractional_value(
    engagement: EngagementAuthorization,
) -> None:
    authorised = authorize_tool_call(
        {
            "tool": "curl",
            "target": "http://10.10.10.5/",
            "args": ["--max-time5"],
            "level": 1,
        },
        engagement=engagement,
    )
    assert "--max-time5" in authorised.argv

    with pytest.raises(PentestRefusal) as excinfo:
        authorize_tool_call(
            {
                "tool": "curl",
                "target": "http://10.10.10.5/",
                "args": ["--max-time5.5"],
                "level": 1,
            },
            engagement=engagement,
        )

    assert excinfo.value.reason_code == "value_malformed"
    assert "not an integer" in str(excinfo.value)


# ----------------------------------------------------------------------
# Every refusal is recorded
# ----------------------------------------------------------------------
def test_every_catalog_refusal_is_audited_with_its_reason_code(
    engagement: EngagementAuthorization, audit_log: AuditLog
) -> None:
    # A refusal nobody can find in the log is a refusal an operator cannot learn
    # from. Each case below is a different refusal reason, and each has to leave
    # exactly one record naming it.
    cases: list[tuple[dict[str, object], str]] = [
        (
            {"tool": "rm -rf /", "target": "10.10.10.5", "args": [], "level": 1,
             "task_id": "pentest-unknown-tool"},
            "unknown_tool",
        ),
        (
            {"tool": "nmap", "target": "10.10.10.5", "args": ["-sS"], "level": 1,
             "task_id": "pentest-forbidden-arg"},
            "forbidden_argument",
        ),
        (
            {"tool": "nmap", "target": "192.168.5.5", "args": [], "level": 1,
             "task_id": "pentest-out-of-scope"},
            "target_out_of_scope",
        ),
        (
            {"tool": "nuclei", "target": "10.10.10.5", "args": [], "level": 2,
             "task_id": "pentest-above-max-level"},
            "level_not_engaged",
        ),
        (
            {"tool": "nmap", "target": "10.10.10.5", "args": [], "level": "1",
             "task_id": "pentest-malformed-level"},
            "level_malformed",
        ),
        (
            {"tool": "nmap", "target": "10.10.10.5", "args": "nope", "level": 1,
             "task_id": "pentest-malformed-args"},
            "args_malformed",
        ),
    ]

    for intent, reason_code in cases:
        with pytest.raises(PentestRefusal) as excinfo:
            authorize_tool_call(intent, engagement=engagement, audit_log=audit_log)
        assert excinfo.value.reason_code == reason_code

    events = audit_log.query(AuditQuery(event_type=AuditEventType.PERMISSION_CHECK))
    assert len(events) == len(cases)
    assert {event.error for event in events} == {reason_code for _, reason_code in cases}
    assert {event.result for event in events} == {"refused"}
    assert {event.agent_id for event in events} == {"pentester"}
    assert len({event.task_id for event in events}) == len(cases)