"""
Phase 11.5 security regression tests.

Each test here encodes a finding from `second-review.md`. They are negative tests:
they assert that an attack does NOT work. The pre-11.5 code passed 209 feature
tests while every one of these attacks succeeded — the suite only ever exercised
happy paths, never an authorization boundary.

If one of these starts failing, a fix has been reverted. Do not "fix" the test.
"""

import base64
import struct
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc_server import WebRTCPeerSession


def _safe_name_re():
    """
    Imported lazily so that a missing allowlist fails the two tests that need it,
    rather than aborting collection of the whole module and hiding every other
    finding's result.
    """
    from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME
    return SAFE_UPLOAD_NAME


# ── C1: the unauthenticated HTTP file API must stay deleted ───────────────────

def test_http_file_api_is_gone():
    """
    C1: transport/http_server.py served GET /index and GET /file/{id} on 0.0.0.0
    with no authentication, for private groups too. It was deleted rather than
    patched. Re-adding any module that serves file bytes outside the MNP handshake
    reintroduces a full confidentiality bypass.
    """
    with pytest.raises(ImportError):
        import meshbay_node.transport.http_server  # noqa: F401

    import meshbay_node.transport as transport
    assert not hasattr(transport, "create_http_app")


def test_tcp_transport_is_gone():
    """C6: the TCP+TLS server accepted a bare JWT with no GEK proof."""
    with pytest.raises(ImportError):
        import meshbay_node.transport.server  # noqa: F401

    import meshbay_node.transport as transport
    assert not hasattr(transport, "ChunkServer")


def test_daemon_exposes_no_plaintext_listener():
    """
    C1: the daemon must not bind anything that serves content without a handshake.
    NodeConfig no longer carries an HTTP port at all.
    """
    from meshbay_node.config import NodeConfig, GroupConfig

    assert "http_port" not in NodeConfig.__dataclass_fields__
    assert "http_port" not in GroupConfig.__dataclass_fields__
    assert "port" not in NodeConfig.__dataclass_fields__


# ── C5a: upload filename allowlist ───────────────────────────────────────────

@pytest.mark.parametrize("name", [
    "../../etc/passwd",
    "..\\windows\\system32",
    "/absolute/path",
    "<img src=x onerror=alert(1)>",          # the H2 stored-XSS vector
    'name";DROP TABLE x;--',
    ".hidden",
    "",
    "a" * 200,
    "file\x00.mp4",
    "sub/dir/file.mp4",
])
def test_upload_rejects_unsafe_filenames(name):
    """C5a/H2: only a conservative allowlist may reach the filesystem."""
    assert not _safe_name_re().match(name), f"should be rejected: {name!r}"


@pytest.mark.parametrize("name", [
    "movie.mp4",
    "My Holiday Video.mkv",
    "report-2026.pdf",
    "track_01.flac",
])
def test_upload_accepts_ordinary_filenames(name):
    """The allowlist must not break normal use."""
    assert _safe_name_re().match(name), f"should be accepted: {name!r}"


def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
    """A peer session wired to a real shared root, with sending stubbed out."""
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node}

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = None
    session._user_id = user_id
    session._pk_user = ""
    session._uploads = {}
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def test_upload_cannot_overwrite_another_members_file(tmp_path):
    """
    C5a: uploads used to land in the shared root under a client-chosen name and
    overwrite whatever was there. That let any member destroy the operator's files,
    and — by becoming the recorded uploader of the replaced file — delete them
    through the uploader path, bypassing the Ed25519 admin challenge entirely.
    """
    victim = _session(tmp_path, "victim-user")
    shared_root = victim._ctx["shared_root"]

    original = shared_root / "important.mp4"
    original.write_bytes(b"operator's original content")

    attacker = _session(tmp_path, "attacker-user")
    attacker._do_file_upload({
        "filename": "important.mp4",
        "chunk_index": 0,
        "total_chunks": 1,
        "data": base64.b64encode(b"attacker content").decode(),
    })

    assert original.read_bytes() == b"operator's original content"
    uploaded = shared_root / ".uploads" / "attacker-user" / "important.mp4"
    assert uploaded.exists(), "upload should be quarantined, not dropped"
    assert uploaded.read_bytes() == b"attacker content"


def test_upload_rejects_out_of_order_chunks(tmp_path):
    """C5a: chunk_index > 0 used to append blindly to any .part file on disk."""
    session = _session(tmp_path, "user-1")
    session._do_file_upload({
        "filename": "movie.mp4", "chunk_index": 3, "total_chunks": 5,
        "data": base64.b64encode(b"spliced").decode(),
    })
    assert any(m.get("type") == "error" for m in session.sent)


def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
    """C5a: even the original uploader goes through a fresh name, not an overwrite."""
    session = _session(tmp_path, "user-1")
    payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1,
               "data": base64.b64encode(b"first").decode()}
    session._do_file_upload(dict(payload))
    session.sent.clear()

    session._do_file_upload(dict(payload))
    assert any(m.get("type") == "error" for m in session.sent)
    stored = session._ctx["shared_root"] / ".uploads" / "user-1" / "movie.mp4"
    assert stored.read_bytes() == b"first"


# ── H1: group isolation ──────────────────────────────────────────────────────

def test_chat_store_and_peers_are_per_group(tmp_path):
    """
    H1: chat_store and the peer registry were read from the shared transport
    context, so on a multi-group node every group's messages went to the first
    group's database and were served back to members of every other group.
    """
    index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate())
    index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate())
    groups = {
        "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path},
        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path},
    }
    ctx = {"groups": groups}

    sess_a = WebRTCPeerSession.__new__(WebRTCPeerSession)
    sess_a._ctx, sess_a._group_id, sess_a._user_id = ctx, "a" * 32, "alice"

    sess_b = WebRTCPeerSession.__new__(WebRTCPeerSession)
    sess_b._ctx, sess_b._group_id, sess_b._user_id = ctx, "b" * 32, "bob"

    assert sess_a._group_ctx()["chat_store"] == "STORE_A"
    assert sess_b._group_ctx()["chat_store"] == "STORE_B"

    sess_a._peer_registry()["alice"] = sess_a
    sess_b._peer_registry()["bob"] = sess_b

    # Alice's broadcast target set must not contain Bob, who is in another group.
    assert "bob" not in sess_a._peer_registry()
    assert "alice" not in sess_b._peer_registry()

    sess_a._user_names()["alice"] = "Alice"
    assert "alice" not in sess_b._user_names()


def test_daemon_sets_no_global_chat_store(tmp_path):
    """H1: the daemon must not hoist one group's chat store onto the transport."""
    source = (Path(__file__).parent.parent
              / "src" / "meshbay_node" / "daemon.py").read_text()
    assert '_ctx["chat_store"]' not in source, (
        "daemon must not assign a transport-wide chat_store — it leaks chat "
        "across groups (H1)"
    )


# ── H2: node admin UI escaping ───────────────────────────────────────────────

def test_gek_bundle_store_requires_admin_challenge(tmp_path):
    """
    C5b: gek_bundle_store used to write whatever any authenticated member sent.
    It must now answer with a challenge and store nothing until a valid
    node-operator signature arrives.
    """
    session = _session(tmp_path, "ordinary-member")
    session._group_id = None
    session._admin_ops = {}
    session._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key()

    stored = []

    class _Store:
        async def store(self, *args):
            stored.append(args)

    session._ctx["bundle_store"] = _Store()
    session._do_gek_bundle_store({
        "user_id": "victim", "group_id": "g" * 32,
        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
    })

    assert stored == [], "bundle written without operator authorization (C5b)"
    assert any(m.get("type") == "admin_challenge" for m in session.sent)


def test_gek_bundle_store_refused_without_pinned_admin_key(tmp_path):
    """C5b: deny by default — no pinned key means no privileged operation."""
    session = _session(tmp_path, "ordinary-member")
    session._group_id = None
    session._admin_ops = {}
    session._ctx["bundle_store"] = object()

    session._do_gek_bundle_store({
        "user_id": "victim", "group_id": "g" * 32,
        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
    })
    assert any(m.get("type") == "error" for m in session.sent)


def test_gek_auto_activation_is_gone():
    """
    C5b: the node used to unwrap and adopt any bundle addressed to the operator.
    Since the operator's X25519 public key is public, any member could hand the
    node a GEK of their choosing. Nothing arriving over MNP may set a live GEK.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text()
    assert "_try_activate_gek" not in source
    assert 'unwrap_gek_aes' not in source, (
        "the MNP path must not unwrap a GEK — activation is local-admin only"
    )


# ── H5: admin challenge is bound, not a blind signing oracle ─────────────────

def _transcript(**kw):
    from meshbay_common.adminop import admin_transcript
    base = dict(op="file_delete", node_pk_b64="NODEPK", group_id="g" * 32,
                subject="file-1", nonce=b"\x01" * 32, ts=1_700_000_000)
    base.update(kw)
    return admin_transcript(**base)


def test_admin_transcript_is_domain_separated():
    """H5: signatures here can never be valid in another MeshBay protocol."""
    assert _transcript().startswith(b"meshbay:admin:v1")


@pytest.mark.parametrize("field,value", [
    ("op", "gek_bundle_store"),
    ("subject", "file-2"),
    ("node_pk_b64", "OTHERNODE"),
    ("group_id", "h" * 32),
    ("nonce", b"\x02" * 32),
    ("ts", 1_700_000_001),
])
def test_admin_transcript_binds_every_field(field, value):
    """
    H5: a signature must not carry over to another operation, subject, node,
    group, challenge or moment in time.
    """
    assert _transcript() != _transcript(**{field: value}), (
        f"transcript ignores {field} — signature would be reusable"
    )


def test_admin_transcript_is_unambiguous():
    """
    H5/L4: fields are length-prefixed. With plain concatenation a crafted subject
    could impersonate the following field and two different operations would
    produce identical signed bytes.
    """
    a = _transcript(subject="file-1", group_id="g")
    b = _transcript(subject="1", group_id="gfile-")
    assert a != b, "concatenation is ambiguous — length prefixes missing"


def test_admin_signature_does_not_transfer_between_operations(tmp_path):
    """
    H5: the concrete attack. A signature collected to delete a file must not
    authorize storing a GEK bundle.
    """
    from meshbay_common.adminop import OP_FILE_DELETE, OP_GEK_BUNDLE_STORE

    sk_admin = Ed25519PrivateKey.generate()
    delete_transcript = _transcript(op=OP_FILE_DELETE)
    signature = sk_admin.sign(delete_transcript)

    store_transcript = _transcript(op=OP_GEK_BUNDLE_STORE)
    with pytest.raises(Exception):
        sk_admin.public_key().verify(signature, store_transcript)


def test_admin_challenge_expires(tmp_path):
    """H5: a stale challenge must not be usable."""
    import time as _time
    from meshbay_common.adminop import ADMIN_CHALLENGE_TTL, OP_FILE_DELETE

    session = _session(tmp_path, "operator")
    session._group_id = None
    session._admin_ops = {
        "op-1": {
            "op": OP_FILE_DELETE, "subject": "file-1", "nonce": b"\x00" * 32,
            "ts": int(_time.time()) - ADMIN_CHALLENGE_TTL - 5, "payload": {},
        }
    }
    session._do_admin_response({"op_id": "op-1", "signature": ""})
    assert any(m.get("type") == "error" and "expired" in m.get("detail", "").lower()
               for m in session.sent)


def test_denylist_persists_and_honours_groups(tmp_path):
    """
    H4: revocations lived only in memory, so a node restart silently un-revoked
    everyone, and 'group' targets were dropped entirely — the hub signed and
    broadcast them, the node's handler understood only 'user' and 'jti'.
    """
    from meshbay_node.transport import Denylist

    path = tmp_path / "denylist.json"
    first = Denylist(path=path)
    first.deny_group("g-revoked")
    first.deny_user("u-revoked")
    first.deny_jti("j-revoked")

    # A fresh instance stands in for a daemon restart.
    reloaded = Denylist(path=path)
    assert reloaded.is_denied("", "", "g-revoked"), "group revocation not honoured"
    assert reloaded.is_denied("u-revoked", "")
    assert reloaded.is_denied("", "j-revoked")
    assert not reloaded.is_denied("someone", "other", "g-allowed")


def test_swarm_registration_skips_private_groups():
    """
    H7: the daemon registered content hashes for every group, private included,
    handing the hub a fingerprint of every private file. The bug was masked by a
    mis-mounted route, so fixing the route without this filter would have turned a
    dormant leak into a live one.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "daemon.py").read_text()
    assert 'visibility' in source and '_register_swarm' in source
    # Both registration sites must gate on public visibility.
    for marker in ['gctx.get("visibility") != "public"',
                   'group_cfg.visibility == "public"']:
        assert marker in source, f"swarm registration not gated: {marker}"


def test_keystore_argon2_is_production_strength():
    """M2: the keystore protects the node's private keys and sat at 64 MB."""
    from meshbay_common.crypto import ARGON2_MEMORY_COST
    assert ARGON2_MEMORY_COST >= 262144


def test_keystore_records_argon2_params_for_migration(tmp_path):
    """
    M2: raising the parameters must not orphan existing keystores, so each
    envelope records the parameters it was written with.
    """
    import json
    from meshbay_node.keystore import create_keystore, load_keystore

    path = tmp_path / "keystore.enc"
    created = create_keystore(path=path, password="correct horse battery")
    envelope = json.loads(path.read_text())
    assert envelope["argon2"]["memory_cost"] >= 262144

    reopened = load_keystore(path=path, password="correct horse battery")
    assert reopened.pk_ed25519_b64 == created.pk_ed25519_b64


def test_legacy_keystore_still_opens(tmp_path):
    """M2: a keystore written under the 64 MB profile must still unlock."""
    import base64 as _b64
    import json
    import msgpack
    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
    from meshbay_common.crypto import (
        LEGACY_ARGON2_ITERATIONS, LEGACY_ARGON2_LANES, LEGACY_ARGON2_MEMORY_COST,
        derive_keystore_key, encrypt_keystore, pk_to_b64, sk_to_b64,
    )
    from meshbay_node.keystore import load_keystore

    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
    payload = msgpack.packb({
        "sk_ed25519_b64": sk_to_b64(sk_ed),
        "sk_x25519_b64": sk_to_b64(sk_x),
    }, use_bin_type=True)

    salt = b"\x01" * 16
    key = derive_keystore_key(
        "legacy-pass", salt,
        iterations=LEGACY_ARGON2_ITERATIONS,
        memory_cost=LEGACY_ARGON2_MEMORY_COST,
        lanes=LEGACY_ARGON2_LANES,
    )
    iv, ct, tag = encrypt_keystore(payload, key)

    path = tmp_path / "legacy.enc"
    # No "argon2" key — exactly how pre-M2 envelopes look.
    path.write_text(json.dumps({
        "version": 1,
        "argon2_salt_b64": _b64.b64encode(salt).decode(),
        "iv_b64": _b64.b64encode(iv).decode(),
        "tag_b64": _b64.b64encode(tag).decode(),
        "ciphertext_b64": _b64.b64encode(ct).decode(),
    }))

    keys = load_keystore(path=path, password="legacy-pass")
    assert keys.pk_ed25519_b64 == pk_to_b64(sk_ed.public_key())
    assert keys.pk_x25519_b64 == pk_to_b64(sk_x.public_key())


def test_dead_gek_protocol_constants_removed():
    """L1: the node never serves a GEK; the message types should not suggest it."""
    from meshbay_common.protocol import MNP
    assert not hasattr(MNP, "GEK_REQUEST")
    assert not hasattr(MNP, "GEK_RESPONSE")


def test_peer_errors_do_not_leak_internals():
    """
    L3: arbitrary exception text carries filesystem paths and internal state, so
    the catch-all handler must not relay it.

    Deliberately narrow: HandshakeError messages ARE sent to the peer, because a
    client needs to know why it was refused, and those strings are authored for
    that purpose. The check targets the generic `except Exception as e` path.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text()
    assert '"detail": str(e)' not in source, (
        "generic exception text relayed to peer — use a fixed message"
    )
    # And the catch-all must still exist, sending something opaque.
    assert '"detail": "Request failed"' in source


def test_pre_handshake_message_budget_is_small():
    """
    H6: the frame limit was a flat 64 MB applied before authentication, so an
    unauthenticated peer could announce a huge frame and dribble bytes into it.
    """
    from meshbay_node.transport.webrtc_server import (
        MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer,
    )
    assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024
    assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG

    buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
    buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x")
    with pytest.raises(ValueError):
        list(buf.messages())


def test_stream_segment_is_not_synchronous():
    """
    H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop,
    stalling every peer on the node for up to thirty seconds per request.

    Asserts the property (the worker is a coroutine, ffmpeg is spawned through
    asyncio) rather than grepping for "subprocess.run" — which also matches the
    comment that documents the old behaviour.
    """
    import ast
    import inspect
    from meshbay_node.transport.webrtc_server import WebRTCPeerSession

    assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async)

    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text()
    tree = ast.parse(source)
    blocking = [
        node for node in ast.walk(tree)
        if isinstance(node, ast.Call)
        and isinstance(node.func, ast.Attribute)
        and node.func.attr == "run"
        and isinstance(node.func.value, ast.Name)
        and node.func.value.id == "subprocess"
    ]
    assert not blocking, "blocking subprocess.run() in the event loop"
    assert "_transcode_sem" in source, "ffmpeg spawns must be capped"


def test_pre_proof_fetches_are_bounded():
    """C4: the pre-proof bundle window is a disclosure surface; bound it."""
    from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES
    assert 0 < MAX_PRE_PROOF_FETCHES <= 10


def test_node_admin_ui_requires_token():
    """
    11.5.3: "localhost only" is not authentication. Any local process — or a
    rebound browser page — could re-initialise a group's GEK and read the audit log.
    """
    from fastapi.testclient import TestClient
    from meshbay_node.ui.app import create_ui_app

    app = create_ui_app({"status": "running", "groups_ctx": {},
                         "indexes": {}, "ui_token": "secret-token"})
    client = TestClient(app)

    assert client.get("/api/status").status_code == 403
    assert client.get("/api/status?t=wrong").status_code == 403
    assert client.get("/api/config?t=wrong").status_code == 403
    assert client.get("/api/status?t=secret-token").status_code == 200
    assert client.get(
        "/api/status", headers={"X-MeshBay-Token": "secret-token"}
    ).status_code == 200


def test_admin_ui_escapes_filenames(tmp_path):
    """
    H2: filenames are chosen by any group member and were rendered into the
    localhost admin UI unescaped, giving script execution against an
    unauthenticated admin API.
    """
    from meshbay_node.ui.app import _render_page

    payload = '<img src=x onerror="fetch(1)">'
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    index.add_entry(IndexEntry(
        id="0" * 64, name=payload, path="", size=1, type="video", added_at=0,
    ))

    html = _render_page({
        "status": "running",
        "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}},
        "indexes": {"g" * 32: index},
    })

    assert payload not in html, "filename rendered unescaped — stored XSS (H2)"
    assert "&lt;img" in html, "filename should appear escaped"