"""
The write path, sealed under the group key (MNP 2.0).

Downloads have been encrypted under a GEK-derived key since the beginning:
`file_chunk` and `stream_data` both go through `chunk_ciphertext`. Uploads did
not. `file_upload` carried the filename and the raw bytes in plain msgpack and
`file_upload_ack` carried the name the node stored them under, so the same file
was ciphertext leaving a node and plaintext arriving at one — an asymmetry with
no threat model behind it.

What sealing buys is what `groupbox.py` says and no more: nothing against a
network observer (DTLS covers that), nothing against the hub (never on this
channel), nothing against a member (they hold the GEK). It buys defence in
depth against our own next handshake bug, of a class already shipped twice —
C1, the unauthenticated node HTTP API, and C6, the transport that took a bare
JWT. Both were "a peer that had not finished the handshake was served data".
Sealed, the equivalent bug on this path leaks ciphertext instead of the
operator's filenames.
"""

from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_common.protocol import MNP, file_upload_wire
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root, opened_ack, sealed_upload

GROUP = "g" * 32


def _session(tmp_path: Path, *, gek: bytes | None = None) -> WebRTCPeerSession:
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {"roots": one_root(shared_root), "index": index,
                    "sk_node": index.sk_node, "gek": gek or generate_gek()}
    session._group_id = GROUP
    session._user_id = "user-1"
    session._pk_user = ""
    session._uploads = {}
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def _root(session):
    return session._ctx["roots"].roots[0]


def _errors(session):
    return [m for m in session.sent if m.get("type") == "error"]


def _wrote_anything(tmp_path) -> bool:
    return any(p.is_file() for p in tmp_path.rglob("*"))


# ── The message itself ───────────────────────────────────────────────────────

def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path):
    """
    The point of the exercise. `upload_id` and `chunk_index` are outside the
    seal because the node routes and orders on them before it can decrypt;
    everything that names or is the operator's content is inside it.
    """
    session = _session(tmp_path)
    msg = sealed_upload(session, filename="holiday.jpg", data=b"JPEGDATA",
                        dir=f"{_root(session).name}")

    assert set(msg) == {"type", "v", "upload_id", "chunk_index",
                        "total_chunks", "nonce", "ct"}
    blob = repr(msg).encode() + msg["ct"]
    assert b"holiday.jpg" not in blob, "the filename is on the wire in clear"
    assert b"JPEGDATA" not in blob, "the file content is on the wire in clear"


async def test_the_ack_carries_no_stored_name(tmp_path):
    """
    `stored_as` is the name the node settled on — it finds a free one rather
    than replacing anything — and naming it in clear would hand back exactly
    what the request took the trouble to hide.
    """
    session = _session(tmp_path)
    await session._do_file_upload(sealed_upload(
        session, filename="holiday.jpg", data=b"x", dir=_root(session).name))

    ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1]
    assert set(ack) == {"type", "v", "upload_id", "chunk_index", "nonce", "ct"}
    assert b"holiday.jpg" not in repr(ack).encode() + ack["ct"]
    assert opened_ack(session, ack) == {
        "filename": "holiday.jpg", "stored_as": "holiday.jpg",
        "dir": _root(session).name}


async def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
    """
    `filename` used to be the correlation key on both sides. It cannot be one
    any more, and `upload_id` replaces it — a client-chosen label, opaque to
    the node, never an authorization input. Without it a client running several
    uploads could only match replies by arrival order, which is how a refusal
    for one file used to fail every upload in flight.
    """
    session = _session(tmp_path)
    await session._do_file_upload(sealed_upload(
        session, filename="a.txt", data=b"x", dir=_root(session).name,
        upload_id="upload-A"))
    await session._do_file_upload(sealed_upload(
        session, filename="../evil", data=b"x", dir=_root(session).name,
        upload_id="upload-B"))

    ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1]
    assert ack["upload_id"] == "upload-A"
    assert _errors(session)[0]["upload_id"] == "upload-B"


# ── What is refused ──────────────────────────────────────────────────────────

async def test_a_plaintext_upload_is_refused(tmp_path):
    """
    The MNP 1.x shape, which is what an un-updated client sends. Refused with a
    code and a message saying which side is old — never accepted "just this
    once", because a path that still takes plaintext is not a sealed path.
    """
    session = _session(tmp_path)
    await session._do_file_upload({
        "filename": "note.txt", "dir": _root(session).name,
        "chunk_index": 0, "total_chunks": 1, "data": b"x",
    })

    assert _errors(session)[0]["code"] == "upload_not_sealed"
    assert not _wrote_anything(tmp_path)


async def test_a_tampered_chunk_is_refused(tmp_path):
    """
    AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext
    must stop the upload, not produce a corrupt file with a plausible name.
    """
    session = _session(tmp_path)
    msg = sealed_upload(session, filename="note.txt", data=b"x" * 64,
                        dir=_root(session).name)
    msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:]
    await session._do_file_upload(msg)

    assert _errors(session)[0]["code"] == "upload_not_sealed"
    assert not _wrote_anything(tmp_path)


async def test_an_upload_sealed_for_another_group_is_refused(tmp_path):
    """
    The group is the AAD, so a node hosting two groups cannot have a chunk
    moved between them — and a member of one cannot write into the other by
    reaching a session that is on it (finding H1's shape, on the write path).
    """
    session = _session(tmp_path)
    msg = file_upload_wire(
        session._ctx["gek"], "some-other-group",
        upload_id="u1", chunk_index=0, total_chunks=1,
        filename="note.txt", data=b"x", dir=_root(session).name)
    await session._do_file_upload(msg)

    assert _errors(session)[0]["code"] == "upload_not_sealed"
    assert not _wrote_anything(tmp_path)


async def test_an_upload_under_another_key_is_refused(tmp_path):
    """A peer past the handshake with the wrong GEK still writes nothing."""
    session = _session(tmp_path)
    msg = file_upload_wire(
        generate_gek(), GROUP,
        upload_id="u1", chunk_index=0, total_chunks=1,
        filename="note.txt", data=b"x", dir=_root(session).name)
    await session._do_file_upload(msg)

    assert _errors(session)[0]["code"] == "upload_not_sealed"
    assert not _wrote_anything(tmp_path)


def test_an_ack_replayed_as_a_request_does_not_open(tmp_path):
    """
    The message type is in the AAD, so the two halves of an upload cannot be
    confused for each other. Cheap, and it closes a class that is tedious to
    reason about after the fact.
    """
    from cryptography.exceptions import InvalidTag
    from meshbay_common.protocol import file_upload_ack_wire, file_upload_payload

    session = _session(tmp_path)
    ack = file_upload_ack_wire(
        session._ctx["gek"], GROUP, upload_id="u1", chunk_index=0,
        filename="note.txt", stored_as="note.txt", dir="shared")
    with pytest.raises(InvalidTag):
        file_upload_payload(session._ctx["gek"], GROUP, ack)


async def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
    """
    A node whose group has no GEK yet cannot open anything. It must say so, not
    read the message as though it were the old plaintext shape.
    """
    session = _session(tmp_path)
    msg = sealed_upload(session, filename="note.txt", data=b"x",
                        dir=_root(session).name)
    session._ctx["gek"] = b""
    await session._do_file_upload(msg)

    assert _errors(session)[0]["code"] == "no_group_key"
    assert not _wrote_anything(tmp_path)


# ── What must still work ─────────────────────────────────────────────────────

async def test_a_multi_chunk_upload_reassembles(tmp_path):
    """
    Every chunk is sealed under its own nonce, and the node appends in order.
    Nothing about the seal may change what lands on disk.
    """
    session = _session(tmp_path)
    body = bytes(range(256)) * 40
    parts = [body[i:i + 1024] for i in range(0, len(body), 1024)]
    for i, part in enumerate(parts):
        await session._do_file_upload(sealed_upload(
            session, filename="blob.bin", data=part,
            chunk_index=i, total_chunks=len(parts), dir=_root(session).name))

    assert (_root(session).path / "blob.bin").read_bytes() == body
    assert not list(_root(session).path.glob("*.part")), "a temp file was left"
    acks = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK]
    assert [m["chunk_index"] for m in acks] == list(range(len(parts)))


def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path):
    """
    A file of repeated bytes is ordinary, so the nonce must come from the RNG
    and never from the payload. Cheap to assert and expensive to discover.
    """
    session = _session(tmp_path)
    a = sealed_upload(session, filename="f", data=b"same", chunk_index=0)
    b = sealed_upload(session, filename="f", data=b"same", chunk_index=0)
    assert a["nonce"] != b["nonce"]
    assert a["ct"] != b["ct"]


async def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
    """
    Opening a payload proves a member wrote it, not that they wrote something
    sensible. A member can seal anything, so the fields still need their types
    checked — `SAFE_UPLOAD_NAME.match(123)` raises where a refusal was meant,
    and the dispatcher's catch-all would turn that into "Request failed".
    """
    from meshbay_common.groupbox import PURPOSE_UPLOAD, seal

    session = _session(tmp_path)
    for payload in ({"filename": 123, "data": b"x"},
                    {"filename": "note.txt", "data": "not bytes"},
                    {"filename": "note.txt"}):
        session.sent.clear()
        await session._do_file_upload({
            "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1",
            "chunk_index": 0, "total_chunks": 1,
            **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD,
                   GROUP, payload),
        })
        errs = _errors(session)
        assert errs, f"{payload!r} was accepted"
        assert errs[0]["code"] in ("upload_incomplete", "bad_chunk_encoding")
    assert not _wrote_anything(tmp_path)


async def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path):
    """`chunk_index` is outside the seal by necessity, so it is unchecked input."""
    session = _session(tmp_path)
    msg = sealed_upload(session, filename="note.txt", data=b"x",
                        dir=_root(session).name)
    msg["chunk_index"] = "zero"
    await session._do_file_upload(msg)

    assert _errors(session)[0]["code"] == "bad_chunk_index"
    assert not _wrote_anything(tmp_path)