"""
The test that asserts the property, rather than the mechanism.

Worth more than checking that a `ct` field is present: this fails for any future
change that puts a name back in the clear, including one nobody thought of as an
index message. Findings C1 (the node HTTP API served the index and plaintext files
on 0.0.0.0 with no authentication) and C6 (the TCP transport accepted a bare JWT
with no GEK proof) were both "a peer that had not completed the handshake was served
data"; sealed, that bug leaks ciphertext instead of a library's filenames.

The distinctive strings below are invented and could not occur by chance in msgpack
framing or in a field name.
"""

import msgpack
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_INDEX, seal, unseal
from meshbay_common.protocol import MNP
from meshbay_node.indexer import DirectoryIndexer, GroupIndex
from meshbay_node.transport.wire import index_delta_message, index_sync_message

from conftest import one_root

# A filename and a folder name that appear nowhere else in the tree.
SECRET_FILE = "quixotry-ledger-2019.pdf"
SECRET_DIR = "zarfwidget-archive"
SECRET_ROOT = "/srv/vasculum-private/library"


@pytest.fixture
def gek():
    return generate_gek()


@pytest.fixture
async def indexer(tmp_path, gek):
    shared = tmp_path / "shared"
    (shared / SECRET_DIR).mkdir(parents=True)
    (shared / SECRET_DIR / SECRET_FILE).write_bytes(b"x" * 64)
    roots = one_root(shared, name="library")
    idx = DirectoryIndexer(
        roots=roots, group_id="g-1", sk_node=Ed25519PrivateKey.generate(), gek=gek)
    await idx.initial_scan()
    return idx


def _assert_absent(frame: bytes, *words: str) -> None:
    for word in words:
        assert word.encode() not in frame, f"{word!r} travels in the clear"


@pytest.mark.asyncio
async def test_index_sync_frame_carries_no_filename(indexer):
    frame = msgpack.packb(index_sync_message(indexer.index, indexer.roots),
                          use_bin_type=True)
    # Not only the whole name — a fragment of it would be just as much of a leak.
    _assert_absent(frame, SECRET_FILE, "quixotry", SECRET_DIR, "zarfwidget",
                   "entries", "dirs")
    # And the routing fields are still readable, or nothing could be dispatched.
    msg = msgpack.unpackb(frame, raw=False)
    assert msg["type"] == MNP.INDEX_SYNC
    assert msg["group_id"] == "g-1"
    assert set(msg) == {"type", "v", "group_id", "nonce", "ct"}


@pytest.mark.asyncio
async def test_index_sync_payload_still_says_everything(indexer, gek):
    """Sealed, not lost: every field a client reads is inside."""
    msg = index_sync_message(indexer.index, indexer.roots)
    payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_SYNC, "g-1", msg)
    assert [e["name"] for e in payload["entries"]] == [SECRET_FILE]
    assert any(d.endswith(SECRET_DIR) for d in payload["dirs"])
    assert payload["roots"]
    # Moved inside deliberately (D4): there is no reason to act on a version
    # carried by a message we have not authenticated.
    assert payload["version"] == indexer.index.version
    assert "version" not in msg


@pytest.mark.asyncio
async def test_index_delta_frame_carries_no_filename(indexer, gek):
    """
    The delta is where a cleartext path would most easily survive: it used to be
    hand-built in the daemon, a third construction site for an index message.
    """
    before = GroupIndex._snapshot(
        "g-1", indexer.index.sk_node, gek, indexer.index.version - 1, {})
    delta = indexer.index.diff(before)
    assert delta.additions

    frame = msgpack.packb(index_delta_message(indexer.index, delta),
                          use_bin_type=True)
    _assert_absent(frame, SECRET_FILE, "quixotry", "additions", "deletions")

    payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_DELTA, "g-1",
                     msgpack.unpackb(frame, raw=False))
    assert [e["name"] for e in payload["additions"]] == [SECRET_FILE]
    assert payload["base_version"] == delta.base_version


def test_handshake_ack_frame_carries_no_configuration(gek):
    """
    The ack is the line that matters most, and it is an integrity gap as much as a
    confidentiality one: the node signs `handshake_transcript(...)`, which names no
    ack field, so every value below was authenticated by the DTLS channel alone.
    """
    config = {
        "is_node_admin": True,
        "video_root": SECRET_ROOT,
        "enabled_apps": ["files", "videos"],
    }
    ack = {
        "type": MNP.HANDSHAKE_ACK,
        "v": "1.0",
        "node_pk": "Tk9ERVBL",
        "proof": "cHJvb2Y=",
        "sig": "c2ln",
        **seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, "g-1", config),
    }
    frame = msgpack.packb(ack, use_bin_type=True)
    _assert_absent(frame, SECRET_ROOT, "vasculum", "video_root",
                   "enabled_apps", "is_node_admin")

    # What a client needs in order to authenticate the node is still in clear —
    # it verifies those *before* it would trust a decryption.
    msg = msgpack.unpackb(frame, raw=False)
    assert msg["node_pk"] and msg["proof"] and msg["sig"]
    assert unseal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, "g-1", msg) == config


def test_index_progress_stays_clear_and_stays_counters():
    """
    Decision D3: `index_progress` is deliberately *not* sealed — counters only,
    pushed every couple of seconds for the whole length of a scan, so sealing it
    would buy a rough library size and cost a decrypt per push.

    The field list is re-derived from the daemon's own source rather than restated
    here, so this fails the day the message grows something that names anything —
    `IndexProgress` already carries a `current_dir` the push deliberately omits,
    and adding it would be one line. That is the moment the trade above is void.
    """
    import ast
    import inspect
    import textwrap

    from meshbay_node.daemon import NodeDaemon

    source = inspect.getsource(NodeDaemon._push_index_progress)
    tree = ast.parse(textwrap.dedent(source))
    dicts = [n for n in ast.walk(tree) if isinstance(n, ast.Dict)]
    assert len(dicts) == 1, "more than one message built here — re-read this test"
    keys = {k.value for k in dicts[0].keys}
    # `kind` is one of four fixed words and `root_pos`/`queued` are integers —
    # the root names themselves stay on the loopback API.
    assert keys == {"type", "v", "group_id",
                    "scanning", "scanned_bytes", "total_bytes",
                    "files_done", "files_total", "kind", "root_pos", "queued"}, (
        f"index_progress now carries {keys} — re-read decision D3 before shipping it")
    values = {k.value: v for k, v in zip(dicts[0].keys, dicts[0].values)}
    assert ast.unparse(values["queued"]) == "len(progress.queued)", (
        "the roots waiting must go out as a count, never as their names")

    assert "seal(" not in source
    assert "D3" in source, "the reason it is not sealed must stay next to the code"